bench: lower attach concurrency

it has been observed that we get transaction serialization issues
otherwise.

evidence: https://neon-github-public-dev.s3.amazonaws.com/reports/main/10402057346/index.html#suites/c62b105f3a4f00dd6be4ad88810e0e02/7c03011d9dbd5ec4/

.github/actionlint.yml

@@ -1,6 +1,7 @@
 self-hosted-runner:
   labels:
     - arm64
+    - gen3
     - large
     - large-arm64
     - small

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

@@ -83,6 +83,7 @@ runs:
       uses: actions/checkout@v4
       with:
         submodules: true
+        fetch-depth: 1
 
     - name: Cache poetry deps
       uses: actions/cache@v4

.github/workflows/_build-and-test-locally.yml

@@ -70,6 +70,7 @@ jobs:
       - uses: actions/checkout@v4
         with:
           submodules: true
+          fetch-depth: 1
 
       - name: Set pg 14 revision for caching
         id: pg_v14_rev
@@ -207,7 +208,7 @@ jobs:
           export LD_LIBRARY_PATH
 
           #nextest does not yet support running doctests
-          ${cov_prefix} cargo test --doc $CARGO_FLAGS $CARGO_FEATURES
+          cargo test --doc $CARGO_FLAGS $CARGO_FEATURES
 
           for io_engine in std-fs tokio-epoll-uring ; do
             NEON_PAGESERVER_UNIT_TEST_VIRTUAL_FILE_IOENGINE=$io_engine ${cov_prefix} cargo nextest run $CARGO_FLAGS $CARGO_FEATURES
@@ -262,6 +263,7 @@ jobs:
       - uses: actions/checkout@v4
         with:
           submodules: true
+          fetch-depth: 1
 
       - name: Pytest regression tests
         uses: ./.github/actions/run-python-test-set

.github/workflows/actionlint.yml

@@ -44,7 +44,7 @@ jobs:
             grep -ERl $PAT .github/workflows |\
             while read -r f
             do
-              l=$(grep -nE $PAT $f | awk -F: '{print $1}' | head -1)
+              l=$(grep -nE $PAT .github/workflows/release.yml | awk -F: '{print $1}' | head -1)
               echo "::error file=$f,line=$l::Please use 'ubuntu-22.04' instead of 'ubuntu-latest'"
             done
             exit 1

.github/workflows/benchmarking.yml

@@ -96,7 +96,7 @@ jobs:
       uses: aws-actions/configure-aws-credentials@v4
       with:
         aws-region: eu-central-1
-        role-to-assume: ${{ vars.DEV_AWS_OIDC_ROLE_ARN }}
+        role-to-assume: ${{ vars.DEV_AWS_OIDC_ROLE_ARN }}  
         role-duration-seconds: 18000 # 5 hours
 
     - name: Download Neon artifact
@@ -146,7 +146,6 @@ jobs:
         api_key: ${{ secrets.NEON_STAGING_API_KEY }}
 
     - name: Create Allure report
-      id: create-allure-report
       if: ${{ !cancelled() }}
       uses: ./.github/actions/allure-report-generate
 
@@ -155,10 +154,7 @@ jobs:
       uses: slackapi/slack-github-action@v1
       with:
         channel-id: "C033QLM5P7D" # dev-staging-stream
-        slack-message: |
-          Periodic perf testing: ${{ job.status }}
-          <${{ github.server_url }}/${{ github.repository }}/actions/runs/${{ github.run_id }}|GitHub Run>
-          <${{ steps.create-allure-report.outputs.report-url }}|Allure report>
+        slack-message: "Periodic perf testing: ${{ job.status }}\n${{ github.server_url }}/${{ github.repository }}/actions/runs/${{ github.run_id }}"
       env:
         SLACK_BOT_TOKEN: ${{ secrets.SLACK_BOT_TOKEN }}
 
@@ -180,7 +176,7 @@ jobs:
     steps:
     - uses: actions/checkout@v4
 
-
+    
     - name: Download Neon artifact
       uses: ./.github/actions/download
       with:
@@ -219,7 +215,6 @@ jobs:
         NEON_API_KEY: ${{ secrets.NEON_STAGING_API_KEY }}
 
     - name: Create Allure report
-      id: create-allure-report
       if: ${{ !cancelled() }}
       uses: ./.github/actions/allure-report-generate
 
@@ -370,7 +365,7 @@ jobs:
         aws-region: eu-central-1
         role-to-assume: ${{ vars.DEV_AWS_OIDC_ROLE_ARN }}
         role-duration-seconds: 18000 # 5 hours
-
+        
     - name: Download Neon artifact
       uses: ./.github/actions/download
       with:
@@ -465,7 +460,6 @@ jobs:
         api_key: ${{ secrets.NEON_STAGING_API_KEY }}
 
     - name: Create Allure report
-      id: create-allure-report
       if: ${{ !cancelled() }}
       uses: ./.github/actions/allure-report-generate
 
@@ -474,10 +468,7 @@ jobs:
       uses: slackapi/slack-github-action@v1
       with:
         channel-id: "C033QLM5P7D" # dev-staging-stream
-        slack-message: |
-          Periodic perf testing on ${{ matrix.platform }}: ${{ job.status }}
-          <${{ github.server_url }}/${{ github.repository }}/actions/runs/${{ github.run_id }}|GitHub Run>
-          <${{ steps.create-allure-report.outputs.report-url }}|Allure report>
+        slack-message: "Periodic perf testing ${{ matrix.platform }}: ${{ job.status }}\n${{ github.server_url }}/${{ github.repository }}/actions/runs/${{ github.run_id }}"
       env:
         SLACK_BOT_TOKEN: ${{ secrets.SLACK_BOT_TOKEN }}
 
@@ -551,7 +542,7 @@ jobs:
         esac
 
         echo "connstr=${CONNSTR}" >> $GITHUB_OUTPUT
-
+        
     - name: Configure AWS credentials # necessary on Azure runners to read/write from/to S3
       uses: aws-actions/configure-aws-credentials@v4
       with:
@@ -586,9 +577,8 @@ jobs:
         BENCHMARK_CONNSTR: ${{ steps.set-up-connstr.outputs.connstr }}
         VIP_VAP_ACCESS_TOKEN: "${{ secrets.VIP_VAP_ACCESS_TOKEN }}"
         PERF_TEST_RESULT_CONNSTR: "${{ secrets.PERF_TEST_RESULT_CONNSTR }}"
-
+    
     - name: Create Allure report
-      id: create-allure-report
       if: ${{ !cancelled() }}
       uses: ./.github/actions/allure-report-generate
 
@@ -597,10 +587,7 @@ jobs:
       uses: slackapi/slack-github-action@v1
       with:
         channel-id: "C033QLM5P7D" # dev-staging-stream
-        slack-message: |
-          Periodic perf testing on ${{ env.PLATFORM }}: ${{ job.status }}
-          <${{ github.server_url }}/${{ github.repository }}/actions/runs/${{ github.run_id }}|GitHub Run>
-          <${{ steps.create-allure-report.outputs.report-url }}|Allure report>
+        slack-message: "Periodic perf testing ${PLATFORM}: ${{ job.status }}\n${{ github.server_url }}/${{ github.repository }}/actions/runs/${{ github.run_id }}"
       env:
         SLACK_BOT_TOKEN: ${{ secrets.SLACK_BOT_TOKEN }}
 
@@ -683,7 +670,6 @@ jobs:
         TEST_OLAP_SCALE: 10
 
     - name: Create Allure report
-      id: create-allure-report
       if: ${{ !cancelled() }}
       uses: ./.github/actions/allure-report-generate
 
@@ -692,10 +678,7 @@ jobs:
       uses: slackapi/slack-github-action@v1
       with:
         channel-id: "C033QLM5P7D" # dev-staging-stream
-        slack-message: |
-          Periodic OLAP perf testing on ${{ matrix.platform }}: ${{ job.status }}
-          <${{ github.server_url }}/${{ github.repository }}/actions/runs/${{ github.run_id }}|GitHub Run>
-          <${{ steps.create-allure-report.outputs.report-url }}|Allure report>
+        slack-message: "Periodic OLAP perf testing ${{ matrix.platform }}: ${{ job.status }}\n${{ github.server_url }}/${{ github.repository }}/actions/runs/${{ github.run_id }}"
       env:
         SLACK_BOT_TOKEN: ${{ secrets.SLACK_BOT_TOKEN }}
 
@@ -781,7 +764,6 @@ jobs:
         TEST_OLAP_SCALE: ${{ matrix.scale }}
 
     - name: Create Allure report
-      id: create-allure-report
       if: ${{ !cancelled() }}
       uses: ./.github/actions/allure-report-generate
 
@@ -790,10 +772,7 @@ jobs:
       uses: slackapi/slack-github-action@v1
       with:
         channel-id: "C033QLM5P7D" # dev-staging-stream
-        slack-message: |
-          Periodic TPC-H perf testing on ${{ matrix.platform }}: ${{ job.status }}
-          <${{ github.server_url }}/${{ github.repository }}/actions/runs/${{ github.run_id }}|GitHub Run>
-          <${{ steps.create-allure-report.outputs.report-url }}|Allure report>
+        slack-message: "Periodic TPC-H perf testing ${{ matrix.platform }}: ${{ job.status }}\n${{ github.server_url }}/${{ github.repository }}/actions/runs/${{ github.run_id }}"
       env:
         SLACK_BOT_TOKEN: ${{ secrets.SLACK_BOT_TOKEN }}
 
@@ -864,7 +843,6 @@ jobs:
         BENCHMARK_CONNSTR: ${{ steps.set-up-connstr.outputs.connstr }}
 
     - name: Create Allure report
-      id: create-allure-report
       if: ${{ !cancelled() }}
       uses: ./.github/actions/allure-report-generate
 
@@ -873,10 +851,6 @@ jobs:
       uses: slackapi/slack-github-action@v1
       with:
         channel-id: "C033QLM5P7D" # dev-staging-stream
-        slack-message: |
-          Periodic TPC-H perf testing on ${{ matrix.platform }}: ${{ job.status }}
-          <${{ github.server_url }}/${{ github.repository }}/actions/runs/${{ github.run_id }}|GitHub Run>
-          <${{ steps.create-allure-report.outputs.report-url }}|Allure report>
-
+        slack-message: "Periodic User example perf testing ${{ matrix.platform }}: ${{ job.status }}\n${{ github.server_url }}/${{ github.repository }}/actions/runs/${{ github.run_id }}"
       env:
         SLACK_BOT_TOKEN: ${{ secrets.SLACK_BOT_TOKEN }}

.github/workflows/build-build-tools-image.yml

@@ -38,7 +38,7 @@ jobs:
       matrix:
         arch: [ x64, arm64 ]
 
-    runs-on: ${{ fromJson(format('["self-hosted", "{0}"]', matrix.arch == 'arm64' && 'large-arm64' || 'large')) }}
+    runs-on: ${{ fromJson(format('["self-hosted", "gen3", "{0}"]', matrix.arch == 'arm64' && 'large-arm64' || 'large')) }}
 
     env:
       IMAGE_TAG: ${{ inputs.image-tag }}

.github/workflows/build_and_test.yml

@@ -48,7 +48,7 @@ jobs:
 
   tag:
     needs: [ check-permissions ]
-    runs-on: [ self-hosted, small ]
+    runs-on: [ self-hosted, gen3, small ]
     container: 369495373322.dkr.ecr.eu-central-1.amazonaws.com/base:pinned
     outputs:
       build-tag: ${{steps.build-tag.outputs.tag}}
@@ -90,7 +90,7 @@ jobs:
 
   check-codestyle-python:
     needs: [ check-permissions, build-build-tools-image ]
-    runs-on: [ self-hosted, small ]
+    runs-on: [ self-hosted, gen3, small ]
     container:
       image: ${{ needs.build-build-tools-image.outputs.image }}
       credentials:
@@ -101,6 +101,9 @@ jobs:
     steps:
       - name: Checkout
         uses: actions/checkout@v4
+        with:
+          submodules: false
+          fetch-depth: 1
 
       - name: Cache poetry deps
         uses: actions/cache@v4
@@ -139,6 +142,7 @@ jobs:
         uses: actions/checkout@v4
         with:
           submodules: true
+          fetch-depth: 1
 
 #      Disabled for now
 #      - name: Restore cargo deps cache
@@ -200,7 +204,7 @@ jobs:
       matrix:
         arch: [ x64 ]
         # Do not build or run tests in debug for release branches
-        build-type: ${{ fromJson((startsWith(github.ref_name, 'release') && github.event_name == 'push') && '["release"]' || '["debug", "release"]') }}
+        build-type: ${{ fromJson((startsWith(github.ref_name, 'release' && github.event_name == 'push')) && '["release"]' || '["debug", "release"]') }}
         include:
           - build-type: release
             arch: arm64
@@ -220,7 +224,7 @@ jobs:
     outputs:
       json: ${{ steps.get-benchmark-durations.outputs.json }}
     needs: [ check-permissions, build-build-tools-image ]
-    runs-on: [ self-hosted, small ]
+    runs-on: [ self-hosted, gen3, small ]
     container:
       image: ${{ needs.build-build-tools-image.outputs.image }}
       credentials:
@@ -253,7 +257,7 @@ jobs:
   benchmarks:
     if: github.ref_name == 'main' || contains(github.event.pull_request.labels.*.name, 'run-benchmarks')
     needs: [ check-permissions, build-and-test-locally, build-build-tools-image, get-benchmarks-durations ]
-    runs-on: [ self-hosted, small ]
+    runs-on: [ self-hosted, gen3, small ]
     container:
       image: ${{ needs.build-build-tools-image.outputs.image }}
       credentials:
@@ -298,8 +302,9 @@ jobs:
       with:
         channel-id: C060CNA47S9 # on-call-staging-storage-stream
         slack-message: |
-          Benchmarks failed on main <${{ github.event.head_commit.url }}|${{ github.sha }}>
-          <${{ needs.create-test-report.outputs.report-url }}|Allure report>
+          Benchmarks failed on main: ${{ github.event.head_commit.url }}
+
+          Allure report: ${{ needs.create-test-report.outputs.report-url }}
       env:
         SLACK_BOT_TOKEN: ${{ secrets.SLACK_BOT_TOKEN }}
 
@@ -309,7 +314,7 @@ jobs:
     outputs:
       report-url: ${{ steps.create-allure-report.outputs.report-url }}
 
-    runs-on: [ self-hosted, small ]
+    runs-on: [ self-hosted, gen3, small ]
     container:
       image: ${{ needs.build-build-tools-image.outputs.image }}
       credentials:
@@ -356,7 +361,7 @@ jobs:
 
   coverage-report:
     needs: [ check-permissions, build-build-tools-image, build-and-test-locally ]
-    runs-on: [ self-hosted, small ]
+    runs-on: [ self-hosted, gen3, small ]
     container:
       image: ${{ needs.build-build-tools-image.outputs.image }}
       credentials:
@@ -470,7 +475,7 @@ jobs:
       matrix:
         arch: [ x64, arm64 ]
 
-    runs-on: ${{ fromJson(format('["self-hosted", "{0}"]', matrix.arch == 'arm64' && 'large-arm64' || 'large')) }}
+    runs-on: ${{ fromJson(format('["self-hosted", "gen3", "{0}"]', matrix.arch == 'arm64' && 'large-arm64' || 'large')) }}
 
     steps:
       - name: Checkout
@@ -498,10 +503,7 @@ jobs:
       - uses: docker/build-push-action@v6
         with:
           context: .
-          # ARM-specific flags are recommended for Graviton ≥ 2, these flags are also supported by Ampere Altra (Azure)
-          # https://github.com/aws/aws-graviton-getting-started/blob/57dc813626d0266f1cc12ef83474745bb1f31fb4/rust.md
           build-args: |
-            ADDITIONAL_RUSTFLAGS=${{ matrix.arch == 'arm64' && '-Ctarget-feature=+lse -Ctarget-cpu=neoverse-n1' || '' }}
             GIT_VERSION=${{ github.event.pull_request.head.sha || github.sha }}
             BUILD_TAG=${{ needs.tag.outputs.build-tag }}
             TAG=${{ needs.build-build-tools-image.outputs.image-tag }}
@@ -549,7 +551,7 @@ jobs:
         version: [ v14, v15, v16 ]
         arch: [ x64, arm64 ]
 
-    runs-on: ${{ fromJson(format('["self-hosted", "{0}"]', matrix.arch == 'arm64' && 'large-arm64' || 'large')) }}
+    runs-on: ${{ fromJson(format('["self-hosted", "gen3", "{0}"]', matrix.arch == 'arm64' && 'large-arm64' || 'large')) }}
 
     steps:
       - name: Checkout
@@ -694,7 +696,7 @@ jobs:
 
   vm-compute-node-image:
     needs: [ check-permissions, tag, compute-node-image ]
-    runs-on: [ self-hosted, large ]
+    runs-on: [ self-hosted, gen3, large ]
     strategy:
       fail-fast: false
       matrix:
@@ -743,7 +745,7 @@ jobs:
       matrix:
         arch: [ x64, arm64 ]
 
-    runs-on: ${{ fromJson(format('["self-hosted", "{0}"]', matrix.arch == 'arm64' && 'small-arm64' || 'small')) }}
+    runs-on: ${{ fromJson(format('["self-hosted", "gen3", "{0}"]', matrix.arch == 'arm64' && 'small-arm64' || 'small')) }}
 
     steps:
       - name: Checkout
@@ -958,7 +960,7 @@ jobs:
     needs: [ check-permissions, promote-images, tag, build-and-test-locally, trigger-custom-extensions-build-and-wait ]
     if: github.ref_name == 'main' || github.ref_name == 'release'|| github.ref_name == 'release-proxy'
 
-    runs-on: [ self-hosted, small ]
+    runs-on: [ self-hosted, gen3, small ]
     container: 369495373322.dkr.ecr.eu-central-1.amazonaws.com/ansible:latest
     steps:
       - name: Fix git ownership
@@ -978,6 +980,7 @@ jobs:
       - name: Checkout
         uses: actions/checkout@v4
         with:
+          submodules: false
           fetch-depth: 0
 
       - name: Trigger deploy workflow
@@ -1058,7 +1061,7 @@ jobs:
     needs: [ check-permissions, promote-images, tag, build-and-test-locally ]
     if: github.ref_name == 'release'
 
-    runs-on: [ self-hosted, small ]
+    runs-on: [ self-hosted, gen3, small ]
     container:
       image: 369495373322.dkr.ecr.eu-central-1.amazonaws.com/base:pinned
       options: --init
@@ -1114,12 +1117,10 @@ jobs:
     # Format `needs` differently to make the list more readable.
     # Usually we do `needs: [...]`
     needs:
-      - build-and-test-locally
       - check-codestyle-python
       - check-codestyle-rust
-      - promote-images
+      - build-and-test-locally
       - test-images
-      - trigger-custom-extensions-build-and-wait
     runs-on: ubuntu-22.04
     steps:
       # The list of possible results:

.github/workflows/label-for-external-users.yml

@@ -15,34 +15,15 @@ env:
   LABEL: external
 
 jobs:
-  check-user:
-    runs-on: ubuntu-22.04
-
-    outputs:
-      is-member: ${{ steps.check-user.outputs.is-member }}
-
-    steps:
-    - name: Check whether `${{ github.actor }}` is a member of `${{ github.repository_owner }}`
-      id: check-user
-      env:
-        GH_TOKEN: ${{ secrets.GITHUB_TOKEN }}
-      run: |
-        if gh api -H "Accept: application/vnd.github+json" -H "X-GitHub-Api-Version: 2022-11-28" "/orgs/${GITHUB_REPOSITORY_OWNER}/members/${GITHUB_ACTOR}"; then
-          is_member=true
-        else
-          is_member=false
-        fi
-
-        echo "is-member=${is_member}" | tee -a ${GITHUB_OUTPUT}
-
   add-label:
-    if: needs.check-user.outputs.is-member == 'false'
-    needs: [ check-user ]
+    # This workflow uses `author_association` for PRs and issues to determine if the user is an external user.
+    # Possible values for `author_association`: https://docs.github.com/en/graphql/reference/enums#commentauthorassociation
+    if: ${{ !contains(fromJSON('["OWNER", "MEMBER", "COLLABORATOR"]'), github.event[github.event_name == 'pull_request' && 'pull_request' || 'issue'].author_association) }}
 
     runs-on: ubuntu-22.04
     permissions:
-      pull-requests: write # for `gh pr edit`
-      issues: write        # for `gh issue edit`
+      pull-requests: write
+      issues: write
 
     steps:
     - name: Label new ${{ github.event_name }}

.github/workflows/neon_extra_builds.yml

@@ -56,6 +56,7 @@ jobs:
         uses: actions/checkout@v4
         with:
           submodules: true
+          fetch-depth: 1
 
       - name: Install macOS postgres dependencies
         run: brew install flex bison openssl protobuf icu4c pkg-config
@@ -157,6 +158,7 @@ jobs:
         uses: actions/checkout@v4
         with:
           submodules: true
+          fetch-depth: 1
 
       # Some of our rust modules use FFI and need those to be checked
       - name: Get postgres headers

.github/workflows/periodic_pagebench.yml

@@ -27,7 +27,7 @@ concurrency:
 
 jobs:
   trigger_bench_on_ec2_machine_in_eu_central_1:
-    runs-on: [ self-hosted, small ]
+    runs-on: [ self-hosted, gen3, small ]
     container:
       image: neondatabase/build-tools:pinned
       credentials:

Cargo.lock

@@ -484,7 +484,7 @@ dependencies = [
  "http 0.2.9",
  "http 1.1.0",
  "once_cell",
- "p256 0.11.1",
+ "p256",
  "percent-encoding",
  "ring 0.17.6",
  "sha2",
@@ -848,12 +848,6 @@ version = "0.1.1"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "349a06037c7bf932dd7e7d1f653678b2038b9ad46a74102f1fc7bd7872678cce"
 
-[[package]]
-name = "base16ct"
-version = "0.2.0"
-source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "4c7f02d4ea65f2c1853089ffd8d2787bdbc63de2f0d29dedbcf8ccdfa0ccd4cf"
-
 [[package]]
 name = "base64"
 version = "0.13.1"
@@ -977,9 +971,9 @@ checksum = "a3e2c3daef883ecc1b5d58c15adae93470a91d425f3532ba1695849656af3fc1"
 
 [[package]]
 name = "bytemuck"
-version = "1.16.3"
+version = "1.16.0"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "102087e286b4677862ea56cf8fc58bb2cdfa8725c40ffb80fe3a008eb7f2fc83"
+checksum = "78834c15cb5d5efe3452d58b1e8ba890dd62d21907f867f383358198e56ebca5"
 
 [[package]]
 name = "byteorder"
@@ -1532,10 +1526,8 @@ version = "0.5.5"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "0dc92fb57ca44df6db8059111ab3af99a63d5d0f8375d9972e319a379c6bab76"
 dependencies = [
- "generic-array",
  "rand_core 0.6.4",
  "subtle",
- "zeroize",
 ]
 
 [[package]]
@@ -1629,7 +1621,6 @@ source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "fffa369a668c8af7dbf8b5e56c9f744fbd399949ed171606040001947de40b1c"
 dependencies = [
  "const-oid",
- "pem-rfc7468",
  "zeroize",
 ]
 
@@ -1729,7 +1720,6 @@ source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "9ed9a281f7bc9b7576e61468ba615a66a5c8cfdff42420a70aa82701a3b1e292"
 dependencies = [
  "block-buffer",
- "const-oid",
  "crypto-common",
  "subtle",
 ]
@@ -1781,25 +1771,11 @@ source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "413301934810f597c1d19ca71c8710e99a3f1ba28a0d2ebc01551a2daeea3c5c"
 dependencies = [
  "der 0.6.1",
- "elliptic-curve 0.12.3",
- "rfc6979 0.3.1",
+ "elliptic-curve",
+ "rfc6979",
  "signature 1.6.4",
 ]
 
-[[package]]
-name = "ecdsa"
-version = "0.16.9"
-source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "ee27f32b5c5292967d2d4a9d7f1e0b0aed2c15daded5a60300e4abb9d8020bca"
-dependencies = [
- "der 0.7.8",
- "digest",
- "elliptic-curve 0.13.8",
- "rfc6979 0.4.0",
- "signature 2.2.0",
- "spki 0.7.3",
-]
-
 [[package]]
 name = "either"
 version = "1.8.1"
@@ -1812,36 +1788,16 @@ version = "0.12.3"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "e7bb888ab5300a19b8e5bceef25ac745ad065f3c9f7efc6de1b91958110891d3"
 dependencies = [
- "base16ct 0.1.1",
+ "base16ct",
  "crypto-bigint 0.4.9",
  "der 0.6.1",
  "digest",
- "ff 0.12.1",
+ "ff",
  "generic-array",
- "group 0.12.1",
- "pkcs8 0.9.0",
+ "group",
+ "pkcs8",
  "rand_core 0.6.4",
- "sec1 0.3.0",
- "subtle",
- "zeroize",
-]
-
-[[package]]
-name = "elliptic-curve"
-version = "0.13.8"
-source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "b5e6043086bf7973472e0c7dff2142ea0b680d30e18d9cc40f267efbf222bd47"
-dependencies = [
- "base16ct 0.2.0",
- "crypto-bigint 0.5.5",
- "digest",
- "ff 0.13.0",
- "generic-array",
- "group 0.13.0",
- "pem-rfc7468",
- "pkcs8 0.10.2",
- "rand_core 0.6.4",
- "sec1 0.7.3",
+ "sec1",
  "subtle",
  "zeroize",
 ]
@@ -1995,16 +1951,6 @@ dependencies = [
  "subtle",
 ]
 
-[[package]]
-name = "ff"
-version = "0.13.0"
-source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "ded41244b729663b1e574f1b4fb731469f69f79c17667b5d776b16cda0479449"
-dependencies = [
- "rand_core 0.6.4",
- "subtle",
-]
-
 [[package]]
 name = "filetime"
 version = "0.2.22"
@@ -2202,7 +2148,6 @@ checksum = "85649ca51fd72272d7821adaf274ad91c288277713d9c18820d8499a7ff69e9a"
 dependencies = [
  "typenum",
  "version_check",
- "zeroize",
 ]
 
 [[package]]
@@ -2269,18 +2214,7 @@ version = "0.12.1"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "5dfbfb3a6cfbd390d5c9564ab283a0349b9b9fcd46a706c1eb10e0db70bfbac7"
 dependencies = [
- "ff 0.12.1",
- "rand_core 0.6.4",
- "subtle",
-]
-
-[[package]]
-name = "group"
-version = "0.13.0"
-source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "f0f9ef7462f7c099f518d754361858f86d8a07af53ba9af0fe635bbccb151a63"
-dependencies = [
- "ff 0.13.0",
+ "ff",
  "rand_core 0.6.4",
  "subtle",
 ]
@@ -2842,42 +2776,6 @@ dependencies = [
  "libc",
 ]
 
-[[package]]
-name = "jose-b64"
-version = "0.1.2"
-source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "bec69375368709666b21c76965ce67549f2d2db7605f1f8707d17c9656801b56"
-dependencies = [
- "base64ct",
- "serde",
- "subtle",
- "zeroize",
-]
-
-[[package]]
-name = "jose-jwa"
-version = "0.1.2"
-source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "9ab78e053fe886a351d67cf0d194c000f9d0dcb92906eb34d853d7e758a4b3a7"
-dependencies = [
- "serde",
-]
-
-[[package]]
-name = "jose-jwk"
-version = "0.1.2"
-source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "280fa263807fe0782ecb6f2baadc28dffc04e00558a58e33bfdb801d11fd58e7"
-dependencies = [
- "jose-b64",
- "jose-jwa",
- "p256 0.13.2",
- "p384",
- "rsa",
- "serde",
- "zeroize",
-]
-
 [[package]]
 name = "js-sys"
 version = "0.3.69"
@@ -2937,9 +2835,6 @@ name = "lazy_static"
 version = "1.4.0"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "e2abad23fbc42b3700f2f279844dc832adb2b2eb069b2df918f455c4e18cc646"
-dependencies = [
- "spin 0.5.2",
-]
 
 [[package]]
 name = "lazycell"
@@ -3309,23 +3204,6 @@ dependencies = [
  "num-traits",
 ]
 
-[[package]]
-name = "num-bigint-dig"
-version = "0.8.4"
-source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "dc84195820f291c7697304f3cbdadd1cb7199c0efc917ff5eafd71225c136151"
-dependencies = [
- "byteorder",
- "lazy_static",
- "libm",
- "num-integer",
- "num-iter",
- "num-traits",
- "rand 0.8.5",
- "smallvec",
- "zeroize",
-]
-
 [[package]]
 name = "num-complex"
 version = "0.4.4"
@@ -3603,33 +3481,11 @@ version = "0.11.1"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "51f44edd08f51e2ade572f141051021c5af22677e42b7dd28a88155151c33594"
 dependencies = [
- "ecdsa 0.14.8",
- "elliptic-curve 0.12.3",
+ "ecdsa",
+ "elliptic-curve",
  "sha2",
 ]
 
-[[package]]
-name = "p256"
-version = "0.13.2"
-source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "c9863ad85fa8f4460f9c48cb909d38a0d689dba1f6f6988a5e3e0d31071bcd4b"
-dependencies = [
- "ecdsa 0.16.9",
- "elliptic-curve 0.13.8",
- "primeorder",
- "sha2",
-]
-
-[[package]]
-name = "p384"
-version = "0.13.0"
-source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "70786f51bcc69f6a4c0360e063a4cac5419ef7c5cd5b3c99ad70f3be5ba79209"
-dependencies = [
- "elliptic-curve 0.13.8",
- "primeorder",
-]
-
 [[package]]
 name = "pagebench"
 version = "0.1.0"
@@ -3991,15 +3847,6 @@ dependencies = [
  "serde",
 ]
 
-[[package]]
-name = "pem-rfc7468"
-version = "0.7.0"
-source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "88b39c9bfcfc231068454382784bb460aae594343fb030d46e9f50a645418412"
-dependencies = [
- "base64ct",
-]
-
 [[package]]
 name = "percent-encoding"
 version = "2.2.0"
@@ -4066,17 +3913,6 @@ version = "0.1.0"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "8b870d8c151b6f2fb93e84a13146138f05d02ed11c7e7c54f8826aaaf7c9f184"
 
-[[package]]
-name = "pkcs1"
-version = "0.7.5"
-source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "c8ffb9f10fa047879315e6625af03c164b16962a5368d724ed16323b68ace47f"
-dependencies = [
- "der 0.7.8",
- "pkcs8 0.10.2",
- "spki 0.7.3",
-]
-
 [[package]]
 name = "pkcs8"
 version = "0.9.0"
@@ -4087,16 +3923,6 @@ dependencies = [
  "spki 0.6.0",
 ]
 
-[[package]]
-name = "pkcs8"
-version = "0.10.2"
-source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "f950b2377845cebe5cf8b5165cb3cc1a5e0fa5cfa3e1f7f55707d8fd82e0a7b7"
-dependencies = [
- "der 0.7.8",
- "spki 0.7.3",
-]
-
 [[package]]
 name = "pkg-config"
 version = "0.3.27"
@@ -4290,15 +4116,6 @@ dependencies = [
  "syn 2.0.52",
 ]
 
-[[package]]
-name = "primeorder"
-version = "0.13.6"
-source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "353e1ca18966c16d9deb1c69278edbc5f194139612772bd9537af60ac231e1e6"
-dependencies = [
- "elliptic-curve 0.13.8",
-]
-
 [[package]]
 name = "proc-macro-hack"
 version = "0.5.20+deprecated"
@@ -4416,7 +4233,6 @@ version = "0.1.0"
 dependencies = [
  "ahash",
  "anyhow",
- "arc-swap",
  "async-compression",
  "async-trait",
  "atomic-take",
@@ -4434,7 +4250,6 @@ dependencies = [
  "consumption_metrics",
  "crossbeam-deque",
  "dashmap",
- "ecdsa 0.16.9",
  "env_logger",
  "fallible-iterator",
  "framed-websockets",
@@ -4455,15 +4270,12 @@ dependencies = [
  "indexmap 2.0.1",
  "ipnet",
  "itertools 0.10.5",
- "jose-jwa",
- "jose-jwk",
  "lasso",
  "md5",
  "measured",
  "metrics",
  "once_cell",
  "opentelemetry",
- "p256 0.13.2",
  "parking_lot 0.12.1",
  "parquet",
  "parquet_derive",
@@ -4484,7 +4296,6 @@ dependencies = [
  "reqwest-retry",
  "reqwest-tracing",
  "routerify",
- "rsa",
  "rstest",
  "rustc-hash",
  "rustls 0.22.4",
@@ -4494,7 +4305,6 @@ dependencies = [
  "serde",
  "serde_json",
  "sha2",
- "signature 2.2.0",
  "smallvec",
  "smol_str",
  "socket2 0.5.5",
@@ -4997,16 +4807,6 @@ dependencies = [
  "zeroize",
 ]
 
-[[package]]
-name = "rfc6979"
-version = "0.4.0"
-source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "f8dd2a808d456c4a54e300a23e9f5a67e122c3024119acbfd73e3bf664491cb2"
-dependencies = [
- "hmac",
- "subtle",
-]
-
 [[package]]
 name = "ring"
 version = "0.16.20"
@@ -5067,26 +4867,6 @@ dependencies = [
  "archery",
 ]
 
-[[package]]
-name = "rsa"
-version = "0.9.6"
-source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "5d0e5124fcb30e76a7e79bfee683a2746db83784b86289f6251b54b7950a0dfc"
-dependencies = [
- "const-oid",
- "digest",
- "num-bigint-dig",
- "num-integer",
- "num-traits",
- "pkcs1",
- "pkcs8 0.10.2",
- "rand_core 0.6.4",
- "signature 2.2.0",
- "spki 0.7.3",
- "subtle",
- "zeroize",
-]
-
 [[package]]
 name = "rstest"
 version = "0.18.2"
@@ -5415,24 +5195,10 @@ version = "0.3.0"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "3be24c1842290c45df0a7bf069e0c268a747ad05a192f2fd7dcfdbc1cba40928"
 dependencies = [
- "base16ct 0.1.1",
+ "base16ct",
  "der 0.6.1",
  "generic-array",
- "pkcs8 0.9.0",
- "subtle",
- "zeroize",
-]
-
-[[package]]
-name = "sec1"
-version = "0.7.3"
-source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "d3e97a565f76233a6003f9f5c54be1d9c5bdfa3eccfb189469f11ec4901c47dc"
-dependencies = [
- "base16ct 0.2.0",
- "der 0.7.8",
- "generic-array",
- "pkcs8 0.10.2",
+ "pkcs8",
  "subtle",
  "zeroize",
 ]
@@ -5779,7 +5545,6 @@ version = "2.2.0"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "77549399552de45a898a580c1b41d445bf730df867cc44e6c0233bbc4b8329de"
 dependencies = [
- "digest",
  "rand_core 0.6.4",
 ]
 
@@ -7614,17 +7379,13 @@ dependencies = [
  "clap",
  "clap_builder",
  "crossbeam-utils",
- "crypto-bigint 0.5.5",
- "der 0.7.8",
  "deranged",
- "digest",
  "either",
  "fail",
  "futures-channel",
  "futures-executor",
  "futures-io",
  "futures-util",
- "generic-array",
  "getrandom 0.2.11",
  "hashbrown 0.14.5",
  "hex",
@@ -7632,7 +7393,6 @@ dependencies = [
  "hyper 0.14.26",
  "indexmap 1.9.3",
  "itertools 0.10.5",
- "lazy_static",
  "libc",
  "log",
  "memchr",
@@ -7656,9 +7416,7 @@ dependencies = [
  "serde",
  "serde_json",
  "sha2",
- "signature 2.2.0",
  "smallvec",
- "spki 0.7.3",
  "subtle",
  "syn 1.0.109",
  "syn 2.0.52",
@@ -7769,7 +7527,6 @@ version = "1.7.0"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "525b4ec142c6b68a2d10f01f7bbf6755599ca3f81ea53b8431b7dd348f5fdb2d"
 dependencies = [
- "serde",
  "zeroize_derive",
 ]
 

Dockerfile

@@ -35,9 +35,8 @@ COPY --from=pg-build /home/nonroot/pg_install/v16/include/postgresql/server pg_i
 COPY --from=pg-build /home/nonroot/pg_install/v16/lib                       pg_install/v16/lib
 COPY --chown=nonroot . .
 
-ARG ADDITIONAL_RUSTFLAGS
 RUN set -e \
-    && PQ_LIB_DIR=$(pwd)/pg_install/v16/lib RUSTFLAGS="-Clinker=clang -Clink-arg=-fuse-ld=mold -Clink-arg=-Wl,--no-rosegment ${ADDITIONAL_RUSTFLAGS}" cargo build \
+    && PQ_LIB_DIR=$(pwd)/pg_install/v16/lib RUSTFLAGS="-Clinker=clang -Clink-arg=-fuse-ld=mold -Clink-arg=-Wl,--no-rosegment" cargo build \
       --bin pg_sni_router  \
       --bin pageserver  \
       --bin pagectl  \

control_plane/storcon_cli/src/main.rs

@@ -622,7 +622,6 @@ async fn main() -> anyhow::Result<()> {
                                 threshold: threshold.into(),
                             },
                         )),
-                        heatmap_period: Some("300s".to_string()),
                         ..Default::default()
                     },
                 })

deny.toml

@@ -22,10 +22,7 @@ feature-depth = 1
 [advisories]
 db-urls = ["https://github.com/rustsec/advisory-db"]
 yanked = "warn"
-
-[[advisories.ignore]]
-id = "RUSTSEC-2023-0071"
-reason = "the marvin attack only affects private key decryption, not public key signature verification"
+ignore = []
 
 # This section is considered when running `cargo deny check licenses`
 # More documentation for the licenses section can be found here:

libs/postgres_ffi/src/lib.rs

@@ -143,8 +143,8 @@ pub use v14::xlog_utils::XLogFileName;
 
 pub use v14::bindings::DBState_DB_SHUTDOWNED;
 
-pub fn bkpimage_is_compressed(bimg_info: u8, version: u32) -> bool {
-    dispatch_pgversion!(version, pgv::bindings::bkpimg_is_compressed(bimg_info))
+pub fn bkpimage_is_compressed(bimg_info: u8, version: u32) -> anyhow::Result<bool> {
+    dispatch_pgversion!(version, Ok(pgv::bindings::bkpimg_is_compressed(bimg_info)))
 }
 
 pub fn generate_wal_segment(

pageserver/src/disk_usage_eviction_task.rs

@@ -64,7 +64,7 @@ use crate::{
         mgr::TenantManager,
         remote_timeline_client::LayerFileMetadata,
         secondary::SecondaryTenant,
-        storage_layer::{AsLayerDesc, EvictionError, Layer, LayerName, LayerVisibilityHint},
+        storage_layer::{AsLayerDesc, EvictionError, Layer, LayerName},
     },
     CancellableTask, DiskUsageEvictionTask,
 };
@@ -114,7 +114,7 @@ fn default_highest_layer_count_loses_first() -> bool {
 }
 
 impl EvictionOrder {
-    fn sort(&self, candidates: &mut [(EvictionPartition, EvictionCandidate)]) {
+    fn sort(&self, candidates: &mut [(MinResidentSizePartition, EvictionCandidate)]) {
         use EvictionOrder::*;
 
         match self {
@@ -644,7 +644,6 @@ pub(crate) struct EvictionCandidate {
     pub(crate) layer: EvictionLayer,
     pub(crate) last_activity_ts: SystemTime,
     pub(crate) relative_last_activity: finite_f32::FiniteF32,
-    pub(crate) visibility: LayerVisibilityHint,
 }
 
 impl std::fmt::Display for EvictionLayer {
@@ -686,22 +685,14 @@ impl std::fmt::Debug for EvictionCandidate {
 }
 
 #[derive(Debug, Clone, Copy, PartialEq, Eq, PartialOrd, Ord)]
-enum EvictionPartition {
-    // A layer that is un-wanted by the tenant: evict all these first, before considering
-    // any other layers
-    EvictNow,
-
-    // Above the minimum size threshold: this layer is a candidate for eviction.
+enum MinResidentSizePartition {
     Above,
-
-    // Below the minimum size threshold: this layer should only be evicted if all the
-    // tenants' layers above the minimum size threshold have already been considered.
     Below,
 }
 
 enum EvictionCandidates {
     Cancelled,
-    Finished(Vec<(EvictionPartition, EvictionCandidate)>),
+    Finished(Vec<(MinResidentSizePartition, EvictionCandidate)>),
 }
 
 /// Gather the eviction candidates.
@@ -899,10 +890,8 @@ async fn collect_eviction_candidates(
             max_layer_size
         };
 
-        // Sort layers most-recently-used first, then calculate [`EvictionPartition`] for each layer,
-        // where the inputs are:
-        //  - whether the layer is visible
-        //  - whether the layer is above/below the min_resident_size cutline
+        // Sort layers most-recently-used first, then partition by
+        // cumsum above/below min_resident_size.
         tenant_candidates
             .sort_unstable_by_key(|layer_info| std::cmp::Reverse(layer_info.last_activity_ts));
         let mut cumsum: i128 = 0;
@@ -919,23 +908,12 @@ async fn collect_eviction_candidates(
                     candidate.relative_last_activity =
                         eviction_order.relative_last_activity(total, i);
 
-                    let partition = match candidate.visibility {
-                        LayerVisibilityHint::Covered => {
-                            // Covered layers are evicted first
-                            EvictionPartition::EvictNow
-                        }
-                        LayerVisibilityHint::Visible => {
-                            cumsum += i128::from(candidate.layer.get_file_size());
-
-                            if cumsum > min_resident_size as i128 {
-                                EvictionPartition::Above
-                            } else {
-                                // The most recent layers below the min_resident_size threshold
-                                // are the last to be evicted.
-                                EvictionPartition::Below
-                            }
-                        }
+                    let partition = if cumsum > min_resident_size as i128 {
+                        MinResidentSizePartition::Above
+                    } else {
+                        MinResidentSizePartition::Below
                     };
+                    cumsum += i128::from(candidate.layer.get_file_size());
 
                     (partition, candidate)
                 });
@@ -1003,7 +981,7 @@ async fn collect_eviction_candidates(
                         // Secondary locations' layers are always considered above the min resident size,
                         // i.e. secondary locations are permitted to be trimmed to zero layers if all
                         // the layers have sufficiently old access times.
-                        EvictionPartition::Above,
+                        MinResidentSizePartition::Above,
                         candidate,
                     )
                 });
@@ -1031,9 +1009,7 @@ async fn collect_eviction_candidates(
         }
     }
 
-    debug_assert!(EvictionPartition::Above < EvictionPartition::Below,
-        "as explained in the function's doc comment, layers that aren't in the tenant's min_resident_size are evicted first");
-    debug_assert!(EvictionPartition::EvictNow < EvictionPartition::Above,
+    debug_assert!(MinResidentSizePartition::Above < MinResidentSizePartition::Below,
         "as explained in the function's doc comment, layers that aren't in the tenant's min_resident_size are evicted first");
 
     eviction_order.sort(&mut candidates);
@@ -1046,7 +1022,7 @@ async fn collect_eviction_candidates(
 ///
 /// Returns the amount of candidates selected, with the planned usage.
 fn select_victims<U: Usage>(
-    candidates: &[(EvictionPartition, EvictionCandidate)],
+    candidates: &[(MinResidentSizePartition, EvictionCandidate)],
     usage_pre: U,
 ) -> VictimSelection<U> {
     let mut usage_when_switched = None;
@@ -1058,7 +1034,7 @@ fn select_victims<U: Usage>(
             break;
         }
 
-        if partition == &EvictionPartition::Below && usage_when_switched.is_none() {
+        if partition == &MinResidentSizePartition::Below && usage_when_switched.is_none() {
             usage_when_switched = Some((usage_planned, i));
         }
 

pageserver/src/http/routes.rs

@@ -178,8 +178,10 @@ fn check_permission(request: &Request<Body>, tenant_id: Option<TenantId>) -> Res
 impl From<PageReconstructError> for ApiError {
     fn from(pre: PageReconstructError) -> ApiError {
         match pre {
-            PageReconstructError::Other(other) => ApiError::InternalServerError(other),
-            PageReconstructError::MissingKey(e) => ApiError::InternalServerError(e.into()),
+            PageReconstructError::Other(pre) => ApiError::InternalServerError(pre),
+            PageReconstructError::MissingKey(e) => {
+                ApiError::InternalServerError(anyhow::anyhow!("{e}"))
+            }
             PageReconstructError::Cancelled => ApiError::Cancelled,
             PageReconstructError::AncestorLsnTimeout(e) => ApiError::Timeout(format!("{e}").into()),
             PageReconstructError::WalRedo(pre) => ApiError::InternalServerError(pre),

pageserver/src/pgdatadir_mapping.rs

@@ -287,7 +287,10 @@ impl Timeline {
         // then check if the database was already initialized.
         // get_rel_exists can be called before dbdir is created.
         let buf = version.get(self, DBDIR_KEY, ctx).await?;
-        let dbdirs = DbDirectory::des(&buf)?.dbdirs;
+        let dbdirs = match DbDirectory::des(&buf).context("deserialization failure") {
+            Ok(dir) => Ok(dir.dbdirs),
+            Err(e) => Err(PageReconstructError::from(e)),
+        }?;
         if !dbdirs.contains_key(&(tag.spcnode, tag.dbnode)) {
             return Ok(false);
         }
@@ -295,8 +298,13 @@ impl Timeline {
         let key = rel_dir_to_key(tag.spcnode, tag.dbnode);
         let buf = version.get(self, key, ctx).await?;
 
-        let dir = RelDirectory::des(&buf)?;
-        Ok(dir.rels.contains(&(tag.relnode, tag.forknum)))
+        match RelDirectory::des(&buf).context("deserialization failure") {
+            Ok(dir) => {
+                let exists = dir.rels.contains(&(tag.relnode, tag.forknum));
+                Ok(exists)
+            }
+            Err(e) => Err(PageReconstructError::from(e)),
+        }
     }
 
     /// Get a list of all existing relations in given tablespace and database.
@@ -315,16 +323,20 @@ impl Timeline {
         let key = rel_dir_to_key(spcnode, dbnode);
         let buf = version.get(self, key, ctx).await?;
 
-        let dir = RelDirectory::des(&buf)?;
-        let rels: HashSet<RelTag> =
-            HashSet::from_iter(dir.rels.iter().map(|(relnode, forknum)| RelTag {
-                spcnode,
-                dbnode,
-                relnode: *relnode,
-                forknum: *forknum,
-            }));
+        match RelDirectory::des(&buf).context("deserialization failure") {
+            Ok(dir) => {
+                let rels: HashSet<RelTag> =
+                    HashSet::from_iter(dir.rels.iter().map(|(relnode, forknum)| RelTag {
+                        spcnode,
+                        dbnode,
+                        relnode: *relnode,
+                        forknum: *forknum,
+                    }));
 
-        Ok(rels)
+                Ok(rels)
+            }
+            Err(e) => Err(PageReconstructError::from(e)),
+        }
     }
 
     /// Get the whole SLRU segment
@@ -386,8 +398,13 @@ impl Timeline {
         let key = slru_dir_to_key(kind);
         let buf = version.get(self, key, ctx).await?;
 
-        let dir = SlruSegmentDirectory::des(&buf)?;
-        Ok(dir.segments.contains(&segno))
+        match SlruSegmentDirectory::des(&buf).context("deserialization failure") {
+            Ok(dir) => {
+                let exists = dir.segments.contains(&segno);
+                Ok(exists)
+            }
+            Err(e) => Err(PageReconstructError::from(e)),
+        }
     }
 
     /// Locate LSN, such that all transactions that committed before
@@ -603,7 +620,10 @@ impl Timeline {
         let key = slru_dir_to_key(kind);
 
         let buf = version.get(self, key, ctx).await?;
-        Ok(SlruSegmentDirectory::des(&buf)?.segments)
+        match SlruSegmentDirectory::des(&buf).context("deserialization failure") {
+            Ok(dir) => Ok(dir.segments),
+            Err(e) => Err(PageReconstructError::from(e)),
+        }
     }
 
     pub(crate) async fn get_relmap_file(
@@ -627,7 +647,10 @@ impl Timeline {
         // fetch directory entry
         let buf = self.get(DBDIR_KEY, lsn, ctx).await?;
 
-        Ok(DbDirectory::des(&buf)?.dbdirs)
+        match DbDirectory::des(&buf).context("deserialization failure") {
+            Ok(dir) => Ok(dir.dbdirs),
+            Err(e) => Err(PageReconstructError::from(e)),
+        }
     }
 
     pub(crate) async fn get_twophase_file(
@@ -649,7 +672,10 @@ impl Timeline {
         // fetch directory entry
         let buf = self.get(TWOPHASEDIR_KEY, lsn, ctx).await?;
 
-        Ok(TwoPhaseDirectory::des(&buf)?.xids)
+        match TwoPhaseDirectory::des(&buf).context("deserialization failure") {
+            Ok(dir) => Ok(dir.xids),
+            Err(e) => Err(PageReconstructError::from(e)),
+        }
     }
 
     pub(crate) async fn get_control_file(
@@ -674,7 +700,10 @@ impl Timeline {
         ctx: &RequestContext,
     ) -> Result<HashMap<String, Bytes>, PageReconstructError> {
         match self.get(AUX_FILES_KEY, lsn, ctx).await {
-            Ok(buf) => Ok(AuxFilesDirectory::des(&buf)?.files),
+            Ok(buf) => match AuxFilesDirectory::des(&buf).context("deserialization failure") {
+                Ok(dir) => Ok(dir.files),
+                Err(e) => Err(PageReconstructError::from(e)),
+            },
             Err(e) => {
                 // This is expected: historical databases do not have the key.
                 debug!("Failed to get info about AUX files: {}", e);
@@ -690,14 +719,13 @@ impl Timeline {
     ) -> Result<HashMap<String, Bytes>, PageReconstructError> {
         let kv = self
             .scan(KeySpace::single(Key::metadata_aux_key_range()), lsn, ctx)
-            .await?;
+            .await
+            .context("scan")?;
         let mut result = HashMap::new();
         let mut sz = 0;
         for (_, v) in kv {
-            let v = v?;
-            let v = aux_file::decode_file_value_bytes(&v)
-                .context("value decode")
-                .map_err(PageReconstructError::Other)?;
+            let v = v.context("get value")?;
+            let v = aux_file::decode_file_value_bytes(&v).context("value decode")?;
             for (fname, content) in v {
                 sz += fname.len();
                 sz += content.len();
@@ -765,10 +793,11 @@ impl Timeline {
     ) -> Result<HashMap<RepOriginId, Lsn>, PageReconstructError> {
         let kv = self
             .scan(KeySpace::single(repl_origin_key_range()), lsn, ctx)
-            .await?;
+            .await
+            .context("scan")?;
         let mut result = HashMap::new();
         for (k, v) in kv {
-            let v = v?;
+            let v = v.context("get value")?;
             let origin_id = k.field6 as RepOriginId;
             let origin_lsn = Lsn::des(&v).unwrap();
             if origin_lsn != Lsn::INVALID {
@@ -1704,17 +1733,12 @@ impl<'a> DatadirModification<'a> {
                     // the original code assumes all other errors are missing keys. Therefore, we keep the code path
                     // the same for now, though in theory, we should only match the `MissingKey` variant.
                     Err(
-                        e @ (PageReconstructError::Other(_)
+                        PageReconstructError::Other(_)
                         | PageReconstructError::WalRedo(_)
-                        | PageReconstructError::MissingKey(_)),
+                        | PageReconstructError::MissingKey { .. },
                     ) => {
                         // Key is missing, we must insert an image as the basis for subsequent deltas.
 
-                        if !matches!(e, PageReconstructError::MissingKey(_)) {
-                            let e = utils::error::report_compact_sources(&e);
-                            tracing::warn!("treating error as if it was a missing key: {}", e);
-                        }
-
                         let mut dir = AuxFilesDirectory {
                             files: HashMap::new(),
                         };
@@ -1869,7 +1893,7 @@ impl<'a> DatadirModification<'a> {
                     // work directly with Images, and we never need to read actual
                     // data pages. We could handle this if we had to, by calling
                     // the walredo manager, but let's keep it simple for now.
-                    Err(PageReconstructError::Other(anyhow::anyhow!(
+                    Err(PageReconstructError::from(anyhow::anyhow!(
                         "unexpected pending WAL record"
                     )))
                 };

pageserver/src/tenant.rs

@@ -4491,13 +4491,10 @@ mod tests {
 
         // This needs to traverse to the parent, and fails.
         let err = newtline.get(*TEST_KEY, Lsn(0x50), &ctx).await.unwrap_err();
-        assert!(
-            err.to_string().starts_with(&format!(
-                "bad state on timeline {}: Broken",
-                tline.timeline_id
-            )),
-            "{err}"
-        );
+        assert!(err.to_string().starts_with(&format!(
+            "Bad state on timeline {}: Broken",
+            tline.timeline_id
+        )));
 
         Ok(())
     }

pageserver/src/tenant/blob_io.rs

@@ -24,7 +24,6 @@ use tracing::warn;
 use crate::context::RequestContext;
 use crate::page_cache::PAGE_SZ;
 use crate::tenant::block_io::BlockCursor;
-use crate::virtual_file::owned_buffers_io::io_buf_ext::{FullSlice, IoBufExt};
 use crate::virtual_file::VirtualFile;
 use std::cmp::min;
 use std::io::{Error, ErrorKind};
@@ -187,11 +186,11 @@ impl<const BUFFERED: bool> BlobWriter<BUFFERED> {
     /// You need to make sure that the internal buffer is empty, otherwise
     /// data will be written in wrong order.
     #[inline(always)]
-    async fn write_all_unbuffered<Buf: IoBuf + Send>(
+    async fn write_all_unbuffered<B: BoundedBuf<Buf = Buf>, Buf: IoBuf + Send>(
         &mut self,
-        src_buf: FullSlice<Buf>,
+        src_buf: B,
         ctx: &RequestContext,
-    ) -> (FullSlice<Buf>, Result<(), Error>) {
+    ) -> (B::Buf, Result<(), Error>) {
         let (src_buf, res) = self.inner.write_all(src_buf, ctx).await;
         let nbytes = match res {
             Ok(nbytes) => nbytes,
@@ -205,9 +204,8 @@ impl<const BUFFERED: bool> BlobWriter<BUFFERED> {
     /// Flushes the internal buffer to the underlying `VirtualFile`.
     pub async fn flush_buffer(&mut self, ctx: &RequestContext) -> Result<(), Error> {
         let buf = std::mem::take(&mut self.buf);
-        let (slice, res) = self.inner.write_all(buf.slice_len(), ctx).await;
+        let (mut buf, res) = self.inner.write_all(buf, ctx).await;
         res?;
-        let mut buf = slice.into_raw_slice().into_inner();
         buf.clear();
         self.buf = buf;
         Ok(())
@@ -224,30 +222,19 @@ impl<const BUFFERED: bool> BlobWriter<BUFFERED> {
     }
 
     /// Internal, possibly buffered, write function
-    async fn write_all<Buf: IoBuf + Send>(
+    async fn write_all<B: BoundedBuf<Buf = Buf>, Buf: IoBuf + Send>(
         &mut self,
-        src_buf: FullSlice<Buf>,
+        src_buf: B,
         ctx: &RequestContext,
-    ) -> (FullSlice<Buf>, Result<(), Error>) {
-        let src_buf = src_buf.into_raw_slice();
-        let src_buf_bounds = src_buf.bounds();
-        let restore = move |src_buf_slice: Slice<_>| {
-            FullSlice::must_new(Slice::from_buf_bounds(
-                src_buf_slice.into_inner(),
-                src_buf_bounds,
-            ))
-        };
-
+    ) -> (B::Buf, Result<(), Error>) {
         if !BUFFERED {
             assert!(self.buf.is_empty());
-            return self
-                .write_all_unbuffered(FullSlice::must_new(src_buf), ctx)
-                .await;
+            return self.write_all_unbuffered(src_buf, ctx).await;
         }
         let remaining = Self::CAPACITY - self.buf.len();
         let src_buf_len = src_buf.bytes_init();
         if src_buf_len == 0 {
-            return (restore(src_buf), Ok(()));
+            return (Slice::into_inner(src_buf.slice_full()), Ok(()));
         }
         let mut src_buf = src_buf.slice(0..src_buf_len);
         // First try to copy as much as we can into the buffer
@@ -258,7 +245,7 @@ impl<const BUFFERED: bool> BlobWriter<BUFFERED> {
         // Then, if the buffer is full, flush it out
         if self.buf.len() == Self::CAPACITY {
             if let Err(e) = self.flush_buffer(ctx).await {
-                return (restore(src_buf), Err(e));
+                return (Slice::into_inner(src_buf), Err(e));
             }
         }
         // Finally, write the tail of src_buf:
@@ -271,29 +258,27 @@ impl<const BUFFERED: bool> BlobWriter<BUFFERED> {
                 let copied = self.write_into_buffer(&src_buf);
                 // We just verified above that src_buf fits into our internal buffer.
                 assert_eq!(copied, src_buf.len());
-                restore(src_buf)
+                Slice::into_inner(src_buf)
             } else {
-                let (src_buf, res) = self
-                    .write_all_unbuffered(FullSlice::must_new(src_buf), ctx)
-                    .await;
+                let (src_buf, res) = self.write_all_unbuffered(src_buf, ctx).await;
                 if let Err(e) = res {
                     return (src_buf, Err(e));
                 }
                 src_buf
             }
         } else {
-            restore(src_buf)
+            Slice::into_inner(src_buf)
         };
         (src_buf, Ok(()))
     }
 
     /// Write a blob of data. Returns the offset that it was written to,
     /// which can be used to retrieve the data later.
-    pub async fn write_blob<Buf: IoBuf + Send>(
+    pub async fn write_blob<B: BoundedBuf<Buf = Buf>, Buf: IoBuf + Send>(
         &mut self,
-        srcbuf: FullSlice<Buf>,
+        srcbuf: B,
         ctx: &RequestContext,
-    ) -> (FullSlice<Buf>, Result<u64, Error>) {
+    ) -> (B::Buf, Result<u64, Error>) {
         let (buf, res) = self
             .write_blob_maybe_compressed(srcbuf, ctx, ImageCompressionAlgorithm::Disabled)
             .await;
@@ -302,40 +287,43 @@ impl<const BUFFERED: bool> BlobWriter<BUFFERED> {
 
     /// Write a blob of data. Returns the offset that it was written to,
     /// which can be used to retrieve the data later.
-    pub(crate) async fn write_blob_maybe_compressed<Buf: IoBuf + Send>(
+    pub async fn write_blob_maybe_compressed<B: BoundedBuf<Buf = Buf>, Buf: IoBuf + Send>(
         &mut self,
-        srcbuf: FullSlice<Buf>,
+        srcbuf: B,
         ctx: &RequestContext,
         algorithm: ImageCompressionAlgorithm,
-    ) -> (FullSlice<Buf>, Result<(u64, CompressionInfo), Error>) {
+    ) -> (B::Buf, Result<(u64, CompressionInfo), Error>) {
         let offset = self.offset;
         let mut compression_info = CompressionInfo {
             written_compressed: false,
             compressed_size: None,
         };
 
-        let len = srcbuf.len();
+        let len = srcbuf.bytes_init();
 
         let mut io_buf = self.io_buf.take().expect("we always put it back below");
         io_buf.clear();
         let mut compressed_buf = None;
-        let ((io_buf_slice, hdr_res), srcbuf) = async {
+        let ((io_buf, hdr_res), srcbuf) = async {
             if len < 128 {
                 // Short blob. Write a 1-byte length header
                 io_buf.put_u8(len as u8);
-                (self.write_all(io_buf.slice_len(), ctx).await, srcbuf)
+                (
+                    self.write_all(io_buf, ctx).await,
+                    srcbuf.slice_full().into_inner(),
+                )
             } else {
                 // Write a 4-byte length header
                 if len > MAX_SUPPORTED_LEN {
                     return (
                         (
-                            io_buf.slice_len(),
+                            io_buf,
                             Err(Error::new(
                                 ErrorKind::Other,
                                 format!("blob too large ({len} bytes)"),
                             )),
                         ),
-                        srcbuf,
+                        srcbuf.slice_full().into_inner(),
                     );
                 }
                 let (high_bit_mask, len_written, srcbuf) = match algorithm {
@@ -348,7 +336,8 @@ impl<const BUFFERED: bool> BlobWriter<BUFFERED> {
                         } else {
                             async_compression::tokio::write::ZstdEncoder::new(Vec::new())
                         };
-                        encoder.write_all(&srcbuf[..]).await.unwrap();
+                        let slice = srcbuf.slice_full();
+                        encoder.write_all(&slice[..]).await.unwrap();
                         encoder.shutdown().await.unwrap();
                         let compressed = encoder.into_inner();
                         compression_info.compressed_size = Some(compressed.len());
@@ -356,29 +345,31 @@ impl<const BUFFERED: bool> BlobWriter<BUFFERED> {
                             compression_info.written_compressed = true;
                             let compressed_len = compressed.len();
                             compressed_buf = Some(compressed);
-                            (BYTE_ZSTD, compressed_len, srcbuf)
+                            (BYTE_ZSTD, compressed_len, slice.into_inner())
                         } else {
-                            (BYTE_UNCOMPRESSED, len, srcbuf)
+                            (BYTE_UNCOMPRESSED, len, slice.into_inner())
                         }
                     }
-                    ImageCompressionAlgorithm::Disabled => (BYTE_UNCOMPRESSED, len, srcbuf),
+                    ImageCompressionAlgorithm::Disabled => {
+                        (BYTE_UNCOMPRESSED, len, srcbuf.slice_full().into_inner())
+                    }
                 };
                 let mut len_buf = (len_written as u32).to_be_bytes();
                 assert_eq!(len_buf[0] & 0xf0, 0);
                 len_buf[0] |= high_bit_mask;
                 io_buf.extend_from_slice(&len_buf[..]);
-                (self.write_all(io_buf.slice_len(), ctx).await, srcbuf)
+                (self.write_all(io_buf, ctx).await, srcbuf)
             }
         }
         .await;
-        self.io_buf = Some(io_buf_slice.into_raw_slice().into_inner());
+        self.io_buf = Some(io_buf);
         match hdr_res {
             Ok(_) => (),
-            Err(e) => return (srcbuf, Err(e)),
+            Err(e) => return (Slice::into_inner(srcbuf.slice(..)), Err(e)),
         }
         let (srcbuf, res) = if let Some(compressed_buf) = compressed_buf {
-            let (_buf, res) = self.write_all(compressed_buf.slice_len(), ctx).await;
-            (srcbuf, res)
+            let (_buf, res) = self.write_all(compressed_buf, ctx).await;
+            (Slice::into_inner(srcbuf.slice(..)), res)
         } else {
             self.write_all(srcbuf, ctx).await
         };
@@ -441,21 +432,21 @@ pub(crate) mod tests {
                 let (_, res) = if compression {
                     let res = wtr
                         .write_blob_maybe_compressed(
-                            blob.clone().slice_len(),
+                            blob.clone(),
                             ctx,
                             ImageCompressionAlgorithm::Zstd { level: Some(1) },
                         )
                         .await;
                     (res.0, res.1.map(|(off, _)| off))
                 } else {
-                    wtr.write_blob(blob.clone().slice_len(), ctx).await
+                    wtr.write_blob(blob.clone(), ctx).await
                 };
                 let offs = res?;
                 offsets.push(offs);
             }
             // Write out one page worth of zeros so that we can
             // read again with read_blk
-            let (_, res) = wtr.write_blob(vec![0; PAGE_SZ].slice_len(), ctx).await;
+            let (_, res) = wtr.write_blob(vec![0; PAGE_SZ], ctx).await;
             let offs = res?;
             println!("Writing final blob at offs={offs}");
             wtr.flush_buffer(ctx).await?;

pageserver/src/tenant/ephemeral_file/page_caching.rs

@@ -4,7 +4,6 @@
 use crate::context::RequestContext;
 use crate::page_cache::{self, PAGE_SZ};
 use crate::tenant::block_io::BlockLease;
-use crate::virtual_file::owned_buffers_io::io_buf_ext::FullSlice;
 use crate::virtual_file::VirtualFile;
 
 use once_cell::sync::Lazy;
@@ -209,11 +208,21 @@ impl PreWarmingWriter {
 }
 
 impl crate::virtual_file::owned_buffers_io::write::OwnedAsyncWriter for PreWarmingWriter {
-    async fn write_all<Buf: tokio_epoll_uring::IoBuf + Send>(
+    async fn write_all<
+        B: tokio_epoll_uring::BoundedBuf<Buf = Buf>,
+        Buf: tokio_epoll_uring::IoBuf + Send,
+    >(
         &mut self,
-        buf: FullSlice<Buf>,
+        buf: B,
         ctx: &RequestContext,
-    ) -> std::io::Result<(usize, FullSlice<Buf>)> {
+    ) -> std::io::Result<(usize, B::Buf)> {
+        let buf = buf.slice(..);
+        let saved_bounds = buf.bounds(); // save for reconstructing the Slice from iobuf after the IO is done
+        let check_bounds_stuff_works = if cfg!(test) && cfg!(debug_assertions) {
+            Some(buf.to_vec())
+        } else {
+            None
+        };
         let buflen = buf.len();
         assert_eq!(
             buflen % PAGE_SZ,
@@ -222,10 +231,10 @@ impl crate::virtual_file::owned_buffers_io::write::OwnedAsyncWriter for PreWarmi
         );
 
         // Do the IO.
-        let buf = match self.file.write_all(buf, ctx).await {
-            (buf, Ok(nwritten)) => {
+        let iobuf = match self.file.write_all(buf, ctx).await {
+            (iobuf, Ok(nwritten)) => {
                 assert_eq!(nwritten, buflen);
-                buf
+                iobuf
             }
             (_, Err(e)) => {
                 return Err(std::io::Error::new(
@@ -239,6 +248,12 @@ impl crate::virtual_file::owned_buffers_io::write::OwnedAsyncWriter for PreWarmi
             }
         };
 
+        // Reconstruct the Slice (the write path consumed the Slice and returned us the underlying IoBuf)
+        let buf = tokio_epoll_uring::Slice::from_buf_bounds(iobuf, saved_bounds);
+        if let Some(check_bounds_stuff_works) = check_bounds_stuff_works {
+            assert_eq!(&check_bounds_stuff_works, &*buf);
+        }
+
         let nblocks = buflen / PAGE_SZ;
         let nblocks32 = u32::try_from(nblocks).unwrap();
 
@@ -285,6 +300,6 @@ impl crate::virtual_file::owned_buffers_io::write::OwnedAsyncWriter for PreWarmi
         }
 
         self.nwritten_blocks = self.nwritten_blocks.checked_add(nblocks32).unwrap();
-        Ok((buflen, buf))
+        Ok((buflen, buf.into_inner()))
     }
 }

pageserver/src/tenant/ephemeral_file/zero_padded_read_write/zero_padded.rs

@@ -5,8 +5,6 @@
 
 use std::mem::MaybeUninit;
 
-use crate::virtual_file::owned_buffers_io::io_buf_ext::FullSlice;
-
 /// See module-level comment.
 pub struct Buffer<const N: usize> {
     allocation: Box<[u8; N]>,
@@ -62,10 +60,10 @@ impl<const N: usize> crate::virtual_file::owned_buffers_io::write::Buffer for Bu
         self.written
     }
 
-    fn flush(self) -> FullSlice<Self> {
+    fn flush(self) -> tokio_epoll_uring::Slice<Self> {
         self.invariants();
         let written = self.written;
-        FullSlice::must_new(tokio_epoll_uring::BoundedBuf::slice(self, 0..written))
+        tokio_epoll_uring::BoundedBuf::slice(self, 0..written)
     }
 
     fn reuse_after_flush(iobuf: Self::IoBuf) -> Self {

pageserver/src/tenant/remote_timeline_client/download.rs

@@ -23,7 +23,6 @@ use crate::span::debug_assert_current_span_has_tenant_and_timeline_id;
 use crate::tenant::remote_timeline_client::{remote_layer_path, remote_timelines_path};
 use crate::tenant::storage_layer::LayerName;
 use crate::tenant::Generation;
-use crate::virtual_file::owned_buffers_io::io_buf_ext::IoBufExt;
 use crate::virtual_file::{on_fatal_io_error, MaybeFatalIo, VirtualFile};
 use crate::TEMP_FILE_SUFFIX;
 use remote_storage::{DownloadError, GenericRemoteStorage, ListingMode, RemotePath};
@@ -220,7 +219,9 @@ async fn download_object<'a>(
                             Ok(chunk) => chunk,
                             Err(e) => return Err(e),
                         };
-                        buffered.write_buffered(chunk.slice_len(), ctx).await?;
+                        buffered
+                            .write_buffered(tokio_epoll_uring::BoundedBuf::slice_full(chunk), ctx)
+                            .await?;
                     }
                     let size_tracking = buffered.flush_and_into_inner(ctx).await?;
                     Ok(size_tracking.into_inner())

pageserver/src/tenant/secondary/downloader.rs

@@ -22,7 +22,7 @@ use crate::{
             FAILED_REMOTE_OP_RETRIES,
         },
         span::debug_assert_current_span_has_tenant_id,
-        storage_layer::{layer::local_layer_path, LayerName, LayerVisibilityHint},
+        storage_layer::{layer::local_layer_path, LayerName},
         tasks::{warn_when_period_overrun, BackgroundLoopKind},
     },
     virtual_file::{on_fatal_io_error, MaybeFatalIo, VirtualFile},
@@ -296,9 +296,6 @@ impl SecondaryDetail {
                         }),
                         last_activity_ts: ods.access_time,
                         relative_last_activity: finite_f32::FiniteF32::ZERO,
-                        // Secondary location layers are presumed visible, because Covered layers
-                        // are excluded from the heatmap
-                        visibility: LayerVisibilityHint::Visible,
                     }
                 }));
 

pageserver/src/tenant/storage_layer/delta_layer.rs

@@ -42,7 +42,6 @@ use crate::tenant::vectored_blob_io::{
     VectoredReadPlanner,
 };
 use crate::tenant::PageReconstructError;
-use crate::virtual_file::owned_buffers_io::io_buf_ext::{FullSlice, IoBufExt};
 use crate::virtual_file::{self, VirtualFile};
 use crate::{walrecord, TEMP_FILE_SUFFIX};
 use crate::{DELTA_FILE_MAGIC, STORAGE_FORMAT_VERSION};
@@ -64,7 +63,6 @@ use std::os::unix::fs::FileExt;
 use std::str::FromStr;
 use std::sync::Arc;
 use tokio::sync::OnceCell;
-use tokio_epoll_uring::IoBufMut;
 use tracing::*;
 
 use utils::{
@@ -438,28 +436,19 @@ impl DeltaLayerWriterInner {
         ctx: &RequestContext,
     ) -> anyhow::Result<()> {
         let (_, res) = self
-            .put_value_bytes(
-                key,
-                lsn,
-                Value::ser(&val)?.slice_len(),
-                val.will_init(),
-                ctx,
-            )
+            .put_value_bytes(key, lsn, Value::ser(&val)?, val.will_init(), ctx)
             .await;
         res
     }
 
-    async fn put_value_bytes<Buf>(
+    async fn put_value_bytes(
         &mut self,
         key: Key,
         lsn: Lsn,
-        val: FullSlice<Buf>,
+        val: Vec<u8>,
         will_init: bool,
         ctx: &RequestContext,
-    ) -> (FullSlice<Buf>, anyhow::Result<()>)
-    where
-        Buf: IoBufMut + Send,
-    {
+    ) -> (Vec<u8>, anyhow::Result<()>) {
         assert!(
             self.lsn_range.start <= lsn,
             "lsn_start={}, lsn={}",
@@ -525,7 +514,7 @@ impl DeltaLayerWriterInner {
         file.seek(SeekFrom::Start(index_start_blk as u64 * PAGE_SZ as u64))
             .await?;
         for buf in block_buf.blocks {
-            let (_buf, res) = file.write_all(buf.slice_len(), ctx).await;
+            let (_buf, res) = file.write_all(buf, ctx).await;
             res?;
         }
         assert!(self.lsn_range.start < self.lsn_range.end);
@@ -545,7 +534,7 @@ impl DeltaLayerWriterInner {
         // TODO: could use smallvec here but it's a pain with Slice<T>
         Summary::ser_into(&summary, &mut buf)?;
         file.seek(SeekFrom::Start(0)).await?;
-        let (_buf, res) = file.write_all(buf.slice_len(), ctx).await;
+        let (_buf, res) = file.write_all(buf, ctx).await;
         res?;
 
         let metadata = file
@@ -657,17 +646,14 @@ impl DeltaLayerWriter {
             .await
     }
 
-    pub async fn put_value_bytes<Buf>(
+    pub async fn put_value_bytes(
         &mut self,
         key: Key,
         lsn: Lsn,
-        val: FullSlice<Buf>,
+        val: Vec<u8>,
         will_init: bool,
         ctx: &RequestContext,
-    ) -> (FullSlice<Buf>, anyhow::Result<()>)
-    where
-        Buf: IoBufMut + Send,
-    {
+    ) -> (Vec<u8>, anyhow::Result<()>) {
         self.inner
             .as_mut()
             .unwrap()
@@ -757,7 +743,7 @@ impl DeltaLayer {
         // TODO: could use smallvec here, but it's a pain with Slice<T>
         Summary::ser_into(&new_summary, &mut buf).context("serialize")?;
         file.seek(SeekFrom::Start(0)).await?;
-        let (_buf, res) = file.write_all(buf.slice_len(), ctx).await;
+        let (_buf, res) = file.write_all(buf, ctx).await;
         res?;
         Ok(())
     }
@@ -1034,7 +1020,7 @@ impl DeltaLayerInner {
                     for (_, blob_meta) in read.blobs_at.as_slice() {
                         reconstruct_state.on_key_error(
                             blob_meta.key,
-                            PageReconstructError::Other(anyhow!(
+                            PageReconstructError::from(anyhow!(
                                 "Failed to read blobs from virtual file {}: {}",
                                 self.file.path,
                                 kind
@@ -1061,7 +1047,7 @@ impl DeltaLayerInner {
                     Err(e) => {
                         reconstruct_state.on_key_error(
                             meta.meta.key,
-                            PageReconstructError::Other(anyhow!(e).context(format!(
+                            PageReconstructError::from(anyhow!(e).context(format!(
                                 "Failed to deserialize blob from virtual file {}",
                                 self.file.path,
                             ))),
@@ -1305,12 +1291,12 @@ impl DeltaLayerInner {
                         .put_value_bytes(
                             key,
                             lsn,
-                            std::mem::take(&mut per_blob_copy).slice_len(),
+                            std::mem::take(&mut per_blob_copy),
                             will_init,
                             ctx,
                         )
                         .await;
-                    per_blob_copy = tmp.into_raw_slice().into_inner();
+                    per_blob_copy = tmp;
 
                     res?;
 
@@ -1885,7 +1871,7 @@ pub(crate) mod test {
 
         for entry in entries {
             let (_, res) = writer
-                .put_value_bytes(entry.key, entry.lsn, entry.value.slice_len(), false, &ctx)
+                .put_value_bytes(entry.key, entry.lsn, entry.value, false, &ctx)
                 .await;
             res?;
         }

pageserver/src/tenant/storage_layer/image_layer.rs

@@ -38,7 +38,6 @@ use crate::tenant::vectored_blob_io::{
     VectoredReadPlanner,
 };
 use crate::tenant::{PageReconstructError, Timeline};
-use crate::virtual_file::owned_buffers_io::io_buf_ext::IoBufExt;
 use crate::virtual_file::{self, VirtualFile};
 use crate::{IMAGE_FILE_MAGIC, STORAGE_FORMAT_VERSION, TEMP_FILE_SUFFIX};
 use anyhow::{anyhow, bail, ensure, Context, Result};
@@ -355,7 +354,7 @@ impl ImageLayer {
         // TODO: could use smallvec here but it's a pain with Slice<T>
         Summary::ser_into(&new_summary, &mut buf).context("serialize")?;
         file.seek(SeekFrom::Start(0)).await?;
-        let (_buf, res) = file.write_all(buf.slice_len(), ctx).await;
+        let (_buf, res) = file.write_all(buf, ctx).await;
         res?;
         Ok(())
     }
@@ -787,7 +786,7 @@ impl ImageLayerWriterInner {
         self.num_keys += 1;
         let (_img, res) = self
             .blob_writer
-            .write_blob_maybe_compressed(img.slice_len(), ctx, compression)
+            .write_blob_maybe_compressed(img, ctx, compression)
             .await;
         // TODO: re-use the buffer for `img` further upstack
         let (off, compression_info) = res?;
@@ -839,7 +838,7 @@ impl ImageLayerWriterInner {
             .await?;
         let (index_root_blk, block_buf) = self.tree.finish()?;
         for buf in block_buf.blocks {
-            let (_buf, res) = file.write_all(buf.slice_len(), ctx).await;
+            let (_buf, res) = file.write_all(buf, ctx).await;
             res?;
         }
 
@@ -859,7 +858,7 @@ impl ImageLayerWriterInner {
         // TODO: could use smallvec here but it's a pain with Slice<T>
         Summary::ser_into(&summary, &mut buf)?;
         file.seek(SeekFrom::Start(0)).await?;
-        let (_buf, res) = file.write_all(buf.slice_len(), ctx).await;
+        let (_buf, res) = file.write_all(buf, ctx).await;
         res?;
 
         let metadata = file

pageserver/src/tenant/storage_layer/inmemory_layer.rs

@@ -12,7 +12,6 @@ use crate::tenant::block_io::{BlockCursor, BlockReader, BlockReaderRef};
 use crate::tenant::ephemeral_file::EphemeralFile;
 use crate::tenant::timeline::GetVectoredError;
 use crate::tenant::PageReconstructError;
-use crate::virtual_file::owned_buffers_io::io_buf_ext::IoBufExt;
 use crate::{l0_flush, page_cache, walrecord};
 use anyhow::{anyhow, Result};
 use camino::Utf8PathBuf;
@@ -582,17 +581,11 @@ impl InMemoryLayer {
                     for (lsn, pos) in vec_map.as_slice() {
                         cursor.read_blob_into_buf(*pos, &mut buf, &ctx).await?;
                         let will_init = Value::des(&buf)?.will_init();
-                        let (tmp, res) = delta_layer_writer
-                            .put_value_bytes(
-                                Key::from_compact(*key),
-                                *lsn,
-                                buf.slice_len(),
-                                will_init,
-                                &ctx,
-                            )
+                        let res;
+                        (buf, res) = delta_layer_writer
+                            .put_value_bytes(Key::from_compact(*key), *lsn, buf, will_init, &ctx)
                             .await;
                         res?;
-                        buf = tmp.into_raw_slice().into_inner();
                     }
                 }
             }
@@ -627,17 +620,11 @@ impl InMemoryLayer {
                         // => https://github.com/neondatabase/neon/issues/8183
                         cursor.read_blob_into_buf(*pos, &mut buf, ctx).await?;
                         let will_init = Value::des(&buf)?.will_init();
-                        let (tmp, res) = delta_layer_writer
-                            .put_value_bytes(
-                                Key::from_compact(*key),
-                                *lsn,
-                                buf.slice_len(),
-                                will_init,
-                                ctx,
-                            )
+                        let res;
+                        (buf, res) = delta_layer_writer
+                            .put_value_bytes(Key::from_compact(*key), *lsn, buf, will_init, ctx)
                             .await;
                         res?;
-                        buf = tmp.into_raw_slice().into_inner();
                     }
                 }
             }

pageserver/src/tenant/storage_layer/layer.rs

@@ -312,9 +312,7 @@ impl Layer {
             .get_or_maybe_download(true, Some(ctx))
             .await
             .map_err(|err| match err {
-                DownloadError::TimelineShutdown | DownloadError::DownloadCancelled => {
-                    GetVectoredError::Cancelled
-                }
+                DownloadError::DownloadCancelled => GetVectoredError::Cancelled,
                 other => GetVectoredError::Other(anyhow::anyhow!(other)),
             })?;
 

pageserver/src/tenant/timeline.rs

@@ -511,7 +511,7 @@ pub(crate) struct TimelineVisitOutcome {
 #[derive(thiserror::Error, Debug)]
 pub(crate) enum PageReconstructError {
     #[error(transparent)]
-    Other(anyhow::Error),
+    Other(#[from] anyhow::Error),
 
     #[error("Ancestor LSN wait error: {0}")]
     AncestorLsnTimeout(WaitLsnError),
@@ -527,22 +527,6 @@ pub(crate) enum PageReconstructError {
     MissingKey(MissingKeyError),
 }
 
-impl From<anyhow::Error> for PageReconstructError {
-    fn from(value: anyhow::Error) -> Self {
-        // with walingest.rs many PageReconstructError are wrapped in as anyhow::Error
-        match value.downcast::<PageReconstructError>() {
-            Ok(pre) => pre,
-            Err(other) => PageReconstructError::Other(other),
-        }
-    }
-}
-
-impl From<utils::bin_ser::DeserializeError> for PageReconstructError {
-    fn from(value: utils::bin_ser::DeserializeError) -> Self {
-        PageReconstructError::Other(anyhow::Error::new(value).context("deserialization failure"))
-    }
-}
-
 impl From<layer_manager::Shutdown> for PageReconstructError {
     fn from(_: layer_manager::Shutdown) -> Self {
         PageReconstructError::Cancelled
@@ -562,7 +546,6 @@ impl From<layer_manager::Shutdown> for GetVectoredError {
     }
 }
 
-#[derive(thiserror::Error)]
 pub struct MissingKeyError {
     key: Key,
     shard: ShardNumber,
@@ -602,8 +585,11 @@ impl PageReconstructError {
     pub(crate) fn is_stopping(&self) -> bool {
         use PageReconstructError::*;
         match self {
+            Other(_) => false,
+            AncestorLsnTimeout(_) => false,
             Cancelled => true,
-            Other(_) | AncestorLsnTimeout(_) | WalRedo(_) | MissingKey(_) => false,
+            WalRedo(_) => false,
+            MissingKey { .. } => false,
         }
     }
 }
@@ -613,11 +599,11 @@ pub(crate) enum CreateImageLayersError {
     #[error("timeline shutting down")]
     Cancelled,
 
-    #[error("read failed")]
-    GetVectoredError(#[source] GetVectoredError),
+    #[error(transparent)]
+    GetVectoredError(GetVectoredError),
 
-    #[error("reconstruction failed")]
-    PageReconstructError(#[source] PageReconstructError),
+    #[error(transparent)]
+    PageReconstructError(PageReconstructError),
 
     #[error(transparent)]
     Other(#[from] anyhow::Error),
@@ -641,10 +627,10 @@ pub(crate) enum FlushLayerError {
 
     // Arc<> the following non-clonable error types: we must be Clone-able because the flush error is propagated from the flush
     // loop via a watch channel, where we can only borrow it.
-    #[error("create image layers (shared)")]
+    #[error(transparent)]
     CreateImageLayersError(Arc<CreateImageLayersError>),
 
-    #[error("other (shared)")]
+    #[error(transparent)]
     Other(#[from] Arc<anyhow::Error>),
 }
 
@@ -677,46 +663,34 @@ pub(crate) enum GetVectoredError {
     #[error("timeline shutting down")]
     Cancelled,
 
-    #[error("requested too many keys: {0} > {}", Timeline::MAX_GET_VECTORED_KEYS)]
+    #[error("Requested too many keys: {0} > {}", Timeline::MAX_GET_VECTORED_KEYS)]
     Oversized(u64),
 
-    #[error("requested at invalid LSN: {0}")]
+    #[error("Requested at invalid LSN: {0}")]
     InvalidLsn(Lsn),
 
-    #[error("requested key not found: {0}")]
+    #[error("Requested key not found: {0}")]
     MissingKey(MissingKeyError),
 
-    #[error("ancestry walk")]
-    GetReadyAncestorError(#[source] GetReadyAncestorError),
+    #[error(transparent)]
+    GetReadyAncestorError(GetReadyAncestorError),
 
     #[error(transparent)]
     Other(#[from] anyhow::Error),
 }
 
-impl From<GetReadyAncestorError> for GetVectoredError {
-    fn from(value: GetReadyAncestorError) -> Self {
-        use GetReadyAncestorError::*;
-        match value {
-            Cancelled => GetVectoredError::Cancelled,
-            AncestorLsnTimeout(_) | BadState { .. } => {
-                GetVectoredError::GetReadyAncestorError(value)
-            }
-        }
-    }
-}
-
 #[derive(thiserror::Error, Debug)]
 pub(crate) enum GetReadyAncestorError {
-    #[error("ancestor LSN wait error")]
+    #[error("Ancestor LSN wait error: {0}")]
     AncestorLsnTimeout(#[from] WaitLsnError),
 
-    #[error("bad state on timeline {timeline_id}: {state:?}")]
+    #[error("Bad state on timeline {timeline_id}: {state:?}")]
     BadState {
         timeline_id: TimelineId,
         state: TimelineState,
     },
 
-    #[error("cancelled")]
+    #[error("Cancelled")]
     Cancelled,
 }
 
@@ -3072,7 +3046,8 @@ impl Timeline {
             cont_lsn = std::cmp::min(Lsn(request_lsn.0 + 1), Lsn(timeline.ancestor_lsn.0 + 1));
             timeline_owned = timeline
                 .get_ready_ancestor_timeline(ancestor_timeline, ctx)
-                .await?;
+                .await
+                .map_err(GetVectoredError::GetReadyAncestorError)?;
             timeline = &*timeline_owned;
         };
 
@@ -3969,7 +3944,7 @@ impl Timeline {
                                     warn!("could not reconstruct FSM or VM key {img_key}, filling with zeros: {err:?}");
                                     ZERO_PAGE.clone()
                                 } else {
-                                    return Err(CreateImageLayersError::from(err));
+                                    return Err(CreateImageLayersError::PageReconstructError(err));
                                 }
                             }
                         };
@@ -4029,7 +4004,7 @@ impl Timeline {
             let mut total_kb_retrieved = 0;
             let mut total_keys_retrieved = 0;
             for (k, v) in data {
-                let v = v?;
+                let v = v.map_err(CreateImageLayersError::PageReconstructError)?;
                 total_kb_retrieved += KEY_SIZE + v.len();
                 total_keys_retrieved += 1;
                 new_data.insert(k, v);
@@ -4540,12 +4515,7 @@ impl Timeline {
         new_images: &[ResidentLayer],
         layers_to_remove: &[Layer],
     ) -> Result<(), CompactionError> {
-        let mut guard = tokio::select! {
-            guard = self.layers.write() => guard,
-            _ = self.cancel.cancelled() => {
-                return Err(CompactionError::ShuttingDown);
-            }
-        };
+        let mut guard = self.layers.write().await;
 
         let mut duplicated_layers = HashSet::new();
 
@@ -5291,7 +5261,6 @@ impl Timeline {
                     layer: layer.to_owned().into(),
                     last_activity_ts,
                     relative_last_activity: finite_f32::FiniteF32::ZERO,
-                    visibility: layer.visibility(),
                 }
             })
             .collect();

pageserver/src/tenant/timeline/compaction.rs

@@ -1048,22 +1048,11 @@ impl Timeline {
         let mut dup_end_lsn: Lsn = Lsn::INVALID; // end LSN of layer containing values of the single key
         let mut next_hole = 0; // index of next hole in holes vector
 
-        let mut keys = 0;
-
         while let Some((key, lsn, value)) = all_values_iter
             .next(ctx)
             .await
             .map_err(CompactionError::Other)?
         {
-            keys += 1;
-
-            if keys % 32_768 == 0 && self.cancel.is_cancelled() {
-                // avoid hitting the cancellation token on every key. in benches, we end up
-                // shuffling an order of million keys per layer, this means we'll check it
-                // around tens of times per layer.
-                return Err(CompactionError::ShuttingDown);
-            }
-
             let same_key = prev_key.map_or(false, |prev_key| prev_key == key);
             // We need to check key boundaries once we reach next key or end of layer with the same key
             if !same_key || lsn == dup_end_lsn {
@@ -1168,8 +1157,6 @@ impl Timeline {
                         .await
                         .map_err(CompactionError::Other)?,
                     );
-
-                    keys = 0;
                 }
 
                 writer
@@ -2338,7 +2325,7 @@ impl CompactionJobExecutor for TimelineAdaptor {
                 key_range,
             ))
         } else {
-            // The current compaction implementation only ever requests the key space
+            // The current compaction implementatin only ever requests the key space
             // at the compaction end LSN.
             anyhow::bail!("keyspace not available for requested lsn");
         }

pageserver/src/tenant/timeline/eviction_task.rs

@@ -30,8 +30,7 @@ use crate::{
     pgdatadir_mapping::CollectKeySpaceError,
     task_mgr::{self, TaskKind, BACKGROUND_RUNTIME},
     tenant::{
-        storage_layer::LayerVisibilityHint, tasks::BackgroundLoopKind, timeline::EvictionError,
-        LogicalSizeCalculationCause, Tenant,
+        tasks::BackgroundLoopKind, timeline::EvictionError, LogicalSizeCalculationCause, Tenant,
     },
 };
 
@@ -242,22 +241,7 @@ impl Timeline {
                         }
                     };
 
-                    match layer.visibility() {
-                        LayerVisibilityHint::Visible => {
-                            // Usual case: a visible layer might be read any time, and we will keep it
-                            // resident until it hits our configured TTL threshold.
-                            no_activity_for > p.threshold
-                        }
-                        LayerVisibilityHint::Covered => {
-                            // Covered layers: this is probably a layer that was recently covered by
-                            // an image layer during compaction.  We don't evict it immediately, but
-                            // it doesn't stay resident for the full `threshold`: we just keep it
-                            // for a shorter time in case
-                            // - it is used for Timestamp->LSN lookups
-                            // - a new branch is created in recent history which will read this layer
-                            no_activity_for > p.period
-                        }
-                    }
+                    no_activity_for > p.threshold
                 })
                 .cloned()
                 .for_each(|layer| {

pageserver/src/virtual_file.rs

@@ -17,7 +17,6 @@ use crate::page_cache::{PageWriteGuard, PAGE_SZ};
 use crate::tenant::TENANTS_SEGMENT_NAME;
 use camino::{Utf8Path, Utf8PathBuf};
 use once_cell::sync::OnceCell;
-use owned_buffers_io::io_buf_ext::FullSlice;
 use pageserver_api::shard::TenantShardId;
 use std::fs::File;
 use std::io::{Error, ErrorKind, Seek, SeekFrom};
@@ -51,7 +50,6 @@ pub(crate) mod owned_buffers_io {
     //! but for the time being we're proving out the primitives in the neon.git repo
     //! for faster iteration.
 
-    pub(crate) mod io_buf_ext;
     pub(crate) mod slice;
     pub(crate) mod write;
     pub(crate) mod util {
@@ -639,24 +637,24 @@ impl VirtualFile {
     }
 
     // Copied from https://doc.rust-lang.org/1.72.0/src/std/os/unix/fs.rs.html#219-235
-    pub async fn write_all_at<Buf: IoBuf + Send>(
+    pub async fn write_all_at<B: BoundedBuf<Buf = Buf>, Buf: IoBuf + Send>(
         &self,
-        buf: FullSlice<Buf>,
+        buf: B,
         mut offset: u64,
         ctx: &RequestContext,
-    ) -> (FullSlice<Buf>, Result<(), Error>) {
-        let buf = buf.into_raw_slice();
-        let bounds = buf.bounds();
-        let restore =
-            |buf: Slice<_>| FullSlice::must_new(Slice::from_buf_bounds(buf.into_inner(), bounds));
-        let mut buf = buf;
+    ) -> (B::Buf, Result<(), Error>) {
+        let buf_len = buf.bytes_init();
+        if buf_len == 0 {
+            return (Slice::into_inner(buf.slice_full()), Ok(()));
+        }
+        let mut buf = buf.slice(0..buf_len);
         while !buf.is_empty() {
-            let (tmp, res) = self.write_at(FullSlice::must_new(buf), offset, ctx).await;
-            buf = tmp.into_raw_slice();
+            let res;
+            (buf, res) = self.write_at(buf, offset, ctx).await;
             match res {
                 Ok(0) => {
                     return (
-                        restore(buf),
+                        Slice::into_inner(buf),
                         Err(Error::new(
                             std::io::ErrorKind::WriteZero,
                             "failed to write whole buffer",
@@ -668,33 +666,33 @@ impl VirtualFile {
                     offset += n as u64;
                 }
                 Err(e) if e.kind() == std::io::ErrorKind::Interrupted => {}
-                Err(e) => return (restore(buf), Err(e)),
+                Err(e) => return (Slice::into_inner(buf), Err(e)),
             }
         }
-        (restore(buf), Ok(()))
+        (Slice::into_inner(buf), Ok(()))
     }
 
-    /// Writes `buf` to the file at the current offset.
-    ///
-    /// Panics if there is an uninitialized range in `buf`, as that is most likely a bug in the caller.
-    pub async fn write_all<Buf: IoBuf + Send>(
+    /// Writes `buf.slice(0..buf.bytes_init())`.
+    /// Returns the IoBuf that is underlying the BoundedBuf `buf`.
+    /// I.e., the returned value's `bytes_init()` method returns something different than the `bytes_init()` that was passed in.
+    /// It's quite brittle and easy to mis-use, so, we return the size in the Ok() variant.
+    pub async fn write_all<B: BoundedBuf<Buf = Buf>, Buf: IoBuf + Send>(
         &mut self,
-        buf: FullSlice<Buf>,
+        buf: B,
         ctx: &RequestContext,
-    ) -> (FullSlice<Buf>, Result<usize, Error>) {
-        let buf = buf.into_raw_slice();
-        let bounds = buf.bounds();
-        let restore =
-            |buf: Slice<_>| FullSlice::must_new(Slice::from_buf_bounds(buf.into_inner(), bounds));
-        let nbytes = buf.len();
-        let mut buf = buf;
+    ) -> (B::Buf, Result<usize, Error>) {
+        let nbytes = buf.bytes_init();
+        if nbytes == 0 {
+            return (Slice::into_inner(buf.slice_full()), Ok(0));
+        }
+        let mut buf = buf.slice(0..nbytes);
         while !buf.is_empty() {
-            let (tmp, res) = self.write(FullSlice::must_new(buf), ctx).await;
-            buf = tmp.into_raw_slice();
+            let res;
+            (buf, res) = self.write(buf, ctx).await;
             match res {
                 Ok(0) => {
                     return (
-                        restore(buf),
+                        Slice::into_inner(buf),
                         Err(Error::new(
                             std::io::ErrorKind::WriteZero,
                             "failed to write whole buffer",
@@ -705,17 +703,17 @@ impl VirtualFile {
                     buf = buf.slice(n..);
                 }
                 Err(ref e) if e.kind() == std::io::ErrorKind::Interrupted => {}
-                Err(e) => return (restore(buf), Err(e)),
+                Err(e) => return (Slice::into_inner(buf), Err(e)),
             }
         }
-        (restore(buf), Ok(nbytes))
+        (Slice::into_inner(buf), Ok(nbytes))
     }
 
     async fn write<B: IoBuf + Send>(
         &mut self,
-        buf: FullSlice<B>,
+        buf: Slice<B>,
         ctx: &RequestContext,
-    ) -> (FullSlice<B>, Result<usize, std::io::Error>) {
+    ) -> (Slice<B>, Result<usize, std::io::Error>) {
         let pos = self.pos;
         let (buf, res) = self.write_at(buf, pos, ctx).await;
         let n = match res {
@@ -758,10 +756,10 @@ impl VirtualFile {
 
     async fn write_at<B: IoBuf + Send>(
         &self,
-        buf: FullSlice<B>,
+        buf: Slice<B>,
         offset: u64,
         _ctx: &RequestContext, /* TODO: use for metrics: https://github.com/neondatabase/neon/issues/6107 */
-    ) -> (FullSlice<B>, Result<usize, Error>) {
+    ) -> (Slice<B>, Result<usize, Error>) {
         let file_guard = match self.lock_file().await {
             Ok(file_guard) => file_guard,
             Err(e) => return (buf, Err(e)),
@@ -1095,11 +1093,11 @@ impl Drop for VirtualFile {
 
 impl OwnedAsyncWriter for VirtualFile {
     #[inline(always)]
-    async fn write_all<Buf: IoBuf + Send>(
+    async fn write_all<B: BoundedBuf<Buf = Buf>, Buf: IoBuf + Send>(
         &mut self,
-        buf: FullSlice<Buf>,
+        buf: B,
         ctx: &RequestContext,
-    ) -> std::io::Result<(usize, FullSlice<Buf>)> {
+    ) -> std::io::Result<(usize, B::Buf)> {
         let (buf, res) = VirtualFile::write_all(self, buf, ctx).await;
         res.map(move |v| (v, buf))
     }
@@ -1161,8 +1159,7 @@ mod tests {
     use crate::task_mgr::TaskKind;
 
     use super::*;
-    use owned_buffers_io::io_buf_ext::IoBufExt;
-    use owned_buffers_io::slice::SliceMutExt;
+    use owned_buffers_io::slice::SliceExt;
     use rand::seq::SliceRandom;
     use rand::thread_rng;
     use rand::Rng;
@@ -1196,9 +1193,9 @@ mod tests {
                 }
             }
         }
-        async fn write_all_at<Buf: IoBuf + Send>(
+        async fn write_all_at<B: BoundedBuf<Buf = Buf>, Buf: IoBuf + Send>(
             &self,
-            buf: FullSlice<Buf>,
+            buf: B,
             offset: u64,
             ctx: &RequestContext,
         ) -> Result<(), Error> {
@@ -1207,7 +1204,13 @@ mod tests {
                     let (_buf, res) = file.write_all_at(buf, offset, ctx).await;
                     res
                 }
-                MaybeVirtualFile::File(file) => file.write_all_at(&buf[..], offset),
+                MaybeVirtualFile::File(file) => {
+                    let buf_len = buf.bytes_init();
+                    if buf_len == 0 {
+                        return Ok(());
+                    }
+                    file.write_all_at(&buf.slice(0..buf_len), offset)
+                }
             }
         }
         async fn seek(&mut self, pos: SeekFrom) -> Result<u64, Error> {
@@ -1216,9 +1219,9 @@ mod tests {
                 MaybeVirtualFile::File(file) => file.seek(pos),
             }
         }
-        async fn write_all<Buf: IoBuf + Send>(
+        async fn write_all<B: BoundedBuf<Buf = Buf>, Buf: IoBuf + Send>(
             &mut self,
-            buf: FullSlice<Buf>,
+            buf: B,
             ctx: &RequestContext,
         ) -> Result<(), Error> {
             match self {
@@ -1226,7 +1229,13 @@ mod tests {
                     let (_buf, res) = file.write_all(buf, ctx).await;
                     res.map(|_| ())
                 }
-                MaybeVirtualFile::File(file) => file.write_all(&buf[..]),
+                MaybeVirtualFile::File(file) => {
+                    let buf_len = buf.bytes_init();
+                    if buf_len == 0 {
+                        return Ok(());
+                    }
+                    file.write_all(&buf.slice(0..buf_len))
+                }
             }
         }
 
@@ -1338,9 +1347,7 @@ mod tests {
             &ctx,
         )
         .await?;
-        file_a
-            .write_all(b"foobar".to_vec().slice_len(), &ctx)
-            .await?;
+        file_a.write_all(b"foobar".to_vec(), &ctx).await?;
 
         // cannot read from a file opened in write-only mode
         let _ = file_a.read_string(&ctx).await.unwrap_err();
@@ -1349,10 +1356,7 @@ mod tests {
         let mut file_a = A::open(path_a, OpenOptions::new().read(true).to_owned(), &ctx).await?;
 
         // cannot write to a file opened in read-only mode
-        let _ = file_a
-            .write_all(b"bar".to_vec().slice_len(), &ctx)
-            .await
-            .unwrap_err();
+        let _ = file_a.write_all(b"bar".to_vec(), &ctx).await.unwrap_err();
 
         // Try simple read
         assert_eq!("foobar", file_a.read_string(&ctx).await?);
@@ -1395,12 +1399,8 @@ mod tests {
             &ctx,
         )
         .await?;
-        file_b
-            .write_all_at(b"BAR".to_vec().slice_len(), 3, &ctx)
-            .await?;
-        file_b
-            .write_all_at(b"FOO".to_vec().slice_len(), 0, &ctx)
-            .await?;
+        file_b.write_all_at(b"BAR".to_vec(), 3, &ctx).await?;
+        file_b.write_all_at(b"FOO".to_vec(), 0, &ctx).await?;
 
         assert_eq!(file_b.read_string_at(2, 3, &ctx).await?, "OBA");
 

pageserver/src/virtual_file/io_engine.rs

@@ -12,7 +12,7 @@
 #[cfg(target_os = "linux")]
 pub(super) mod tokio_epoll_uring_ext;
 
-use tokio_epoll_uring::IoBuf;
+use tokio_epoll_uring::{IoBuf, Slice};
 use tracing::Instrument;
 
 pub(crate) use super::api::IoEngineKind;
@@ -107,10 +107,7 @@ use std::{
     sync::atomic::{AtomicU8, Ordering},
 };
 
-use super::{
-    owned_buffers_io::{io_buf_ext::FullSlice, slice::SliceMutExt},
-    FileGuard, Metadata,
-};
+use super::{owned_buffers_io::slice::SliceExt, FileGuard, Metadata};
 
 #[cfg(target_os = "linux")]
 fn epoll_uring_error_to_std(e: tokio_epoll_uring::Error<std::io::Error>) -> std::io::Error {
@@ -209,8 +206,8 @@ impl IoEngine {
         &self,
         file_guard: FileGuard,
         offset: u64,
-        buf: FullSlice<B>,
-    ) -> ((FileGuard, FullSlice<B>), std::io::Result<usize>) {
+        buf: Slice<B>,
+    ) -> ((FileGuard, Slice<B>), std::io::Result<usize>) {
         match self {
             IoEngine::NotSet => panic!("not initialized"),
             IoEngine::StdFs => {
@@ -220,12 +217,8 @@ impl IoEngine {
             #[cfg(target_os = "linux")]
             IoEngine::TokioEpollUring => {
                 let system = tokio_epoll_uring_ext::thread_local_system().await;
-                let ((file_guard, slice), res) =
-                    system.write(file_guard, offset, buf.into_raw_slice()).await;
-                (
-                    (file_guard, FullSlice::must_new(slice)),
-                    res.map_err(epoll_uring_error_to_std),
-                )
+                let (resources, res) = system.write(file_guard, offset, buf).await;
+                (resources, res.map_err(epoll_uring_error_to_std))
             }
         }
     }

pageserver/src/virtual_file/owned_buffers_io/io_buf_ext.rs

@@ -1,78 +0,0 @@
-//! See [`FullSlice`].
-
-use bytes::{Bytes, BytesMut};
-use std::ops::{Deref, Range};
-use tokio_epoll_uring::{BoundedBuf, IoBuf, Slice};
-
-/// The true owned equivalent for Rust [`slice`]. Use this for the write path.
-///
-/// Unlike [`tokio_epoll_uring::Slice`], which we unfortunately inherited from `tokio-uring`,
-/// [`FullSlice`] is guaranteed to have all its bytes initialized. This means that
-/// [`<FullSlice as Deref<Target = [u8]>>::len`] is equal to [`Slice::bytes_init`] and [`Slice::bytes_total`].
-///
-pub struct FullSlice<B> {
-    slice: Slice<B>,
-}
-
-impl<B> FullSlice<B>
-where
-    B: IoBuf,
-{
-    pub(crate) fn must_new(slice: Slice<B>) -> Self {
-        assert_eq!(slice.bytes_init(), slice.bytes_total());
-        FullSlice { slice }
-    }
-    pub(crate) fn into_raw_slice(self) -> Slice<B> {
-        let FullSlice { slice: s } = self;
-        s
-    }
-}
-
-impl<B> Deref for FullSlice<B>
-where
-    B: IoBuf,
-{
-    type Target = [u8];
-
-    fn deref(&self) -> &[u8] {
-        let rust_slice = &self.slice[..];
-        assert_eq!(rust_slice.len(), self.slice.bytes_init());
-        assert_eq!(rust_slice.len(), self.slice.bytes_total());
-        rust_slice
-    }
-}
-
-pub(crate) trait IoBufExt {
-    /// Get a [`FullSlice`] for the entire buffer, i.e., `self[..]` or `self[0..self.len()]`.
-    fn slice_len(self) -> FullSlice<Self>
-    where
-        Self: Sized;
-}
-
-macro_rules! impl_io_buf_ext {
-    ($T:ty) => {
-        impl IoBufExt for $T {
-            #[inline(always)]
-            fn slice_len(self) -> FullSlice<Self> {
-                let len = self.len();
-                let s = if len == 0 {
-                    // `BoundedBuf::slice(0..len)` or `BoundedBuf::slice(..)` has an incorrect assertion,
-                    // causing a panic if len == 0.
-                    // The Slice::from_buf_bounds has the correct assertion (<= instead of <).
-                    // => https://github.com/neondatabase/tokio-epoll-uring/issues/46
-                    let slice = self.slice_full();
-                    let mut bounds: Range<_> = slice.bounds();
-                    bounds.end = bounds.start;
-                    Slice::from_buf_bounds(slice.into_inner(), bounds)
-                } else {
-                    self.slice(0..len)
-                };
-                FullSlice::must_new(s)
-            }
-        }
-    };
-}
-
-impl_io_buf_ext!(Bytes);
-impl_io_buf_ext!(BytesMut);
-impl_io_buf_ext!(Vec<u8>);

pageserver/src/virtual_file/owned_buffers_io/slice.rs

@@ -3,14 +3,14 @@ use tokio_epoll_uring::BoundedBufMut;
 use tokio_epoll_uring::IoBufMut;
 use tokio_epoll_uring::Slice;
 
-pub(crate) trait SliceMutExt {
+pub(crate) trait SliceExt {
     /// Get a `&mut[0..self.bytes_total()`] slice, for when you need to do borrow-based IO.
     ///
     /// See the test case `test_slice_full_zeroed` for the difference to just doing `&slice[..]`
     fn as_mut_rust_slice_full_zeroed(&mut self) -> &mut [u8];
 }
 
-impl<B> SliceMutExt for Slice<B>
+impl<B> SliceExt for Slice<B>
 where
     B: IoBufMut,
 {

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

@@ -1,8 +1,5 @@
-use crate::{
-    context::RequestContext,
-    virtual_file::owned_buffers_io::{io_buf_ext::FullSlice, write::OwnedAsyncWriter},
-};
-use tokio_epoll_uring::IoBuf;
+use crate::{context::RequestContext, virtual_file::owned_buffers_io::write::OwnedAsyncWriter};
+use tokio_epoll_uring::{BoundedBuf, IoBuf};
 
 pub struct Writer<W> {
     dst: W,
@@ -38,11 +35,11 @@ where
     W: OwnedAsyncWriter,
 {
     #[inline(always)]
-    async fn write_all<Buf: IoBuf + Send>(
+    async fn write_all<B: BoundedBuf<Buf = Buf>, Buf: IoBuf + Send>(
         &mut self,
-        buf: FullSlice<Buf>,
+        buf: B,
         ctx: &RequestContext,
-    ) -> std::io::Result<(usize, FullSlice<Buf>)> {
+    ) -> std::io::Result<(usize, B::Buf)> {
         let (nwritten, buf) = self.dst.write_all(buf, ctx).await?;
         self.bytes_amount += u64::try_from(nwritten).unwrap();
         Ok((nwritten, buf))

pageserver/src/virtual_file/owned_buffers_io/write.rs

@@ -1,18 +1,16 @@
 use bytes::BytesMut;
-use tokio_epoll_uring::IoBuf;
+use tokio_epoll_uring::{BoundedBuf, IoBuf, Slice};
 
 use crate::context::RequestContext;
 
-use super::io_buf_ext::{FullSlice, IoBufExt};
-
 /// A trait for doing owned-buffer write IO.
 /// Think [`tokio::io::AsyncWrite`] but with owned buffers.
 pub trait OwnedAsyncWriter {
-    async fn write_all<Buf: IoBuf + Send>(
+    async fn write_all<B: BoundedBuf<Buf = Buf>, Buf: IoBuf + Send>(
         &mut self,
-        buf: FullSlice<Buf>,
+        buf: B,
         ctx: &RequestContext,
-    ) -> std::io::Result<(usize, FullSlice<Buf>)>;
+    ) -> std::io::Result<(usize, B::Buf)>;
 }
 
 /// A wrapper aorund an [`OwnedAsyncWriter`] that uses a [`Buffer`] to batch
@@ -81,11 +79,9 @@ where
     #[cfg_attr(target_os = "macos", allow(dead_code))]
     pub async fn write_buffered<S: IoBuf + Send>(
         &mut self,
-        chunk: FullSlice<S>,
+        chunk: Slice<S>,
         ctx: &RequestContext,
-    ) -> std::io::Result<(usize, FullSlice<S>)> {
-        let chunk = chunk.into_raw_slice();
-
+    ) -> std::io::Result<(usize, S)> {
         let chunk_len = chunk.len();
         // avoid memcpy for the middle of the chunk
         if chunk.len() >= self.buf().cap() {
@@ -98,10 +94,7 @@ where
                     .pending(),
                 0
             );
-            let (nwritten, chunk) = self
-                .writer
-                .write_all(FullSlice::must_new(chunk), ctx)
-                .await?;
+            let (nwritten, chunk) = self.writer.write_all(chunk, ctx).await?;
             assert_eq!(nwritten, chunk_len);
             return Ok((nwritten, chunk));
         }
@@ -121,7 +114,7 @@ where
             }
         }
         assert!(slice.is_empty(), "by now we should have drained the chunk");
-        Ok((chunk_len, FullSlice::must_new(chunk)))
+        Ok((chunk_len, chunk.into_inner()))
     }
 
     /// Strictly less performant variant of [`Self::write_buffered`] that allows writing borrowed data.
@@ -157,12 +150,9 @@ where
             self.buf = Some(buf);
             return Ok(());
         }
-        let slice = buf.flush();
-        let (nwritten, slice) = self.writer.write_all(slice, ctx).await?;
+        let (nwritten, io_buf) = self.writer.write_all(buf.flush(), ctx).await?;
         assert_eq!(nwritten, buf_len);
-        self.buf = Some(Buffer::reuse_after_flush(
-            slice.into_raw_slice().into_inner(),
-        ));
+        self.buf = Some(Buffer::reuse_after_flush(io_buf));
         Ok(())
     }
 }
@@ -182,9 +172,9 @@ pub trait Buffer {
     /// Number of bytes in the buffer.
     fn pending(&self) -> usize;
 
-    /// Turns `self` into a [`FullSlice`] of the pending data
+    /// Turns `self` into a [`tokio_epoll_uring::Slice`] of the pending data
     /// so we can use [`tokio_epoll_uring`] to write it to disk.
-    fn flush(self) -> FullSlice<Self::IoBuf>;
+    fn flush(self) -> Slice<Self::IoBuf>;
 
     /// After the write to disk is done and we have gotten back the slice,
     /// [`BufferedWriter`] uses this method to re-use the io buffer.
@@ -208,8 +198,12 @@ impl Buffer for BytesMut {
         self.len()
     }
 
-    fn flush(self) -> FullSlice<BytesMut> {
-        self.slice_len()
+    fn flush(self) -> Slice<BytesMut> {
+        if self.is_empty() {
+            return self.slice_full();
+        }
+        let len = self.len();
+        self.slice(0..len)
     }
 
     fn reuse_after_flush(mut iobuf: BytesMut) -> Self {
@@ -219,13 +213,18 @@ impl Buffer for BytesMut {
 }
 
 impl OwnedAsyncWriter for Vec<u8> {
-    async fn write_all<Buf: IoBuf + Send>(
+    async fn write_all<B: BoundedBuf<Buf = Buf>, Buf: IoBuf + Send>(
         &mut self,
-        buf: FullSlice<Buf>,
+        buf: B,
         _: &RequestContext,
-    ) -> std::io::Result<(usize, FullSlice<Buf>)> {
+    ) -> std::io::Result<(usize, B::Buf)> {
+        let nbytes = buf.bytes_init();
+        if nbytes == 0 {
+            return Ok((0, Slice::into_inner(buf.slice_full())));
+        }
+        let buf = buf.slice(0..nbytes);
         self.extend_from_slice(&buf[..]);
-        Ok((buf.len(), buf))
+        Ok((buf.len(), Slice::into_inner(buf)))
     }
 }
 
@@ -242,13 +241,19 @@ mod tests {
         writes: Vec<Vec<u8>>,
     }
     impl OwnedAsyncWriter for RecorderWriter {
-        async fn write_all<Buf: IoBuf + Send>(
+        async fn write_all<B: BoundedBuf<Buf = Buf>, Buf: IoBuf + Send>(
             &mut self,
-            buf: FullSlice<Buf>,
+            buf: B,
             _: &RequestContext,
-        ) -> std::io::Result<(usize, FullSlice<Buf>)> {
+        ) -> std::io::Result<(usize, B::Buf)> {
+            let nbytes = buf.bytes_init();
+            if nbytes == 0 {
+                self.writes.push(vec![]);
+                return Ok((0, Slice::into_inner(buf.slice_full())));
+            }
+            let buf = buf.slice(0..nbytes);
             self.writes.push(Vec::from(&buf[..]));
-            Ok((buf.len(), buf))
+            Ok((buf.len(), Slice::into_inner(buf)))
         }
     }
 
@@ -259,7 +264,7 @@ mod tests {
     macro_rules! write {
         ($writer:ident, $data:literal) => {{
             $writer
-                .write_buffered(::bytes::Bytes::from_static($data).slice_len(), &test_ctx())
+                .write_buffered(::bytes::Bytes::from_static($data).slice_full(), &test_ctx())
                 .await?;
         }};
     }

pageserver/src/walingest.rs

@@ -515,7 +515,7 @@ impl WalIngest {
             && (decoded.xl_info == pg_constants::XLOG_FPI
                 || decoded.xl_info == pg_constants::XLOG_FPI_FOR_HINT)
             // compression of WAL is not yet supported: fall back to storing the original WAL record
-            && !postgres_ffi::bkpimage_is_compressed(blk.bimg_info, modification.tline.pg_version)
+            && !postgres_ffi::bkpimage_is_compressed(blk.bimg_info, modification.tline.pg_version)?
             // do not materialize null pages because them most likely be soon replaced with real data
             && blk.bimg_len != 0
         {
@@ -1702,7 +1702,7 @@ async fn get_relsize(
     modification: &DatadirModification<'_>,
     rel: RelTag,
     ctx: &RequestContext,
-) -> Result<BlockNumber, PageReconstructError> {
+) -> anyhow::Result<BlockNumber> {
     let nblocks = if !modification
         .tline
         .get_rel_exists(rel, Version::Modified(modification), ctx)

pageserver/src/walrecord.rs

@@ -1018,7 +1018,7 @@ pub fn decode_wal_record(
                     );
 
                     let blk_img_is_compressed =
-                        postgres_ffi::bkpimage_is_compressed(blk.bimg_info, pg_version);
+                        postgres_ffi::bkpimage_is_compressed(blk.bimg_info, pg_version)?;
 
                     if blk_img_is_compressed {
                         debug!("compressed block image , pg_version = {}", pg_version);

proxy/Cargo.toml

@@ -11,7 +11,6 @@ testing = []
 [dependencies]
 ahash.workspace = true
 anyhow.workspace = true
-arc-swap.workspace = true
 async-compression.workspace = true
 async-trait.workspace = true
 atomic-take.workspace = true
@@ -74,7 +73,7 @@ rustls.workspace = true
 scopeguard.workspace = true
 serde.workspace = true
 serde_json.workspace = true
-sha2 = { workspace = true, features = ["asm", "oid"] }
+sha2 = { workspace = true, features = ["asm"] }
 smol_str.workspace = true
 smallvec.workspace = true
 socket2.workspace = true
@@ -104,14 +103,6 @@ x509-parser.workspace = true
 postgres-protocol.workspace = true
 redis.workspace = true
 
-# jwt stuff
-jose-jwa = "0.1.2"
-jose-jwk = { version = "0.1.2", features = ["p256", "p384", "rsa"] }
-signature = "2"
-ecdsa = "0.16"
-p256 = "0.13"
-rsa = "0.9"
-
 workspace_hack.workspace = true
 
 [dev-dependencies]

proxy/src/auth/backend.rs

@@ -1,6 +1,5 @@
 mod classic;
 mod hacks;
-pub mod jwt;
 mod link;
 
 use std::net::IpAddr;

proxy/src/auth/backend/jwt.rs

@@ -1,556 +0,0 @@
-use std::{future::Future, sync::Arc, time::Duration};
-
-use anyhow::{bail, ensure, Context};
-use arc_swap::ArcSwapOption;
-use dashmap::DashMap;
-use jose_jwk::crypto::KeyInfo;
-use signature::Verifier;
-use tokio::time::Instant;
-
-use crate::{http::parse_json_body_with_limit, intern::EndpointIdInt};
-
-// TODO(conrad): make these configurable.
-const MIN_RENEW: Duration = Duration::from_secs(30);
-const AUTO_RENEW: Duration = Duration::from_secs(300);
-const MAX_RENEW: Duration = Duration::from_secs(3600);
-const MAX_JWK_BODY_SIZE: usize = 64 * 1024;
-
-/// How to get the JWT auth rules
-pub trait FetchAuthRules: Clone + Send + Sync + 'static {
-    fn fetch_auth_rules(&self) -> impl Future<Output = anyhow::Result<AuthRules>> + Send;
-}
-
-#[derive(Clone)]
-struct FetchAuthRulesFromCplane {
-    #[allow(dead_code)]
-    endpoint: EndpointIdInt,
-}
-
-impl FetchAuthRules for FetchAuthRulesFromCplane {
-    async fn fetch_auth_rules(&self) -> anyhow::Result<AuthRules> {
-        Err(anyhow::anyhow!("not yet implemented"))
-    }
-}
-
-pub struct AuthRules {
-    jwks_urls: Vec<url::Url>,
-}
-
-#[derive(Default)]
-pub struct JwkCache {
-    client: reqwest::Client,
-
-    map: DashMap<EndpointIdInt, Arc<JwkCacheEntryLock>>,
-}
-
-pub struct JwkCacheEntryLock {
-    cached: ArcSwapOption<JwkCacheEntry>,
-    lookup: tokio::sync::Semaphore,
-}
-
-impl Default for JwkCacheEntryLock {
-    fn default() -> Self {
-        JwkCacheEntryLock {
-            cached: ArcSwapOption::empty(),
-            lookup: tokio::sync::Semaphore::new(1),
-        }
-    }
-}
-
-pub struct JwkCacheEntry {
-    /// Should refetch at least every hour to verify when old keys have been removed.
-    /// Should refetch when new key IDs are seen only every 5 minutes or so
-    last_retrieved: Instant,
-
-    /// cplane will return multiple JWKs urls that we need to scrape.
-    key_sets: ahash::HashMap<url::Url, jose_jwk::JwkSet>,
-}
-
-impl JwkCacheEntryLock {
-    async fn acquire_permit<'a>(self: &'a Arc<Self>) -> JwkRenewalPermit<'a> {
-        JwkRenewalPermit::acquire_permit(self).await
-    }
-
-    fn try_acquire_permit<'a>(self: &'a Arc<Self>) -> Option<JwkRenewalPermit<'a>> {
-        JwkRenewalPermit::try_acquire_permit(self)
-    }
-
-    async fn renew_jwks<F: FetchAuthRules>(
-        &self,
-        _permit: JwkRenewalPermit<'_>,
-        client: &reqwest::Client,
-        auth_rules: &F,
-    ) -> anyhow::Result<Arc<JwkCacheEntry>> {
-        // double check that no one beat us to updating the cache.
-        let now = Instant::now();
-        let guard = self.cached.load_full();
-        if let Some(cached) = guard {
-            let last_update = now.duration_since(cached.last_retrieved);
-            if last_update < Duration::from_secs(300) {
-                return Ok(cached);
-            }
-        }
-
-        let rules = auth_rules.fetch_auth_rules().await?;
-        let mut key_sets = ahash::HashMap::with_capacity_and_hasher(
-            rules.jwks_urls.len(),
-            ahash::RandomState::new(),
-        );
-        // TODO(conrad): run concurrently
-        // TODO(conrad): strip the JWKs urls (should be checked by cplane as well - cloud#16284)
-        for url in rules.jwks_urls {
-            let req = client.get(url.clone());
-            // TODO(conrad): eventually switch to using reqwest_middleware/`new_client_with_timeout`.
-            match req.send().await.and_then(|r| r.error_for_status()) {
-                // todo: should we re-insert JWKs if we want to keep this JWKs URL?
-                // I expect these failures would be quite sparse.
-                Err(e) => tracing::warn!(?url, error=?e, "could not fetch JWKs"),
-                Ok(r) => {
-                    let resp: http::Response<reqwest::Body> = r.into();
-                    match parse_json_body_with_limit::<jose_jwk::JwkSet>(
-                        resp.into_body(),
-                        MAX_JWK_BODY_SIZE,
-                    )
-                    .await
-                    {
-                        Err(e) => tracing::warn!(?url, error=?e, "could not decode JWKs"),
-                        Ok(jwks) => {
-                            key_sets.insert(url, jwks);
-                        }
-                    }
-                }
-            }
-        }
-
-        let entry = Arc::new(JwkCacheEntry {
-            last_retrieved: now,
-            key_sets,
-        });
-        self.cached.swap(Some(Arc::clone(&entry)));
-
-        Ok(entry)
-    }
-
-    async fn get_or_update_jwk_cache<F: FetchAuthRules>(
-        self: &Arc<Self>,
-        client: &reqwest::Client,
-        fetch: &F,
-    ) -> Result<Arc<JwkCacheEntry>, anyhow::Error> {
-        let now = Instant::now();
-        let guard = self.cached.load_full();
-
-        // if we have no cached JWKs, try and get some
-        let Some(cached) = guard else {
-            let permit = self.acquire_permit().await;
-            return self.renew_jwks(permit, client, fetch).await;
-        };
-
-        let last_update = now.duration_since(cached.last_retrieved);
-
-        // check if the cached JWKs need updating.
-        if last_update > MAX_RENEW {
-            let permit = self.acquire_permit().await;
-
-            // it's been too long since we checked the keys. wait for them to update.
-            return self.renew_jwks(permit, client, fetch).await;
-        }
-
-        // every 5 minutes we should spawn a job to eagerly update the token.
-        if last_update > AUTO_RENEW {
-            if let Some(permit) = self.try_acquire_permit() {
-                tracing::debug!("JWKs should be renewed. Renewal permit acquired");
-                let permit = permit.into_owned();
-                let entry = self.clone();
-                let client = client.clone();
-                let fetch = fetch.clone();
-                tokio::spawn(async move {
-                    if let Err(e) = entry.renew_jwks(permit, &client, &fetch).await {
-                        tracing::warn!(error=?e, "could not fetch JWKs in background job");
-                    }
-                });
-            } else {
-                tracing::debug!("JWKs should be renewed. Renewal permit already taken, skipping");
-            }
-        }
-
-        Ok(cached)
-    }
-
-    async fn check_jwt<F: FetchAuthRules>(
-        self: &Arc<Self>,
-        jwt: String,
-        client: &reqwest::Client,
-        fetch: &F,
-    ) -> Result<(), anyhow::Error> {
-        // JWT compact form is defined to be
-        // <B64(Header)> || . || <B64(Payload)> || . || <B64(Signature)>
-        // where Signature = alg(<B64(Header)> || . || <B64(Payload)>);
-
-        let (header_payload, signature) = jwt
-            .rsplit_once(".")
-            .context("Provided authentication token is not a valid JWT encoding")?;
-        let (header, _payload) = header_payload
-            .split_once(".")
-            .context("Provided authentication token is not a valid JWT encoding")?;
-
-        let header = base64::decode_config(header, base64::URL_SAFE_NO_PAD)
-            .context("Provided authentication token is not a valid JWT encoding")?;
-        let header = serde_json::from_slice::<JWTHeader>(&header)
-            .context("Provided authentication token is not a valid JWT encoding")?;
-
-        let sig = base64::decode_config(signature, base64::URL_SAFE_NO_PAD)
-            .context("Provided authentication token is not a valid JWT encoding")?;
-
-        ensure!(header.typ == "JWT");
-        let kid = header.kid.context("missing key id")?;
-
-        let mut guard = self.get_or_update_jwk_cache(client, fetch).await?;
-
-        // get the key from the JWKs if possible. If not, wait for the keys to update.
-        let jwk = loop {
-            let jwk = guard
-                .key_sets
-                .values()
-                .flat_map(|jwks| &jwks.keys)
-                .find(|jwk| jwk.prm.kid.as_deref() == Some(kid));
-
-            match jwk {
-                Some(jwk) => break jwk,
-                None if guard.last_retrieved.elapsed() > MIN_RENEW => {
-                    let permit = self.acquire_permit().await;
-                    guard = self.renew_jwks(permit, client, fetch).await?;
-                }
-                _ => {
-                    bail!("jwk not found");
-                }
-            }
-        };
-
-        ensure!(
-            jwk.is_supported(&header.alg),
-            "signature algorithm not supported"
-        );
-
-        match &jwk.key {
-            jose_jwk::Key::Ec(key) => {
-                verify_ec_signature(header_payload.as_bytes(), &sig, key)?;
-            }
-            jose_jwk::Key::Rsa(key) => {
-                verify_rsa_signature(header_payload.as_bytes(), &sig, key, &jwk.prm.alg)?;
-            }
-            key => bail!("unsupported key type {key:?}"),
-        };
-
-        // TODO(conrad): verify iss, exp, nbf, etc...
-
-        Ok(())
-    }
-}
-
-impl JwkCache {
-    pub async fn check_jwt(
-        &self,
-        endpoint: EndpointIdInt,
-        jwt: String,
-    ) -> Result<(), anyhow::Error> {
-        // try with just a read lock first
-        let entry = self.map.get(&endpoint).as_deref().map(Arc::clone);
-        let entry = match entry {
-            Some(entry) => entry,
-            None => {
-                // acquire a write lock after to insert.
-                let entry = self.map.entry(endpoint).or_default();
-                Arc::clone(&*entry)
-            }
-        };
-
-        let fetch = FetchAuthRulesFromCplane { endpoint };
-        entry.check_jwt(jwt, &self.client, &fetch).await
-    }
-}
-
-fn verify_ec_signature(data: &[u8], sig: &[u8], key: &jose_jwk::Ec) -> anyhow::Result<()> {
-    use ecdsa::Signature;
-    use signature::Verifier;
-
-    match key.crv {
-        jose_jwk::EcCurves::P256 => {
-            let pk =
-                p256::PublicKey::try_from(key).map_err(|_| anyhow::anyhow!("invalid P256 key"))?;
-            let key = p256::ecdsa::VerifyingKey::from(&pk);
-            let sig = Signature::from_slice(sig)?;
-            key.verify(data, &sig)?;
-        }
-        key => bail!("unsupported ec key type {key:?}"),
-    }
-
-    Ok(())
-}
-
-fn verify_rsa_signature(
-    data: &[u8],
-    sig: &[u8],
-    key: &jose_jwk::Rsa,
-    alg: &Option<jose_jwa::Algorithm>,
-) -> anyhow::Result<()> {
-    use jose_jwa::{Algorithm, Signing};
-    use rsa::{
-        pkcs1v15::{Signature, VerifyingKey},
-        RsaPublicKey,
-    };
-
-    let key = RsaPublicKey::try_from(key).map_err(|_| anyhow::anyhow!("invalid RSA key"))?;
-
-    match alg {
-        Some(Algorithm::Signing(Signing::Rs256)) => {
-            let key = VerifyingKey::<sha2::Sha256>::new(key);
-            let sig = Signature::try_from(sig)?;
-            key.verify(data, &sig)?;
-        }
-        _ => bail!("invalid RSA signing algorithm"),
-    };
-
-    Ok(())
-}
-
-/// <https://datatracker.ietf.org/doc/html/rfc7515#section-4.1>
-#[derive(serde::Deserialize, serde::Serialize)]
-struct JWTHeader<'a> {
-    /// must be "JWT"
-    typ: &'a str,
-    /// must be a supported alg
-    alg: jose_jwa::Algorithm,
-    /// key id, must be provided for our usecase
-    kid: Option<&'a str>,
-}
-
-struct JwkRenewalPermit<'a> {
-    inner: Option<JwkRenewalPermitInner<'a>>,
-}
-
-enum JwkRenewalPermitInner<'a> {
-    Owned(Arc<JwkCacheEntryLock>),
-    Borrowed(&'a Arc<JwkCacheEntryLock>),
-}
-
-impl JwkRenewalPermit<'_> {
-    fn into_owned(mut self) -> JwkRenewalPermit<'static> {
-        JwkRenewalPermit {
-            inner: self.inner.take().map(JwkRenewalPermitInner::into_owned),
-        }
-    }
-
-    async fn acquire_permit(from: &Arc<JwkCacheEntryLock>) -> JwkRenewalPermit {
-        match from.lookup.acquire().await {
-            Ok(permit) => {
-                permit.forget();
-                JwkRenewalPermit {
-                    inner: Some(JwkRenewalPermitInner::Borrowed(from)),
-                }
-            }
-            Err(_) => panic!("semaphore should not be closed"),
-        }
-    }
-
-    fn try_acquire_permit(from: &Arc<JwkCacheEntryLock>) -> Option<JwkRenewalPermit> {
-        match from.lookup.try_acquire() {
-            Ok(permit) => {
-                permit.forget();
-                Some(JwkRenewalPermit {
-                    inner: Some(JwkRenewalPermitInner::Borrowed(from)),
-                })
-            }
-            Err(tokio::sync::TryAcquireError::NoPermits) => None,
-            Err(tokio::sync::TryAcquireError::Closed) => panic!("semaphore should not be closed"),
-        }
-    }
-}
-
-impl JwkRenewalPermitInner<'_> {
-    fn into_owned(self) -> JwkRenewalPermitInner<'static> {
-        match self {
-            JwkRenewalPermitInner::Owned(p) => JwkRenewalPermitInner::Owned(p),
-            JwkRenewalPermitInner::Borrowed(p) => JwkRenewalPermitInner::Owned(Arc::clone(p)),
-        }
-    }
-}
-
-impl Drop for JwkRenewalPermit<'_> {
-    fn drop(&mut self) {
-        let entry = match &self.inner {
-            None => return,
-            Some(JwkRenewalPermitInner::Owned(p)) => p,
-            Some(JwkRenewalPermitInner::Borrowed(p)) => *p,
-        };
-        entry.lookup.add_permits(1);
-    }
-}
-
-#[cfg(test)]
-mod tests {
-    use super::*;
-
-    use std::{future::IntoFuture, net::SocketAddr, time::SystemTime};
-
-    use base64::URL_SAFE_NO_PAD;
-    use bytes::Bytes;
-    use http::Response;
-    use http_body_util::Full;
-    use hyper1::service::service_fn;
-    use hyper_util::rt::TokioIo;
-    use rand::rngs::OsRng;
-    use signature::Signer;
-    use tokio::net::TcpListener;
-
-    fn new_ec_jwk(kid: String) -> (p256::SecretKey, jose_jwk::Jwk) {
-        let sk = p256::SecretKey::random(&mut OsRng);
-        let pk = sk.public_key().into();
-        let jwk = jose_jwk::Jwk {
-            key: jose_jwk::Key::Ec(pk),
-            prm: jose_jwk::Parameters {
-                kid: Some(kid),
-                alg: Some(jose_jwa::Algorithm::Signing(jose_jwa::Signing::Es256)),
-                ..Default::default()
-            },
-        };
-        (sk, jwk)
-    }
-
-    fn new_rsa_jwk(kid: String) -> (rsa::RsaPrivateKey, jose_jwk::Jwk) {
-        let sk = rsa::RsaPrivateKey::new(&mut OsRng, 2048).unwrap();
-        let pk = sk.to_public_key().into();
-        let jwk = jose_jwk::Jwk {
-            key: jose_jwk::Key::Rsa(pk),
-            prm: jose_jwk::Parameters {
-                kid: Some(kid),
-                alg: Some(jose_jwa::Algorithm::Signing(jose_jwa::Signing::Rs256)),
-                ..Default::default()
-            },
-        };
-        (sk, jwk)
-    }
-
-    fn build_jwt_payload(kid: String, sig: jose_jwa::Signing) -> String {
-        let header = JWTHeader {
-            typ: "JWT",
-            alg: jose_jwa::Algorithm::Signing(sig),
-            kid: Some(&kid),
-        };
-        let body = typed_json::json! {{
-            "exp": SystemTime::now().duration_since(SystemTime::UNIX_EPOCH).unwrap().as_secs() + 3600,
-        }};
-
-        let header =
-            base64::encode_config(serde_json::to_string(&header).unwrap(), URL_SAFE_NO_PAD);
-        let body = base64::encode_config(body.to_string(), URL_SAFE_NO_PAD);
-
-        format!("{header}.{body}")
-    }
-
-    fn new_ec_jwt(kid: String, key: p256::SecretKey) -> String {
-        use p256::ecdsa::{Signature, SigningKey};
-
-        let payload = build_jwt_payload(kid, jose_jwa::Signing::Es256);
-        let sig: Signature = SigningKey::from(key).sign(payload.as_bytes());
-        let sig = base64::encode_config(sig.to_bytes(), URL_SAFE_NO_PAD);
-
-        format!("{payload}.{sig}")
-    }
-
-    fn new_rsa_jwt(kid: String, key: rsa::RsaPrivateKey) -> String {
-        use rsa::pkcs1v15::SigningKey;
-        use rsa::signature::SignatureEncoding;
-
-        let payload = build_jwt_payload(kid, jose_jwa::Signing::Rs256);
-        let sig = SigningKey::<sha2::Sha256>::new(key).sign(payload.as_bytes());
-        let sig = base64::encode_config(sig.to_bytes(), URL_SAFE_NO_PAD);
-
-        format!("{payload}.{sig}")
-    }
-
-    #[tokio::test]
-    async fn renew() {
-        let (rs1, jwk1) = new_rsa_jwk("1".into());
-        let (rs2, jwk2) = new_rsa_jwk("2".into());
-        let (ec1, jwk3) = new_ec_jwk("3".into());
-        let (ec2, jwk4) = new_ec_jwk("4".into());
-
-        let jwt1 = new_rsa_jwt("1".into(), rs1);
-        let jwt2 = new_rsa_jwt("2".into(), rs2);
-        let jwt3 = new_ec_jwt("3".into(), ec1);
-        let jwt4 = new_ec_jwt("4".into(), ec2);
-
-        let foo_jwks = jose_jwk::JwkSet {
-            keys: vec![jwk1, jwk3],
-        };
-        let bar_jwks = jose_jwk::JwkSet {
-            keys: vec![jwk2, jwk4],
-        };
-
-        let service = service_fn(move |req| {
-            let foo_jwks = foo_jwks.clone();
-            let bar_jwks = bar_jwks.clone();
-            async move {
-                let jwks = match req.uri().path() {
-                    "/foo" => &foo_jwks,
-                    "/bar" => &bar_jwks,
-                    _ => {
-                        return Response::builder()
-                            .status(404)
-                            .body(Full::new(Bytes::new()));
-                    }
-                };
-                let body = serde_json::to_vec(jwks).unwrap();
-                Response::builder()
-                    .status(200)
-                    .body(Full::new(Bytes::from(body)))
-            }
-        });
-
-        let listener = TcpListener::bind("0.0.0.0:0").await.unwrap();
-        let server = hyper1::server::conn::http1::Builder::new();
-        let addr = listener.local_addr().unwrap();
-        tokio::spawn(async move {
-            loop {
-                let (s, _) = listener.accept().await.unwrap();
-                let serve = server.serve_connection(TokioIo::new(s), service.clone());
-                tokio::spawn(serve.into_future());
-            }
-        });
-
-        let client = reqwest::Client::new();
-
-        #[derive(Clone)]
-        struct Fetch(SocketAddr);
-
-        impl FetchAuthRules for Fetch {
-            async fn fetch_auth_rules(&self) -> anyhow::Result<AuthRules> {
-                Ok(AuthRules {
-                    jwks_urls: vec![
-                        format!("http://{}/foo", self.0).parse().unwrap(),
-                        format!("http://{}/bar", self.0).parse().unwrap(),
-                    ],
-                })
-            }
-        }
-
-        let jwk_cache = Arc::new(JwkCacheEntryLock::default());
-
-        jwk_cache
-            .check_jwt(jwt1, &client, &Fetch(addr))
-            .await
-            .unwrap();
-        jwk_cache
-            .check_jwt(jwt2, &client, &Fetch(addr))
-            .await
-            .unwrap();
-        jwk_cache
-            .check_jwt(jwt3, &client, &Fetch(addr))
-            .await
-            .unwrap();
-        jwk_cache
-            .check_jwt(jwt4, &client, &Fetch(addr))
-            .await
-            .unwrap();
-    }
-}

proxy/src/http.rs

@@ -6,12 +6,6 @@ pub mod health_server;
 
 use std::time::Duration;
 
-use anyhow::bail;
-use bytes::Bytes;
-use http_body_util::BodyExt;
-use hyper1::body::Body;
-use serde::de::DeserializeOwned;
-
 pub use reqwest::{Request, Response, StatusCode};
 pub use reqwest_middleware::{ClientWithMiddleware, Error};
 pub use reqwest_retry::{policies::ExponentialBackoff, RetryTransientMiddleware};
@@ -102,33 +96,6 @@ impl Endpoint {
     }
 }
 
-pub async fn parse_json_body_with_limit<D: DeserializeOwned>(
-    mut b: impl Body<Data = Bytes, Error = reqwest::Error> + Unpin,
-    limit: usize,
-) -> anyhow::Result<D> {
-    // We could use `b.limited().collect().await.to_bytes()` here
-    // but this ends up being slightly more efficient as far as I can tell.
-
-    // check the lower bound of the size hint.
-    // in reqwest, this value is influenced by the Content-Length header.
-    let lower_bound = match usize::try_from(b.size_hint().lower()) {
-        Ok(bound) if bound <= limit => bound,
-        _ => bail!("Content length exceeds limit of {limit} bytes"),
-    };
-    let mut bytes = Vec::with_capacity(lower_bound);
-
-    while let Some(frame) = b.frame().await.transpose()? {
-        if let Ok(data) = frame.into_data() {
-            if bytes.len() + data.len() > limit {
-                bail!("Content length exceeds limit of {limit} bytes")
-            }
-            bytes.extend_from_slice(&data);
-        }
-    }
-
-    Ok(serde_json::from_slice::<D>(&bytes)?)
-}
-
 #[cfg(test)]
 mod tests {
     use super::*;

test_runner/fixtures/pageserver/many_tenants.py

@@ -65,7 +65,7 @@ def single_timeline(
             override_storage_controller_generation=True,
         )
 
-    with concurrent.futures.ThreadPoolExecutor(max_workers=22) as executor:
+    with concurrent.futures.ThreadPoolExecutor(max_workers=1) as executor:
         executor.map(attach, tenants)
 
     # Benchmarks will start the pageserver explicitly themselves

workspace_hack/Cargo.toml

@@ -30,17 +30,13 @@ chrono = { version = "0.4", default-features = false, features = ["clock", "serd
 clap = { version = "4", features = ["derive", "string"] }
 clap_builder = { version = "4", default-features = false, features = ["color", "help", "std", "string", "suggestions", "usage"] }
 crossbeam-utils = { version = "0.8" }
-crypto-bigint = { version = "0.5", features = ["generic-array", "zeroize"] }
-der = { version = "0.7", default-features = false, features = ["oid", "pem", "std"] }
 deranged = { version = "0.3", default-features = false, features = ["powerfmt", "serde", "std"] }
-digest = { version = "0.10", features = ["mac", "oid", "std"] }
 either = { version = "1" }
 fail = { version = "0.5", default-features = false, features = ["failpoints"] }
 futures-channel = { version = "0.3", features = ["sink"] }
 futures-executor = { version = "0.3" }
 futures-io = { version = "0.3" }
 futures-util = { version = "0.3", features = ["channel", "io", "sink"] }
-generic-array = { version = "0.14", default-features = false, features = ["more_lengths", "zeroize"] }
 getrandom = { version = "0.2", default-features = false, features = ["std"] }
 hashbrown = { version = "0.14", features = ["raw"] }
 hex = { version = "0.4", features = ["serde"] }
@@ -48,7 +44,6 @@ hmac = { version = "0.12", default-features = false, features = ["reset"] }
 hyper = { version = "0.14", features = ["full"] }
 indexmap = { version = "1", default-features = false, features = ["std"] }
 itertools = { version = "0.10" }
-lazy_static = { version = "1", default-features = false, features = ["spin_no_std"] }
 libc = { version = "0.2", features = ["extra_traits", "use_std"] }
 log = { version = "0.4", default-features = false, features = ["std"] }
 memchr = { version = "2" }
@@ -69,10 +64,8 @@ rustls = { version = "0.21", features = ["dangerous_configuration"] }
 scopeguard = { version = "1" }
 serde = { version = "1", features = ["alloc", "derive"] }
 serde_json = { version = "1", features = ["raw_value"] }
-sha2 = { version = "0.10", features = ["asm", "oid"] }
-signature = { version = "2", default-features = false, features = ["digest", "rand_core", "std"] }
+sha2 = { version = "0.10", features = ["asm"] }
 smallvec = { version = "1", default-features = false, features = ["const_new", "write"] }
-spki = { version = "0.7", default-features = false, features = ["pem", "std"] }
 subtle = { version = "2" }
 sync_wrapper = { version = "0.1", default-features = false, features = ["futures"] }
 tikv-jemalloc-sys = { version = "0.5" }
@@ -88,7 +81,7 @@ tracing = { version = "0.1", features = ["log"] }
 tracing-core = { version = "0.1" }
 url = { version = "2", features = ["serde"] }
 uuid = { version = "1", features = ["serde", "v4", "v7"] }
-zeroize = { version = "1", features = ["derive", "serde"] }
+zeroize = { version = "1", features = ["derive"] }
 zstd = { version = "0.13" }
 zstd-safe = { version = "7", default-features = false, features = ["arrays", "legacy", "std", "zdict_builder"] }
 zstd-sys = { version = "2", default-features = false, features = ["legacy", "std", "zdict_builder"] }
@@ -104,7 +97,6 @@ getrandom = { version = "0.2", default-features = false, features = ["std"] }
 hashbrown = { version = "0.14", features = ["raw"] }
 indexmap = { version = "1", default-features = false, features = ["std"] }
 itertools = { version = "0.10" }
-lazy_static = { version = "1", default-features = false, features = ["spin_no_std"] }
 libc = { version = "0.2", features = ["extra_traits", "use_std"] }
 log = { version = "0.4", default-features = false, features = ["std"] }
 memchr = { version = "2" }