hacky way to make tests pass

DNM: test-log
storage controller: fix a disagreement between schedule + optimize
2026-07-08 14:40:37 +00:00 · 2024-10-04 09:56:56 +00:00 · 2024-10-03 11:29:06 +00:00 · 2024-10-03 11:14:34 +00:00
412 changed files with 6373 additions and 11379 deletions
--- a/.dockerignore
+++ b/.dockerignore
@@ -5,7 +5,9 @@
 !Cargo.toml
 !Makefile
 !rust-toolchain.toml
+!scripts/combine_control_files.py
 !scripts/ninstall.sh
+!vm-cgconfig.conf
 !docker-compose/run-tests.sh

 # Directories
@@ -15,12 +17,15 @@
 !compute_tools/
 !control_plane/
 !libs/
+!neon_local/
 !pageserver/
+!patches/
 !pgxn/
 !proxy/
 !storage_scrubber/
 !safekeeper/
 !storage_broker/
 !storage_controller/
+!trace/
 !vendor/postgres-*/
 !workspace_hack/
--- a/.github/actions/allure-report-generate/action.yml
+++ b/.github/actions/allure-report-generate/action.yml
@@ -183,7 +183,7 @@ runs:
      uses: actions/cache@v4
      with:
        path: ~/.cache/pypoetry/virtualenvs
-        key: v2-${{ runner.os }}-${{ runner.arch }}-python-deps-bookworm-${{ hashFiles('poetry.lock') }}
+        key: v2-${{ runner.os }}-${{ runner.arch }}-python-deps-${{ hashFiles('poetry.lock') }}

    - name: Store Allure test stat in the DB (new)
      if: ${{ !cancelled() && inputs.store-test-results-into-db == 'true' }}
--- a/.github/actions/run-python-test-set/action.yml
+++ b/.github/actions/run-python-test-set/action.yml
@@ -88,7 +88,7 @@ runs:
      uses: actions/cache@v4
      with:
        path: ~/.cache/pypoetry/virtualenvs
-        key: v2-${{ runner.os }}-${{ runner.arch }}-python-deps-bookworm-${{ hashFiles('poetry.lock') }}
+        key: v2-${{ runner.os }}-${{ runner.arch }}-python-deps-${{ hashFiles('poetry.lock') }}

    - name: Install Python deps
      shell: bash -euxo pipefail {0}
@@ -218,9 +218,6 @@ runs:
        name: compatibility-snapshot-${{ runner.arch }}-${{ inputs.build_type }}-pg${{ inputs.pg_version }}
        # Directory is created by test_compatibility.py::test_create_snapshot, keep the path in sync with the test
        path: /tmp/test_output/compatibility_snapshot_pg${{ inputs.pg_version }}/
-        # The lack of compatibility snapshot shouldn't fail the job
-        # (for example if we didn't run the test for non build-and-test workflow)
-        skip-if-does-not-exist: true

    - name: Upload test results
      if: ${{ !cancelled() }}
--- a/.github/actions/upload/action.yml
+++ b/.github/actions/upload/action.yml
@@ -7,10 +7,6 @@ inputs:
  path:
    description: "A directory or file to upload"
    required: true
-  skip-if-does-not-exist:
-    description: "Allow to skip if path doesn't exist, fail otherwise"
-    default: false
-    required: false
  prefix:
    description: "S3 prefix. Default is '${GITHUB_SHA}/${GITHUB_RUN_ID}/${GITHUB_RUN_ATTEMPT}'"
    required: false
@@ -19,12 +15,10 @@ runs:
  using: "composite"
  steps:
    - name: Prepare artifact
-      id: prepare-artifact
      shell: bash -euxo pipefail {0}
      env:
        SOURCE: ${{ inputs.path }}
        ARCHIVE: /tmp/uploads/${{ inputs.name }}.tar.zst
-        SKIP_IF_DOES_NOT_EXIST: ${{ inputs.skip-if-does-not-exist }}
      run: |
        mkdir -p $(dirname $ARCHIVE)

@@ -39,22 +33,14 @@ runs:
        elif [ -f ${SOURCE} ]; then
          time tar -cf ${ARCHIVE} --zstd ${SOURCE}
        elif ! ls ${SOURCE} > /dev/null 2>&1; then
-          if [ "${SKIP_IF_DOES_NOT_EXIST}" = "true" ]; then
-            echo 'SKIPPED=true' >> $GITHUB_OUTPUT
-            exit 0
-          else
-            echo >&2 "${SOURCE} does not exist"
-            exit 2
-          fi
+          echo >&2 "${SOURCE} does not exist"
+          exit 2
        else
          echo >&2 "${SOURCE} is neither a directory nor a file, do not know how to handle it"
          exit 3
        fi

-        echo 'SKIPPED=false' >> $GITHUB_OUTPUT
-
    - name: Upload artifact
-      if: ${{ steps.prepare-artifact.outputs.SKIPPED == 'false' }}
      shell: bash -euxo pipefail {0}
      env:
        SOURCE: ${{ inputs.path }}
--- a/.github/workflows/_build-and-test-locally.yml
+++ b/.github/workflows/_build-and-test-locally.yml
@@ -124,28 +124,28 @@ jobs:
        uses: actions/cache@v4
        with:
          path: pg_install/v14
-          key: v1-${{ runner.os }}-${{ runner.arch }}-${{ inputs.build-type }}-pg-${{ steps.pg_v14_rev.outputs.pg_rev }}-bookworm-${{ hashFiles('Makefile', 'Dockerfile.build-tools') }}
+          key: v1-${{ runner.os }}-${{ runner.arch }}-${{ inputs.build-type }}-pg-${{ steps.pg_v14_rev.outputs.pg_rev }}-${{ hashFiles('Makefile', 'Dockerfile.build-tools') }}

      - name: Cache postgres v15 build
        id: cache_pg_15
        uses: actions/cache@v4
        with:
          path: pg_install/v15
-          key: v1-${{ runner.os }}-${{ runner.arch }}-${{ inputs.build-type }}-pg-${{ steps.pg_v15_rev.outputs.pg_rev }}-bookworm-${{ hashFiles('Makefile', 'Dockerfile.build-tools') }}
+          key: v1-${{ runner.os }}-${{ runner.arch }}-${{ inputs.build-type }}-pg-${{ steps.pg_v15_rev.outputs.pg_rev }}-${{ hashFiles('Makefile', 'Dockerfile.build-tools') }}

      - name: Cache postgres v16 build
        id: cache_pg_16
        uses: actions/cache@v4
        with:
          path: pg_install/v16
-          key: v1-${{ runner.os }}-${{ runner.arch }}-${{ inputs.build-type }}-pg-${{ steps.pg_v16_rev.outputs.pg_rev }}-bookworm-${{ hashFiles('Makefile', 'Dockerfile.build-tools') }}
+          key: v1-${{ runner.os }}-${{ runner.arch }}-${{ inputs.build-type }}-pg-${{ steps.pg_v16_rev.outputs.pg_rev }}-${{ hashFiles('Makefile', 'Dockerfile.build-tools') }}

      - name: Cache postgres v17 build
        id: cache_pg_17
        uses: actions/cache@v4
        with:
          path: pg_install/v17
-          key: v1-${{ runner.os }}-${{ runner.arch }}-${{ inputs.build-type }}-pg-${{ steps.pg_v17_rev.outputs.pg_rev }}-bookworm-${{ hashFiles('Makefile', 'Dockerfile.build-tools') }}
+          key: v1-${{ runner.os }}-${{ runner.arch }}-${{ inputs.build-type }}-pg-${{ steps.pg_v17_rev.outputs.pg_rev }}-${{ hashFiles('Makefile', 'Dockerfile.build-tools') }}

      - name: Build postgres v14
        if: steps.cache_pg_14.outputs.cache-hit != 'true'
--- a/.github/workflows/build-build-tools-image.yml
+++ b/.github/workflows/build-build-tools-image.yml
@@ -19,16 +19,9 @@ defaults:
  run:
    shell: bash -euo pipefail {0}

-# The initial idea was to prevent the waste of resources by not re-building the `build-tools` image
-# for the same tag in parallel workflow runs, and queue them to be skipped once we have
-# the first image pushed to Docker registry, but GitHub's concurrency mechanism is not working as expected.
-# GitHub can't have more than 1 job in a queue and removes the previous one, it causes failures if the dependent jobs.
-#
-# Ref https://github.com/orgs/community/discussions/41518
-#
-# concurrency:
-#   group: build-build-tools-image-${{ inputs.image-tag }}
-#   cancel-in-progress: false
+concurrency:
+  group: build-build-tools-image-${{ inputs.image-tag }}
+  cancel-in-progress: false

 # No permission for GITHUB_TOKEN by default; the **minimal required** set of permissions should be granted in each job.
 permissions: {}
@@ -43,7 +36,6 @@ jobs:

    strategy:
      matrix:
-        debian-version: [ bullseye, bookworm ]
        arch: [ x64, arm64 ]

    runs-on: ${{ fromJson(format('["self-hosted", "{0}"]', matrix.arch == 'arm64' && 'large-arm64' || 'large')) }}
@@ -82,22 +74,22 @@ jobs:

      - uses: docker/build-push-action@v6
        with:
-          file: Dockerfile.build-tools
          context: .
          provenance: false
          push: true
          pull: true
-          build-args: |
-            DEBIAN_VERSION=${{ matrix.debian-version }}
-          cache-from: type=registry,ref=cache.neon.build/build-tools:cache-${{ matrix.debian-version }}-${{ matrix.arch }}
-          cache-to: ${{ github.ref_name == 'main' && format('type=registry,ref=cache.neon.build/build-tools:cache-{0}-{1},mode=max', matrix.debian-version, matrix.arch) || '' }}
-          tags: |
-            neondatabase/build-tools:${{ inputs.image-tag }}-${{ matrix.debian-version }}-${{ matrix.arch }}
+          file: Dockerfile.build-tools
+          cache-from: type=registry,ref=cache.neon.build/build-tools:cache-${{ matrix.arch }}
+          cache-to: ${{ github.ref_name == 'main' && format('type=registry,ref=cache.neon.build/build-tools:cache-{0},mode=max', matrix.arch) || '' }}
+          tags: neondatabase/build-tools:${{ inputs.image-tag }}-${{ matrix.arch }}

  merge-images:
    needs: [ build-image ]
    runs-on: ubuntu-22.04

+    env:
+      IMAGE_TAG: ${{ inputs.image-tag }}
+
    steps:
      - uses: docker/login-action@v3
        with:
@@ -105,17 +97,7 @@ jobs:
          password: ${{ secrets.NEON_DOCKERHUB_PASSWORD }}

      - name: Create multi-arch image
-        env:
-          DEFAULT_DEBIAN_VERSION: bullseye
-          IMAGE_TAG: ${{ inputs.image-tag }}
        run: |
-          for debian_version in bullseye bookworm; do
-            tags=("-t" "neondatabase/build-tools:${IMAGE_TAG}-${debian_version}")
-            if [ "${debian_version}" == "${DEFAULT_DEBIAN_VERSION}" ]; then
-              tags+=("-t" "neondatabase/build-tools:${IMAGE_TAG}")
-            fi
-
-            docker buildx imagetools create "${tags[@]}" \
-                                              neondatabase/build-tools:${IMAGE_TAG}-${debian_version}-x64 \
-                                              neondatabase/build-tools:${IMAGE_TAG}-${debian_version}-arm64
-          done
+          docker buildx imagetools create -t neondatabase/build-tools:${IMAGE_TAG} \
+                                             neondatabase/build-tools:${IMAGE_TAG}-x64 \
+                                             neondatabase/build-tools:${IMAGE_TAG}-arm64
--- a/.github/workflows/build_and_test.yml
+++ b/.github/workflows/build_and_test.yml
@@ -92,7 +92,7 @@ jobs:
    needs: [ check-permissions, build-build-tools-image ]
    runs-on: [ self-hosted, small ]
    container:
-      image: ${{ needs.build-build-tools-image.outputs.image }}-bookworm
+      image: ${{ needs.build-build-tools-image.outputs.image }}
      credentials:
        username: ${{ secrets.NEON_DOCKERHUB_USERNAME }}
        password: ${{ secrets.NEON_DOCKERHUB_PASSWORD }}
@@ -106,7 +106,7 @@ jobs:
        uses: actions/cache@v4
        with:
          path: ~/.cache/pypoetry/virtualenvs
-          key: v2-${{ runner.os }}-${{ runner.arch }}-python-deps-bookworm-${{ hashFiles('poetry.lock') }}
+          key: v2-${{ runner.os }}-${{ runner.arch }}-python-deps-${{ hashFiles('poetry.lock') }}

      - name: Install Python deps
        run: ./scripts/pysync
@@ -181,7 +181,7 @@ jobs:
    runs-on: ${{ fromJson(format('["self-hosted", "{0}"]', matrix.arch == 'arm64' && 'small-arm64' || 'small')) }}

    container:
-      image: ${{ needs.build-build-tools-image.outputs.image }}-bookworm
+      image: ${{ needs.build-build-tools-image.outputs.image }}
      credentials:
        username: ${{ secrets.NEON_DOCKERHUB_USERNAME }}
        password: ${{ secrets.NEON_DOCKERHUB_PASSWORD }}
@@ -193,15 +193,16 @@ jobs:
        with:
          submodules: true

-      - name: Cache cargo deps
-        uses: actions/cache@v4
-        with:
-          path: |
-            ~/.cargo/registry
-            !~/.cargo/registry/src
-            ~/.cargo/git
-            target
-          key: v1-${{ runner.os }}-${{ runner.arch }}-cargo-${{ hashFiles('./Cargo.lock') }}-${{ hashFiles('./rust-toolchain.toml') }}-rust
+#      Disabled for now
+#      - name: Restore cargo deps cache
+#        id: cache_cargo
+#        uses: actions/cache@v4
+#        with:
+#          path: |
+#            !~/.cargo/registry/src
+#            ~/.cargo/git/
+#            target/
+#          key: v1-${{ runner.os }}-${{ runner.arch }}-cargo-clippy-${{ hashFiles('rust-toolchain.toml') }}-${{ hashFiles('Cargo.lock') }}

      # Some of our rust modules use FFI and need those to be checked
      - name: Get postgres headers
@@ -261,7 +262,7 @@ jobs:
    uses: ./.github/workflows/_build-and-test-locally.yml
    with:
      arch: ${{ matrix.arch }}
-      build-tools-image: ${{ needs.build-build-tools-image.outputs.image }}-bookworm
+      build-tools-image: ${{ needs.build-build-tools-image.outputs.image }}
      build-tag: ${{ needs.tag.outputs.build-tag }}
      build-type: ${{ matrix.build-type }}
      # Run tests on all Postgres versions in release builds and only on the latest version in debug builds
@@ -276,7 +277,7 @@ jobs:
    needs: [ check-permissions, build-build-tools-image ]
    runs-on: [ self-hosted, small ]
    container:
-      image: ${{ needs.build-build-tools-image.outputs.image }}-bookworm
+      image: ${{ needs.build-build-tools-image.outputs.image }}
      credentials:
        username: ${{ secrets.NEON_DOCKERHUB_USERNAME }}
        password: ${{ secrets.NEON_DOCKERHUB_PASSWORD }}
@@ -289,7 +290,7 @@ jobs:
        uses: actions/cache@v4
        with:
          path: ~/.cache/pypoetry/virtualenvs
-          key: v2-${{ runner.os }}-${{ runner.arch }}-python-deps-bookworm-${{ hashFiles('poetry.lock') }}
+          key: v1-${{ runner.os }}-${{ runner.arch }}-python-deps-${{ hashFiles('poetry.lock') }}

      - name: Install Python deps
        run: ./scripts/pysync
@@ -309,7 +310,7 @@ jobs:
    needs: [ check-permissions, build-and-test-locally, build-build-tools-image, get-benchmarks-durations ]
    runs-on: [ self-hosted, small ]
    container:
-      image: ${{ needs.build-build-tools-image.outputs.image }}-bookworm
+      image: ${{ needs.build-build-tools-image.outputs.image }}
      credentials:
        username: ${{ secrets.NEON_DOCKERHUB_USERNAME }}
        password: ${{ secrets.NEON_DOCKERHUB_PASSWORD }}
@@ -367,7 +368,7 @@ jobs:

    runs-on: [ self-hosted, small ]
    container:
-      image: ${{ needs.build-build-tools-image.outputs.image }}-bookworm
+      image: ${{ needs.build-build-tools-image.outputs.image }}
      credentials:
        username: ${{ secrets.NEON_DOCKERHUB_USERNAME }}
        password: ${{ secrets.NEON_DOCKERHUB_PASSWORD }}
@@ -415,7 +416,7 @@ jobs:
    needs: [ check-permissions, build-build-tools-image, build-and-test-locally ]
    runs-on: [ self-hosted, small ]
    container:
-      image: ${{ needs.build-build-tools-image.outputs.image }}-bookworm
+      image: ${{ needs.build-build-tools-image.outputs.image }}
      credentials:
        username: ${{ secrets.NEON_DOCKERHUB_USERNAME }}
        password: ${{ secrets.NEON_DOCKERHUB_PASSWORD }}
@@ -559,16 +560,15 @@ jobs:
            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 }}-bookworm
-            DEBIAN_VERSION=bookworm
+            TAG=${{ needs.build-build-tools-image.outputs.image-tag }}
          provenance: false
          push: true
          pull: true
          file: Dockerfile
-          cache-from: type=registry,ref=cache.neon.build/neon:cache-bookworm-${{ matrix.arch }}
-          cache-to: ${{ github.ref_name == 'main' && format('type=registry,ref=cache.neon.build/neon:cache-{0}-{1},mode=max', 'bookworm', matrix.arch) || '' }}
+          cache-from: type=registry,ref=cache.neon.build/neon:cache-${{ matrix.arch }}
+          cache-to: ${{ github.ref_name == 'main' && format('type=registry,ref=cache.neon.build/neon:cache-{0},mode=max', matrix.arch) || '' }}
          tags: |
-            neondatabase/neon:${{ needs.tag.outputs.build-tag }}-bookworm-${{ matrix.arch }}
+            neondatabase/neon:${{ needs.tag.outputs.build-tag }}-${{ matrix.arch }}

  neon-image:
    needs: [ neon-image-arch, tag ]
@@ -583,9 +583,8 @@ jobs:
      - name: Create multi-arch image
        run: |
          docker buildx imagetools create -t neondatabase/neon:${{ needs.tag.outputs.build-tag }} \
-                                          -t neondatabase/neon:${{ needs.tag.outputs.build-tag }}-bookworm \
-                                             neondatabase/neon:${{ needs.tag.outputs.build-tag }}-bookworm-x64 \
-                                             neondatabase/neon:${{ needs.tag.outputs.build-tag }}-bookworm-arm64
+                                             neondatabase/neon:${{ needs.tag.outputs.build-tag }}-x64 \
+                                             neondatabase/neon:${{ needs.tag.outputs.build-tag }}-arm64

      - uses: docker/login-action@v3
        with:
@@ -606,16 +605,17 @@ jobs:
        version:
          # Much data was already generated on old PG versions with bullseye's
          # libraries, the locales of which can cause data incompatibilities.
-          # However, new PG versions should be build on newer images,
-          # as that reduces the support burden of old and ancient distros.
+          # However, new PG versions should check if they can be built on newer
+          # images, as that reduces the support burden of old and ancient
+          # distros.
          - pg: v14
-            debian: bullseye
+            debian: bullseye-slim
          - pg: v15
-            debian: bullseye
+            debian: bullseye-slim
          - pg: v16
-            debian: bullseye
+            debian: bullseye-slim
          - pg: v17
-            debian: bookworm
+            debian: bookworm-slim
        arch: [ x64, arm64 ]

    runs-on: ${{ fromJson(format('["self-hosted", "{0}"]', matrix.arch == 'arm64' && 'large-arm64' || 'large')) }}
@@ -660,16 +660,16 @@ jobs:
            GIT_VERSION=${{ github.event.pull_request.head.sha || github.sha }}
            PG_VERSION=${{ matrix.version.pg }}
            BUILD_TAG=${{ needs.tag.outputs.build-tag }}
-            TAG=${{ needs.build-build-tools-image.outputs.image-tag }}-${{ matrix.version.debian }}
-            DEBIAN_VERSION=${{ matrix.version.debian }}
+            TAG=${{ needs.build-build-tools-image.outputs.image-tag }}
+            DEBIAN_FLAVOR=${{ matrix.version.debian }}
          provenance: false
          push: true
          pull: true
          file: compute/Dockerfile.compute-node
-          cache-from: type=registry,ref=cache.neon.build/compute-node-${{ matrix.version.pg }}:cache-${{ matrix.version.debian }}-${{ matrix.arch }}
-          cache-to: ${{ github.ref_name == 'main' && format('type=registry,ref=cache.neon.build/compute-node-{0}:cache-{1}-{2},mode=max', matrix.version.pg, matrix.version.debian, matrix.arch) || '' }}
+          cache-from: type=registry,ref=cache.neon.build/compute-node-${{ matrix.version.pg }}:cache-${{ matrix.arch }}
+          cache-to: ${{ github.ref_name == 'main' && format('type=registry,ref=cache.neon.build/compute-node-{0}:cache-{1},mode=max', matrix.version.pg, matrix.arch) || '' }}
          tags: |
-            neondatabase/compute-node-${{ matrix.version.pg }}:${{ needs.tag.outputs.build-tag }}-${{ matrix.version.debian }}-${{ matrix.arch }}
+            neondatabase/compute-node-${{ matrix.version.pg }}:${{ needs.tag.outputs.build-tag }}-${{ matrix.arch }}

      - name: Build neon extensions test image
        if: matrix.version.pg == 'v16'
@@ -680,17 +680,17 @@ jobs:
            GIT_VERSION=${{ github.event.pull_request.head.sha || github.sha }}
            PG_VERSION=${{ matrix.version.pg }}
            BUILD_TAG=${{ needs.tag.outputs.build-tag }}
-            TAG=${{ needs.build-build-tools-image.outputs.image-tag }}-${{ matrix.version.debian }}
-            DEBIAN_VERSION=${{ matrix.version.debian }}
+            TAG=${{ needs.build-build-tools-image.outputs.image-tag }}
+            DEBIAN_FLAVOR=${{ matrix.version.debian }}
          provenance: false
          push: true
          pull: true
          file: compute/Dockerfile.compute-node
          target: neon-pg-ext-test
-          cache-from: type=registry,ref=cache.neon.build/neon-test-extensions-${{ matrix.version.pg }}:cache-${{ matrix.version.debian }}-${{ matrix.arch }}
-          cache-to: ${{ github.ref_name == 'main' && format('type=registry,ref=cache.neon.build/neon-test-extensions-{0}:cache-{1}-{2},mode=max', matrix.version.pg, matrix.version.debian, matrix.arch) || '' }}
+          cache-from: type=registry,ref=cache.neon.build/neon-test-extensions-${{ matrix.version.pg }}:cache-${{ matrix.arch }}
+          cache-to: ${{ github.ref_name == 'main' && format('type=registry,ref=cache.neon.build/neon-test-extensions-{0}:cache-{1},mode=max', matrix.version.pg, matrix.arch) || '' }}
          tags: |
-            neondatabase/neon-test-extensions-${{ matrix.version.pg }}:${{needs.tag.outputs.build-tag}}-${{ matrix.version.debian }}-${{ matrix.arch }}
+            neondatabase/neon-test-extensions-${{ matrix.version.pg }}:${{needs.tag.outputs.build-tag}}-${{ matrix.arch }}

      - name: Build compute-tools image
        # compute-tools are Postgres independent, so build it only once
@@ -705,16 +705,14 @@ jobs:
          build-args: |
            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 }}-${{ matrix.version.debian }}
-            DEBIAN_VERSION=${{ matrix.version.debian }}
+            TAG=${{ needs.build-build-tools-image.outputs.image-tag }}
+            DEBIAN_FLAVOR=${{ matrix.version.debian }}
          provenance: false
          push: true
          pull: true
          file: compute/Dockerfile.compute-node
-          cache-from: type=registry,ref=cache.neon.build/neon-test-extensions-${{ matrix.version.pg }}:cache-${{ matrix.version.debian }}-${{ matrix.arch }}
-          cache-to: ${{ github.ref_name == 'main' && format('type=registry,ref=cache.neon.build/compute-tools-{0}:cache-{1}-{2},mode=max', matrix.version.pg, matrix.version.debian, matrix.arch) || '' }}
          tags: |
-            neondatabase/compute-tools:${{ needs.tag.outputs.build-tag }}-${{ matrix.version.debian }}-${{ matrix.arch }}
+            neondatabase/compute-tools:${{ needs.tag.outputs.build-tag }}-${{ matrix.arch }}

  compute-node-image:
    needs: [ compute-node-image-arch, tag ]
@@ -722,16 +720,7 @@ jobs:

    strategy:
      matrix:
-        version:
-          # see the comment for `compute-node-image-arch` job
-          - pg: v14
-            debian: bullseye
-          - pg: v15
-            debian: bullseye
-          - pg: v16
-            debian: bullseye
-          - pg: v17
-            debian: bookworm
+        version: [ v14, v15, v16, v17 ]

    steps:
      - uses: docker/login-action@v3
@@ -741,26 +730,23 @@ jobs:

      - name: Create multi-arch compute-node image
        run: |
-          docker buildx imagetools create -t neondatabase/compute-node-${{ matrix.version.pg }}:${{ needs.tag.outputs.build-tag }} \
-                                          -t neondatabase/compute-node-${{ matrix.version.pg }}:${{ needs.tag.outputs.build-tag }}-${{ matrix.version.debian }} \
-                                             neondatabase/compute-node-${{ matrix.version.pg }}:${{ needs.tag.outputs.build-tag }}-${{ matrix.version.debian }}-x64 \
-                                             neondatabase/compute-node-${{ matrix.version.pg }}:${{ needs.tag.outputs.build-tag }}-${{ matrix.version.debian }}-arm64
+          docker buildx imagetools create -t neondatabase/compute-node-${{ matrix.version }}:${{ needs.tag.outputs.build-tag }} \
+                                             neondatabase/compute-node-${{ matrix.version }}:${{ needs.tag.outputs.build-tag }}-x64 \
+                                             neondatabase/compute-node-${{ matrix.version }}:${{ needs.tag.outputs.build-tag }}-arm64

      - name: Create multi-arch neon-test-extensions image
-        if: matrix.version.pg == 'v16'
+        if: matrix.version == 'v16'
        run: |
-          docker buildx imagetools create -t neondatabase/neon-test-extensions-${{ matrix.version.pg }}:${{ needs.tag.outputs.build-tag }} \
-                                          -t neondatabase/neon-test-extensions-${{ matrix.version.pg }}:${{ needs.tag.outputs.build-tag }}-${{ matrix.version.debian }} \
-                                             neondatabase/neon-test-extensions-${{ matrix.version.pg }}:${{ needs.tag.outputs.build-tag }}-${{ matrix.version.debian }}-x64 \
-                                             neondatabase/neon-test-extensions-${{ matrix.version.pg }}:${{ needs.tag.outputs.build-tag }}-${{ matrix.version.debian }}-arm64
+          docker buildx imagetools create -t neondatabase/neon-test-extensions-${{ matrix.version }}:${{ needs.tag.outputs.build-tag }} \
+                                             neondatabase/neon-test-extensions-${{ matrix.version }}:${{ needs.tag.outputs.build-tag }}-x64 \
+                                             neondatabase/neon-test-extensions-${{ matrix.version }}:${{ needs.tag.outputs.build-tag }}-arm64

      - name: Create multi-arch compute-tools image
-        if: matrix.version.pg == 'v16'
+        if: matrix.version == 'v17'
        run: |
          docker buildx imagetools create -t neondatabase/compute-tools:${{ needs.tag.outputs.build-tag }} \
-                                          -t neondatabase/compute-tools:${{ needs.tag.outputs.build-tag }}-${{ matrix.version.debian }} \
-                                             neondatabase/compute-tools:${{ needs.tag.outputs.build-tag }}-${{ matrix.version.debian }}-x64 \
-                                             neondatabase/compute-tools:${{ needs.tag.outputs.build-tag }}-${{ matrix.version.debian }}-arm64
+                                             neondatabase/compute-tools:${{ needs.tag.outputs.build-tag }}-x64 \
+                                             neondatabase/compute-tools:${{ needs.tag.outputs.build-tag }}-arm64

      - uses: docker/login-action@v3
        with:
@@ -768,13 +754,13 @@ jobs:
          username: ${{ secrets.AWS_ACCESS_KEY_DEV }}
          password: ${{ secrets.AWS_SECRET_KEY_DEV }}

-      - name: Push multi-arch compute-node-${{ matrix.version.pg }} image to ECR
+      - name: Push multi-arch compute-node-${{ matrix.version }} image to ECR
        run: |
-          docker buildx imagetools create -t 369495373322.dkr.ecr.eu-central-1.amazonaws.com/compute-node-${{ matrix.version.pg }}:${{ needs.tag.outputs.build-tag }} \
-                                                                                neondatabase/compute-node-${{ matrix.version.pg }}:${{ needs.tag.outputs.build-tag }}
+          docker buildx imagetools create -t 369495373322.dkr.ecr.eu-central-1.amazonaws.com/compute-node-${{ matrix.version }}:${{ needs.tag.outputs.build-tag }} \
+                                                                                neondatabase/compute-node-${{ matrix.version }}:${{ needs.tag.outputs.build-tag }}

      - name: Push multi-arch compute-tools image to ECR
-        if: matrix.version.pg == 'v16'
+        if: matrix.version == 'v17'
        run: |
          docker buildx imagetools create -t 369495373322.dkr.ecr.eu-central-1.amazonaws.com/compute-tools:${{ needs.tag.outputs.build-tag }} \
                                                                                neondatabase/compute-tools:${{ needs.tag.outputs.build-tag }}
@@ -785,16 +771,7 @@ jobs:
    strategy:
      fail-fast: false
      matrix:
-        version:
-          # see the comment for `compute-node-image-arch` job
-          - pg: v14
-            debian: bullseye
-          - pg: v15
-            debian: bullseye
-          - pg: v16
-            debian: bullseye
-          - pg: v17
-            debian: bookworm
+        version: [ v14, v15, v16, v17 ]
    env:
      VM_BUILDER_VERSION: v0.35.0

@@ -816,18 +793,18 @@ jobs:
      # it won't have the proper authentication (written at v0.6.0)
      - name: Pulling compute-node image
        run: |
-          docker pull neondatabase/compute-node-${{ matrix.version.pg }}:${{ needs.tag.outputs.build-tag }}
+          docker pull neondatabase/compute-node-${{ matrix.version }}:${{ needs.tag.outputs.build-tag }}

      - name: Build vm image
        run: |
          ./vm-builder \
-            -spec=compute/vm-image-spec-${{ matrix.version.debian }}.yaml \
-            -src=neondatabase/compute-node-${{ matrix.version.pg }}:${{ needs.tag.outputs.build-tag }} \
-            -dst=neondatabase/vm-compute-node-${{ matrix.version.pg }}:${{ needs.tag.outputs.build-tag }}
+            -spec=compute/vm-image-spec.yaml \
+            -src=neondatabase/compute-node-${{ matrix.version }}:${{ needs.tag.outputs.build-tag }} \
+            -dst=neondatabase/vm-compute-node-${{ matrix.version }}:${{ needs.tag.outputs.build-tag }}

      - name: Pushing vm-compute-node image
        run: |
-          docker push neondatabase/vm-compute-node-${{ matrix.version.pg }}:${{ needs.tag.outputs.build-tag }}
+          docker push neondatabase/vm-compute-node-${{ matrix.version }}:${{ needs.tag.outputs.build-tag }}

  test-images:
    needs: [ check-permissions, tag, neon-image, compute-node-image ]
--- a/.github/workflows/neon_extra_builds.yml
+++ b/.github/workflows/neon_extra_builds.yml
@@ -155,7 +155,7 @@ jobs:
      github.ref_name == 'main'
    runs-on: [ self-hosted, large ]
    container:
-      image: ${{ needs.build-build-tools-image.outputs.image }}-bookworm
+      image: ${{ needs.build-build-tools-image.outputs.image }}
      credentials:
        username: ${{ secrets.NEON_DOCKERHUB_USERNAME }}
        password: ${{ secrets.NEON_DOCKERHUB_PASSWORD }}
--- a/.github/workflows/pg-clients.yml
+++ b/.github/workflows/pg-clients.yml
@@ -55,7 +55,7 @@ jobs:
    runs-on: ubuntu-22.04

    container:
-      image: ${{ needs.build-build-tools-image.outputs.image }}-bookworm
+      image: ${{ needs.build-build-tools-image.outputs.image }}
      credentials:
        username: ${{ secrets.NEON_DOCKERHUB_USERNAME }}
        password: ${{ secrets.NEON_DOCKERHUB_PASSWORD }}
@@ -150,7 +150,7 @@ jobs:
    runs-on: ubuntu-22.04

    container:
-      image: ${{ needs.build-build-tools-image.outputs.image }}-bookworm
+      image: ${{ needs.build-build-tools-image.outputs.image }}
      credentials:
        username: ${{ secrets.NEON_DOCKERHUB_USERNAME }}
        password: ${{ secrets.NEON_DOCKERHUB_PASSWORD }}
--- a/.github/workflows/pin-build-tools-image.yml
+++ b/.github/workflows/pin-build-tools-image.yml
@@ -71,6 +71,7 @@ jobs:

    steps:
      - uses: docker/login-action@v3
+
        with:
          username: ${{ secrets.NEON_DOCKERHUB_USERNAME }}
          password: ${{ secrets.NEON_DOCKERHUB_PASSWORD }}
@@ -93,22 +94,8 @@ jobs:
          az acr login --name=neoneastus2

      - name: Tag build-tools with `${{ env.TO_TAG }}` in Docker Hub, ECR, and ACR
-        env:
-          DEFAULT_DEBIAN_VERSION: bullseye
        run: |
-          for debian_version in bullseye bookworm; do
-            tags=()
-
-            tags+=("-t" "neondatabase/build-tools:${TO_TAG}-${debian_version}")
-            tags+=("-t" "369495373322.dkr.ecr.eu-central-1.amazonaws.com/build-tools:${TO_TAG}-${debian_version}")
-            tags+=("-t" "neoneastus2.azurecr.io/neondatabase/build-tools:${TO_TAG}-${debian_version}")
-
-            if [ "${debian_version}" == "${DEFAULT_DEBIAN_VERSION}" ]; then
-              tags+=("-t" "neondatabase/build-tools:${TO_TAG}")
-              tags+=("-t" "369495373322.dkr.ecr.eu-central-1.amazonaws.com/build-tools:${TO_TAG}")
-              tags+=("-t" "neoneastus2.azurecr.io/neondatabase/build-tools:${TO_TAG}")
-            fi
-
-            docker buildx imagetools create "${tags[@]}" \
-                                              neondatabase/build-tools:${FROM_TAG}-${debian_version}
-          done
+          docker buildx imagetools create -t 369495373322.dkr.ecr.eu-central-1.amazonaws.com/build-tools:${TO_TAG} \
+                                          -t neoneastus2.azurecr.io/neondatabase/build-tools:${TO_TAG} \
+                                          -t neondatabase/build-tools:${TO_TAG} \
+                                             neondatabase/build-tools:${FROM_TAG}
--- a/.github/workflows/report-workflow-stats.yml
+++ b/.github/workflows/report-workflow-stats.yml
@@ -1,41 +0,0 @@
-name: Report Workflow Stats
-
-on:
-  workflow_run:
-    workflows:
-    - Add `external` label to issues and PRs created by external users
-    - Benchmarking
-    - Build and Test
-    - Build and Test Locally
-    - Build build-tools image
-    - Check Permissions
-    - Check build-tools image
-    - Check neon with extra platform builds
-    - Cloud Regression Test
-    - Create Release Branch
-    - Handle `approved-for-ci-run` label
-    - Lint GitHub Workflows
-    - Notify Slack channel about upcoming release
-    - Periodic pagebench performance test on dedicated EC2 machine in eu-central-1 region
-    - Pin build-tools image
-    - Prepare benchmarking databases by restoring dumps
-    - Push images to ACR
-    - Test Postgres client libraries
-    - Trigger E2E Tests
-    - cleanup caches by a branch
-    types: [completed]
-
-jobs:
-  gh-workflow-stats:
-    name: Github Workflow Stats
-    runs-on: ubuntu-22.04
-    permissions:
-      actions: read
-    steps:
-    - name: Export GH Workflow Stats
-      uses: neondatabase/gh-workflow-stats-action@v0.1.4
-      with:
-        DB_URI: ${{ secrets.GH_REPORT_STATS_DB_RW_CONNSTR }}
-        DB_TABLE: "gh_workflow_stats_neon"
-        GH_TOKEN: ${{ secrets.GITHUB_TOKEN }}
-        GH_RUN_ID: ${{ github.event.workflow_run.id }}
--- a/1
+++ b/1
@@ -1,6 +1,5 @@
 /compute_tools/ @neondatabase/control-plane @neondatabase/compute
 /storage_controller @neondatabase/storage
-/storage_scrubber @neondatabase/storage
 /libs/pageserver_api/ @neondatabase/storage
 /libs/postgres_ffi/ @neondatabase/compute @neondatabase/storage
 /libs/remote_storage/ @neondatabase/storage
--- a/Cargo.lock
+++ b/Cargo.lock
@@ -82,12 +82,27 @@ dependencies = [
 "anstyle",
 "anstyle-parse",
 "anstyle-query",
- "anstyle-wincon",
+ "anstyle-wincon 1.0.1",
 "colorchoice",
 "is-terminal",
 "utf8parse",
 ]

+[[package]]
+name = "anstream"
+version = "0.6.15"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "64e15c1ab1f89faffbf04a634d5e1962e9074f2741eef6d97f3c4e322426d526"
+dependencies = [
+ "anstyle",
+ "anstyle-parse",
+ "anstyle-query",
+ "anstyle-wincon 3.0.4",
+ "colorchoice",
+ "is_terminal_polyfill",
+ "utf8parse",
+]
+
 [[package]]
 name = "anstyle"
 version = "1.0.8"
@@ -122,6 +137,16 @@ dependencies = [
 "windows-sys 0.48.0",
 ]

+[[package]]
+name = "anstyle-wincon"
+version = "3.0.4"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "5bf74e1b6e971609db8ca7a9ce79fd5768ab6ae46441c572e46cf596f59e57f8"
+dependencies = [
+ "anstyle",
+ "windows-sys 0.52.0",
+]
+
 [[package]]
 name = "anyhow"
 version = "1.0.71"
@@ -668,20 +693,19 @@ dependencies = [

 [[package]]
 name = "axum"
-version = "0.7.5"
+version = "0.6.20"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "3a6c9af12842a67734c9a2e355436e5d03b22383ed60cf13cd0c18fbfe3dcbcf"
+checksum = "3b829e4e32b91e643de6eafe82b1d90675f5874230191a4ffbc1b336dec4d6bf"
 dependencies = [
 "async-trait",
- "axum-core",
+ "axum-core 0.3.4",
 "base64 0.21.1",
+ "bitflags 1.3.2",
 "bytes",
 "futures-util",
- "http 1.1.0",
- "http-body 1.0.0",
- "http-body-util",
- "hyper 1.4.1",
- "hyper-util",
+ "http 0.2.9",
+ "http-body 0.4.5",
+ "hyper 0.14.30",
 "itoa",
 "matchit 0.7.0",
 "memchr",
@@ -694,13 +718,56 @@ dependencies = [
 "serde_path_to_error",
 "serde_urlencoded",
 "sha1",
- "sync_wrapper 1.0.1",
+ "sync_wrapper 0.1.2",
 "tokio",
 "tokio-tungstenite",
 "tower",
 "tower-layer",
 "tower-service",
- "tracing",
+]
+
+[[package]]
+name = "axum"
+version = "0.7.5"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "3a6c9af12842a67734c9a2e355436e5d03b22383ed60cf13cd0c18fbfe3dcbcf"
+dependencies = [
+ "async-trait",
+ "axum-core 0.4.5",
+ "bytes",
+ "futures-util",
+ "http 1.1.0",
+ "http-body 1.0.0",
+ "http-body-util",
+ "itoa",
+ "matchit 0.7.0",
+ "memchr",
+ "mime",
+ "percent-encoding",
+ "pin-project-lite",
+ "rustversion",
+ "serde",
+ "sync_wrapper 1.0.1",
+ "tower",
+ "tower-layer",
+ "tower-service",
+]
+
+[[package]]
+name = "axum-core"
+version = "0.3.4"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "759fa577a247914fd3f7f76d62972792636412fbfd634cd452f6a385a74d2d2c"
+dependencies = [
+ "async-trait",
+ "bytes",
+ "futures-util",
+ "http 0.2.9",
+ "http-body 0.4.5",
+ "mime",
+ "rustversion",
+ "tower-layer",
+ "tower-service",
 ]

 [[package]]
@@ -721,7 +788,6 @@ dependencies = [
 "sync_wrapper 1.0.1",
 "tower-layer",
 "tower-service",
- "tracing",
 ]

 [[package]]
@@ -1144,7 +1210,7 @@ version = "4.3.0"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "4f423e341edefb78c9caba2d9c7f7687d0e72e89df3ce3394554754393ac3990"
 dependencies = [
- "anstream",
+ "anstream 0.3.2",
 "anstyle",
 "bitflags 1.3.2",
 "clap_lex",
@@ -1220,7 +1286,6 @@ version = "0.1.0"
 dependencies = [
 "anyhow",
 "bytes",
- "camino",
 "cfg-if",
 "chrono",
 "clap",
@@ -1820,7 +1885,6 @@ source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "b5e6043086bf7973472e0c7dff2142ea0b680d30e18d9cc40f267efbf222bd47"
 dependencies = [
 "base16ct 0.2.0",
- "base64ct",
 "crypto-bigint 0.5.5",
 "digest",
 "ff 0.13.0",
@@ -1830,8 +1894,6 @@ dependencies = [
 "pkcs8 0.10.2",
 "rand_core 0.6.4",
 "sec1 0.7.3",
- "serde_json",
- "serdect",
 "subtle",
 "zeroize",
 ]
@@ -1887,6 +1949,15 @@ dependencies = [
 "syn 2.0.52",
 ]

+[[package]]
+name = "env_filter"
+version = "0.1.2"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "4f2c92ceda6ceec50f43169f9ee8424fe2db276791afde7b2cd8bc084cb376ab"
+dependencies = [
+ "log",
+]
+
 [[package]]
 name = "env_logger"
 version = "0.10.2"
@@ -1900,6 +1971,18 @@ dependencies = [
 "termcolor",
 ]

+[[package]]
+name = "env_logger"
+version = "0.11.2"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "6c012a26a7f605efc424dd53697843a72be7dc86ad2d01f7814337794a12231d"
+dependencies = [
+ "anstream 0.6.15",
+ "anstyle",
+ "env_filter",
+ "log",
+]
+
 [[package]]
 name = "equivalent"
 version = "1.0.1"
@@ -2695,7 +2778,6 @@ checksum = "ad227c3af19d4914570ad36d30409928b75967c298feb9ea1969db3a610bb14e"
 dependencies = [
 "equivalent",
 "hashbrown 0.14.5",
- "serde",
 ]

 [[package]]
@@ -2775,6 +2857,12 @@ dependencies = [
 "windows-sys 0.52.0",
 ]

+[[package]]
+name = "is_terminal_polyfill"
+version = "1.70.1"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "7943c866cc5cd64cbc25b2e01621d07fa8eb2a1a23160ee81ce38704e97b8ecf"
+
 [[package]]
 name = "itertools"
 version = "0.10.5"
@@ -2795,9 +2883,9 @@ dependencies = [

 [[package]]
 name = "itoa"
-version = "1.0.11"
+version = "1.0.6"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "49f1f14873335454500d59611f1cf4a4b0f786f9ac11f4312a78e4cf2566695b"
+checksum = "453ad9f582a441959e5f0d088b02ce04cfe8d51a8eaf077f12ac6d3e94164ca6"

 [[package]]
 name = "jobserver"
@@ -3218,6 +3306,16 @@ dependencies = [
 "winapi",
 ]

+[[package]]
+name = "nu-ansi-term"
+version = "0.46.0"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "77a8165726e8236064dbb45459242600304b42a5ea24ee2948e18e023bf7ba84"
+dependencies = [
+ "overload",
+ "winapi",
+]
+
 [[package]]
 name = "num"
 version = "0.4.1"
@@ -3501,6 +3599,12 @@ version = "0.5.1"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "4030760ffd992bef45b0ae3f10ce1aba99e33464c90d14dd7c039884963ddc7a"

+[[package]]
+name = "overload"
+version = "0.1.1"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "b15813163c1d831bf4a13c3610c05c0d03b39feb07f7e09fa234dac9b15aaf39"
+
 [[package]]
 name = "p256"
 version = "0.11.1"
@@ -3656,7 +3760,6 @@ dependencies = [
 "tracing",
 "url",
 "utils",
- "wal_decoder",
 "walkdir",
 "workspace_hack",
 ]
@@ -4042,8 +4145,6 @@ dependencies = [
 "bytes",
 "fallible-iterator",
 "postgres-protocol",
- "serde",
- "serde_json",
 ]

 [[package]]
@@ -4086,7 +4187,7 @@ dependencies = [
 "bindgen",
 "bytes",
 "crc32c",
- "env_logger",
+ "env_logger 0.10.2",
 "log",
 "memoffset 0.9.0",
 "once_cell",
@@ -4094,7 +4195,6 @@ dependencies = [
 "regex",
 "serde",
 "thiserror",
- "tracing",
 "utils",
 ]

@@ -4280,7 +4380,7 @@ dependencies = [
 "consumption_metrics",
 "dashmap",
 "ecdsa 0.16.9",
- "env_logger",
+ "env_logger 0.10.2",
 "fallible-iterator",
 "framed-websockets",
 "futures",
@@ -4299,7 +4399,6 @@ dependencies = [
 "indexmap 2.0.1",
 "ipnet",
 "itertools 0.10.5",
- "itoa",
 "jose-jwa",
 "jose-jwk",
 "lasso",
@@ -5265,7 +5364,6 @@ dependencies = [
 "der 0.7.8",
 "generic-array",
 "pkcs8 0.10.2",
- "serdect",
 "subtle",
 "zeroize",
 ]
@@ -5520,16 +5618,6 @@ dependencies = [
 "syn 2.0.52",
 ]

-[[package]]
-name = "serdect"
-version = "0.2.0"
-source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "a84f14a19e9a014bb9f4512488d9829a68e04ecabffb0f9904cd1ace94598177"
-dependencies = [
- "base16ct 0.2.0",
- "serde",
-]
-
 [[package]]
 name = "sha1"
 version = "0.10.5"
@@ -5767,6 +5855,7 @@ dependencies = [
 "serde_json",
 "strum",
 "strum_macros",
+ "test-log",
 "thiserror",
 "tokio",
 "tokio-util",
@@ -6019,6 +6108,28 @@ dependencies = [
 "syn 2.0.52",
 ]

+[[package]]
+name = "test-log"
+version = "0.2.16"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "3dffced63c2b5c7be278154d76b479f9f9920ed34e7574201407f0b14e2bbb93"
+dependencies = [
+ "env_logger 0.11.2",
+ "test-log-macros",
+ "tracing-subscriber",
+]
+
+[[package]]
+name = "test-log-macros"
+version = "0.2.16"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "5999e24eaa32083191ba4e425deb75cdf25efefabe5aaccb7446dd0d4122a3f5"
+dependencies = [
+ "proc-macro2",
+ "quote",
+ "syn 2.0.52",
+]
+
 [[package]]
 name = "thiserror"
 version = "1.0.57"
@@ -6310,9 +6421,9 @@ dependencies = [

 [[package]]
 name = "tokio-tungstenite"
-version = "0.21.0"
+version = "0.20.0"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "c83b561d025642014097b66e6c1bb422783339e0909e4429cde4749d1990bc38"
+checksum = "2b2dbec703c26b00d74844519606ef15d09a7d6857860f84ad223dec002ddea2"
 dependencies = [
 "futures-util",
 "log",
@@ -6379,7 +6490,7 @@ checksum = "877c5b330756d856ffcc4553ab34a5684481ade925ecc54bcd1bf02b1d0d4d52"
 dependencies = [
 "async-stream",
 "async-trait",
- "axum",
+ "axum 0.7.5",
 "base64 0.22.1",
 "bytes",
 "h2 0.4.4",
@@ -6550,6 +6661,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "ad0f048c97dbd9faa9b7df56362b8ebcaa52adb06b498c050d2f4e32f90a7a8b"
 dependencies = [
 "matchers",
+ "nu-ansi-term",
 "once_cell",
 "regex",
 "serde",
@@ -6586,14 +6698,14 @@ checksum = "e421abadd41a4225275504ea4d6566923418b7f05506fbc9c0fe86ba7396114b"

 [[package]]
 name = "tungstenite"
-version = "0.21.0"
+version = "0.20.1"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "9ef1a641ea34f399a848dea702823bbecfb4c486f911735368f1f137cb8257e1"
+checksum = "9e3dac10fd62eaf6617d3a904ae222845979aec67c615d1c842b4002c7666fb9"
 dependencies = [
 "byteorder",
 "bytes",
 "data-encoding",
- "http 1.1.0",
+ "http 0.2.9",
 "httparse",
 "log",
 "rand 0.8.5",
@@ -6821,7 +6933,7 @@ name = "vm_monitor"
 version = "0.1.0"
 dependencies = [
 "anyhow",
- "axum",
+ "axum 0.6.20",
 "cgroups-rs",
 "clap",
 "futures",
@@ -6854,7 +6966,7 @@ dependencies = [
 "anyhow",
 "camino-tempfile",
 "clap",
- "env_logger",
+ "env_logger 0.10.2",
 "log",
 "postgres",
 "postgres_ffi",
@@ -6862,19 +6974,6 @@ dependencies = [
 "utils",
 ]

-[[package]]
-name = "wal_decoder"
-version = "0.1.0"
-dependencies = [
- "anyhow",
- "bytes",
- "pageserver_api",
- "postgres_ffi",
- "serde",
- "tracing",
- "utils",
-]
-
 [[package]]
 name = "walkdir"
 version = "2.3.3"
@@ -7293,8 +7392,12 @@ version = "0.1.0"
 dependencies = [
 "ahash",
 "anyhow",
- "axum",
- "axum-core",
+ "aws-config",
+ "aws-runtime",
+ "aws-sigv4",
+ "aws-smithy-async",
+ "aws-smithy-http",
+ "aws-smithy-types",
 "base64 0.21.1",
 "base64ct",
 "bytes",
@@ -7324,7 +7427,6 @@ dependencies = [
 "hyper 1.4.1",
 "hyper-util",
 "indexmap 1.9.3",
- "indexmap 2.0.1",
 "itertools 0.12.1",
 "lazy_static",
 "libc",
@@ -7336,7 +7438,6 @@ dependencies = [
 "num-traits",
 "once_cell",
 "parquet",
- "postgres-types",
 "prettyplease",
 "proc-macro2",
 "prost",
@@ -7346,6 +7447,7 @@ dependencies = [
 "regex-automata 0.4.3",
 "regex-syntax 0.8.2",
 "reqwest 0.12.4",
+ "rustls 0.21.11",
 "scopeguard",
 "serde",
 "serde_json",
@@ -7361,7 +7463,6 @@ dependencies = [
 "time",
 "time-macros",
 "tokio",
- "tokio-postgres",
 "tokio-stream",
 "tokio-util",
 "toml_edit",
@@ -7370,6 +7471,7 @@ dependencies = [
 "tracing",
 "tracing-core",
 "url",
+ "uuid",
 "zeroize",
 "zstd",
 "zstd-safe",
--- a/Cargo.toml
+++ b/Cargo.toml
@@ -33,7 +33,6 @@ members = [
    "libs/postgres_ffi/wal_craft",
    "libs/vm_monitor",
    "libs/walproposer",
-    "libs/wal_decoder",
 ]

 [workspace.package]
@@ -54,7 +53,7 @@ azure_storage_blobs = { version = "0.19", default-features = false, features = [
 flate2 = "1.0.26"
 async-stream = "0.3"
 async-trait = "0.1"
-aws-config = { version = "1.5", default-features = false, features=["rustls", "sso"] }
+aws-config = { version = "1.5", default-features = false, features=["rustls"] }
 aws-sdk-s3 = "1.52"
 aws-sdk-iam = "1.46.0"
 aws-smithy-async = { version = "1.2.1", default-features = false, features=["rt-tokio"] }
@@ -62,7 +61,7 @@ aws-smithy-types = "1.2"
 aws-credential-types = "1.2.0"
 aws-sigv4 = { version = "1.2", features = ["sign-http"] }
 aws-types = "1.3"
-axum = { version = "0.7.5", features = ["ws"] }
+axum = { version = "0.6.20", features = ["ws"] }
 base64 = "0.13.0"
 bincode = "1.3"
 bindgen = "0.70"
@@ -100,15 +99,14 @@ http-types = { version = "2", default-features = false }
 http-body-util = "0.1.2"
 humantime = "2.1"
 humantime-serde = "1.1.1"
-hyper0 = { package = "hyper", version = "0.14" }
-hyper = "1.4"
+hyper = "0.14"
+hyper_1 = { package = "hyper", version = "1.4" }
 hyper-util = "0.1"
-tokio-tungstenite = "0.21.0"
+tokio-tungstenite = "0.20.0"
 indexmap = "2"
 indoc = "2"
 ipnet = "2.9.0"
 itertools = "0.10"
-itoa = "1.0.11"
 jsonwebtoken = "9"
 lasso = "0.7"
 libc = "0.2"
@@ -239,7 +237,6 @@ tracing-utils = { version = "0.1", path = "./libs/tracing-utils/" }
 utils = { version = "0.1", path = "./libs/utils/" }
 vm_monitor = { version = "0.1", path = "./libs/vm_monitor/" }
 walproposer = { version = "0.1", path = "./libs/walproposer/" }
-wal_decoder = { version = "0.1", path = "./libs/wal_decoder" }

 ## Common library dependency
 workspace_hack = { version = "0.1", path = "./workspace_hack/" }
--- a/4
+++ b/4
@@ -7,8 +7,6 @@ ARG IMAGE=build-tools
 ARG TAG=pinned
 ARG DEFAULT_PG_VERSION=17
 ARG STABLE_PG_VERSION=16
-ARG DEBIAN_VERSION=bullseye
-ARG DEBIAN_FLAVOR=${DEBIAN_VERSION}-slim

 # Build Postgres
 FROM $REPOSITORY/$IMAGE:$TAG AS pg-build
@@ -59,7 +57,7 @@ RUN set -e \

 # Build final image
 #
-FROM debian:${DEBIAN_FLAVOR}
+FROM debian:bullseye-slim
 ARG DEFAULT_PG_VERSION
 WORKDIR /data

--- a/Dockerfile.build-tools
+++ b/Dockerfile.build-tools
@@ -1,7 +1,12 @@
-ARG DEBIAN_VERSION=bullseye
+FROM debian:bullseye-slim

-FROM debian:${DEBIAN_VERSION}-slim
-ARG DEBIAN_VERSION
+# Use ARG as a build-time environment variable here to allow.
+# It's not supposed to be set outside.
+# Alternatively it can be obtained using the following command
+# ```
+# . /etc/os-release && echo "${VERSION_CODENAME}"
+# ```
+ARG DEBIAN_VERSION_CODENAME=bullseye

 # Add nonroot user
 RUN useradd -ms /bin/bash nonroot -b /home
@@ -37,14 +42,14 @@ RUN set -e \
        libseccomp-dev \
        libsqlite3-dev \
        libssl-dev \
-        $([[ "${DEBIAN_VERSION}" = "bullseye" ]] && libstdc++-10-dev || libstdc++-11-dev) \
+        libstdc++-10-dev \
        libtool \
        libxml2-dev \
        libxmlsec1-dev \
        libxxhash-dev \
        lsof \
        make \
-        netcat-openbsd \
+        netcat \
        net-tools \
        openssh-client \
        parallel \
@@ -73,7 +78,7 @@ RUN curl -sL "https://github.com/peak/s5cmd/releases/download/v${S5CMD_VERSION}/
 # LLVM
 ENV LLVM_VERSION=18
 RUN curl -fsSL 'https://apt.llvm.org/llvm-snapshot.gpg.key' | apt-key add - \
-    && echo "deb http://apt.llvm.org/${DEBIAN_VERSION}/ llvm-toolchain-${DEBIAN_VERSION}-${LLVM_VERSION} main" > /etc/apt/sources.list.d/llvm.stable.list \
+    && echo "deb http://apt.llvm.org/${DEBIAN_VERSION_CODENAME}/ llvm-toolchain-${DEBIAN_VERSION_CODENAME}-${LLVM_VERSION} main" > /etc/apt/sources.list.d/llvm.stable.list \
    && apt update \
    && apt install -y clang-${LLVM_VERSION} llvm-${LLVM_VERSION} \
    && bash -c 'for f in /usr/bin/clang*-${LLVM_VERSION} /usr/bin/llvm*-${LLVM_VERSION}; do ln -s "${f}" "${f%-${LLVM_VERSION}}"; done' \
@@ -81,7 +86,7 @@ RUN curl -fsSL 'https://apt.llvm.org/llvm-snapshot.gpg.key' | apt-key add - \

 # Install docker
 RUN curl -fsSL https://download.docker.com/linux/ubuntu/gpg | gpg --dearmor -o /usr/share/keyrings/docker-archive-keyring.gpg \
-    && echo "deb [arch=$(dpkg --print-architecture) signed-by=/usr/share/keyrings/docker-archive-keyring.gpg] https://download.docker.com/linux/debian ${DEBIAN_VERSION} stable" > /etc/apt/sources.list.d/docker.list \
+    && echo "deb [arch=$(dpkg --print-architecture) signed-by=/usr/share/keyrings/docker-archive-keyring.gpg] https://download.docker.com/linux/debian ${DEBIAN_VERSION_CODENAME} stable" > /etc/apt/sources.list.d/docker.list \
    && apt update \
    && apt install -y docker-ce docker-ce-cli \
    && rm -rf /var/lib/apt/lists/* /tmp/* /var/tmp/*
--- a/14
+++ b/14
@@ -168,27 +168,27 @@ postgres-check-%: postgres-%
 neon-pg-ext-%: postgres-%
 	+@echo "Compiling neon $*"
 	mkdir -p $(POSTGRES_INSTALL_DIR)/build/neon-$*
-	$(MAKE) PG_CONFIG=$(POSTGRES_INSTALL_DIR)/$*/bin/pg_config COPT='$(COPT)' \
+	$(MAKE) PG_CONFIG=$(POSTGRES_INSTALL_DIR)/$*/bin/pg_config CFLAGS='$(PG_CFLAGS) $(COPT)' \
 		-C $(POSTGRES_INSTALL_DIR)/build/neon-$* \
 		-f $(ROOT_PROJECT_DIR)/pgxn/neon/Makefile install
 	+@echo "Compiling neon_walredo $*"
 	mkdir -p $(POSTGRES_INSTALL_DIR)/build/neon-walredo-$*
-	$(MAKE) PG_CONFIG=$(POSTGRES_INSTALL_DIR)/$*/bin/pg_config COPT='$(COPT)' \
+	$(MAKE) PG_CONFIG=$(POSTGRES_INSTALL_DIR)/$*/bin/pg_config CFLAGS='$(PG_CFLAGS) $(COPT)' \
 		-C $(POSTGRES_INSTALL_DIR)/build/neon-walredo-$* \
 		-f $(ROOT_PROJECT_DIR)/pgxn/neon_walredo/Makefile install
 	+@echo "Compiling neon_rmgr $*"
 	mkdir -p $(POSTGRES_INSTALL_DIR)/build/neon-rmgr-$*
-	$(MAKE) PG_CONFIG=$(POSTGRES_INSTALL_DIR)/$*/bin/pg_config COPT='$(COPT)' \
+	$(MAKE) PG_CONFIG=$(POSTGRES_INSTALL_DIR)/$*/bin/pg_config CFLAGS='$(PG_CFLAGS) $(COPT)' \
 		-C $(POSTGRES_INSTALL_DIR)/build/neon-rmgr-$* \
 		-f $(ROOT_PROJECT_DIR)/pgxn/neon_rmgr/Makefile install
 	+@echo "Compiling neon_test_utils $*"
 	mkdir -p $(POSTGRES_INSTALL_DIR)/build/neon-test-utils-$*
-	$(MAKE) PG_CONFIG=$(POSTGRES_INSTALL_DIR)/$*/bin/pg_config COPT='$(COPT)' \
+	$(MAKE) PG_CONFIG=$(POSTGRES_INSTALL_DIR)/$*/bin/pg_config CFLAGS='$(PG_CFLAGS) $(COPT)' \
 		-C $(POSTGRES_INSTALL_DIR)/build/neon-test-utils-$* \
 		-f $(ROOT_PROJECT_DIR)/pgxn/neon_test_utils/Makefile install
 	+@echo "Compiling neon_utils $*"
 	mkdir -p $(POSTGRES_INSTALL_DIR)/build/neon-utils-$*
-	$(MAKE) PG_CONFIG=$(POSTGRES_INSTALL_DIR)/$*/bin/pg_config COPT='$(COPT)' \
+	$(MAKE) PG_CONFIG=$(POSTGRES_INSTALL_DIR)/$*/bin/pg_config CFLAGS='$(PG_CFLAGS) $(COPT)' \
 		-C $(POSTGRES_INSTALL_DIR)/build/neon-utils-$* \
 		-f $(ROOT_PROJECT_DIR)/pgxn/neon_utils/Makefile install

@@ -220,7 +220,7 @@ neon-pg-clean-ext-%:
 walproposer-lib: neon-pg-ext-v17
 	+@echo "Compiling walproposer-lib"
 	mkdir -p $(POSTGRES_INSTALL_DIR)/build/walproposer-lib
-	$(MAKE) PG_CONFIG=$(POSTGRES_INSTALL_DIR)/v17/bin/pg_config COPT='$(COPT)' \
+	$(MAKE) PG_CONFIG=$(POSTGRES_INSTALL_DIR)/v17/bin/pg_config CFLAGS='$(PG_CFLAGS) $(COPT)' \
 		-C $(POSTGRES_INSTALL_DIR)/build/walproposer-lib \
 		-f $(ROOT_PROJECT_DIR)/pgxn/neon/Makefile walproposer-lib
 	cp $(POSTGRES_INSTALL_DIR)/v17/lib/libpgport.a $(POSTGRES_INSTALL_DIR)/build/walproposer-lib
@@ -333,7 +333,7 @@ postgres-%-pgindent: postgres-%-pg-bsd-indent postgres-%-typedefs.list
 # Indent pxgn/neon.
 .PHONY: neon-pgindent
 neon-pgindent: postgres-v17-pg-bsd-indent neon-pg-ext-v17
-	$(MAKE) PG_CONFIG=$(POSTGRES_INSTALL_DIR)/v17/bin/pg_config COPT='$(COPT)' \
+	$(MAKE) PG_CONFIG=$(POSTGRES_INSTALL_DIR)/v17/bin/pg_config CFLAGS='$(PG_CFLAGS) $(COPT)' \
 		FIND_TYPEDEF=$(ROOT_PROJECT_DIR)/vendor/postgres-v17/src/tools/find_typedef \
 		INDENT=$(POSTGRES_INSTALL_DIR)/build/v17/src/tools/pg_bsd_indent/pg_bsd_indent \
 		PGINDENT_SCRIPT=$(ROOT_PROJECT_DIR)/vendor/postgres-v17/src/tools/pgindent/pgindent \
--- a/README.md
+++ b/README.md
@@ -58,7 +58,7 @@ curl --proto '=https' --tlsv1.2 -sSf https://sh.rustup.rs | sh
 1. Install XCode and dependencies
 ```
 xcode-select --install
-brew install protobuf openssl flex bison icu4c pkg-config m4
+brew install protobuf openssl flex bison icu4c pkg-config

 # add openssl to PATH, required for ed25519 keys generation in neon_local
 echo 'export PATH="$(brew --prefix openssl)/bin:$PATH"' >> ~/.zshrc
--- a/compute/Dockerfile.compute-node
+++ b/compute/Dockerfile.compute-node
@@ -3,8 +3,7 @@ ARG REPOSITORY=neondatabase
 ARG IMAGE=build-tools
 ARG TAG=pinned
 ARG BUILD_TAG
-ARG DEBIAN_VERSION=bullseye
-ARG DEBIAN_FLAVOR=${DEBIAN_VERSION}-slim
+ARG DEBIAN_FLAVOR=bullseye-slim

 #########################################################################################
 #
@@ -12,23 +11,20 @@ ARG DEBIAN_FLAVOR=${DEBIAN_VERSION}-slim
 #
 #########################################################################################
 FROM debian:$DEBIAN_FLAVOR AS build-deps
-ARG DEBIAN_VERSION
+ARG DEBIAN_FLAVOR

-RUN case $DEBIAN_VERSION in \
+RUN case $DEBIAN_FLAVOR in \
      # Version-specific installs for Bullseye (PG14-PG16):
      # The h3_pg extension needs a cmake 3.20+, but Debian bullseye has 3.18.
      # Install newer version (3.25) from backports.
-      bullseye) \
+      bullseye*) \
        echo "deb http://deb.debian.org/debian bullseye-backports main" > /etc/apt/sources.list.d/bullseye-backports.list; \
        VERSION_INSTALLS="cmake/bullseye-backports cmake-data/bullseye-backports"; \
      ;; \
      # Version-specific installs for Bookworm (PG17):
-      bookworm) \
+      bookworm*) \
        VERSION_INSTALLS="cmake"; \
      ;; \
-      *) \
-        echo "Unknown Debian version ${DEBIAN_VERSION}" && exit 1 \
-      ;; \
    esac && \
    apt update &&  \
    apt install --no-install-recommends -y git autoconf automake libtool build-essential bison flex libreadline-dev \
@@ -113,30 +109,13 @@ RUN apt update && \
    libcgal-dev libgdal-dev libgmp-dev libmpfr-dev libopenscenegraph-dev libprotobuf-c-dev \
    protobuf-c-compiler xsltproc

-
-# Postgis 3.5.0 requires SFCGAL 1.4+
-#
-# It would be nice to update all versions together, but we must solve the SFCGAL dependency first.
 # SFCGAL > 1.3 requires CGAL > 5.2, Bullseye's libcgal-dev is 5.2
-# and also we must check backward compatibility with older versions of PostGIS.
-#
-# Use new version only for v17
-RUN case "${PG_VERSION}" in \
-    "v17") \
-        export SFCGAL_VERSION=1.4.1 \
-        export SFCGAL_CHECKSUM=1800c8a26241588f11cddcf433049e9b9aea902e923414d2ecef33a3295626c3 \
-    ;; \
-    "v14" | "v15" | "v16") \
-        export SFCGAL_VERSION=1.3.10 \
-        export SFCGAL_CHECKSUM=4e39b3b2adada6254a7bdba6d297bb28e1a9835a9f879b74f37e2dab70203232 \
-    ;; \
-    *) \
-        echo "unexpected PostgreSQL version" && exit 1 \
-    ;; \
-    esac && \
+RUN case "${PG_VERSION}" in "v17") \
    mkdir -p /sfcgal && \
-    wget https://gitlab.com/sfcgal/SFCGAL/-/archive/v${SFCGAL_VERSION}/SFCGAL-v${SFCGAL_VERSION}.tar.gz -O SFCGAL.tar.gz && \
-    echo "${SFCGAL_CHECKSUM} SFCGAL.tar.gz" | sha256sum --check && \
+    echo "Postgis doensn't yet support PG17 (needs 3.4.3, if not higher)" && exit 0;; \
+    esac && \
+    wget https://gitlab.com/Oslandia/SFCGAL/-/archive/v1.3.10/SFCGAL-v1.3.10.tar.gz -O SFCGAL.tar.gz && \
+    echo "4e39b3b2adada6254a7bdba6d297bb28e1a9835a9f879b74f37e2dab70203232 SFCGAL.tar.gz" | sha256sum --check && \
    mkdir sfcgal-src && cd sfcgal-src && tar xzf ../SFCGAL.tar.gz --strip-components=1 -C . && \
    cmake -DCMAKE_BUILD_TYPE=Release . && make -j $(getconf _NPROCESSORS_ONLN) && \
    DESTDIR=/sfcgal make install -j $(getconf _NPROCESSORS_ONLN) && \
@@ -144,27 +123,15 @@ RUN case "${PG_VERSION}" in \

 ENV PATH="/usr/local/pgsql/bin:$PATH"

-# Postgis 3.5.0 supports v17
-RUN case "${PG_VERSION}" in \
-    "v17") \
-        export POSTGIS_VERSION=3.5.0 \
-        export POSTGIS_CHECKSUM=ca698a22cc2b2b3467ac4e063b43a28413f3004ddd505bdccdd74c56a647f510 \
-    ;; \
-    "v14" | "v15" | "v16") \
-        export POSTGIS_VERSION=3.3.3 \
-        export POSTGIS_CHECKSUM=74eb356e3f85f14233791013360881b6748f78081cc688ff9d6f0f673a762d13 \
-    ;; \
-    *) \
-        echo "unexpected PostgreSQL version" && exit 1 \
-    ;; \
+RUN case "${PG_VERSION}" in "v17") \
+    echo "Postgis doensn't yet support PG17 (needs 3.4.3, if not higher)" && exit 0;; \
    esac && \
-    wget https://download.osgeo.org/postgis/source/postgis-${POSTGIS_VERSION}.tar.gz -O postgis.tar.gz && \
-    echo "${POSTGIS_CHECKSUM} postgis.tar.gz" | sha256sum --check && \
+    wget https://download.osgeo.org/postgis/source/postgis-3.3.3.tar.gz -O postgis.tar.gz && \
+    echo "74eb356e3f85f14233791013360881b6748f78081cc688ff9d6f0f673a762d13 postgis.tar.gz" | sha256sum --check && \
    mkdir postgis-src && cd postgis-src && tar xzf ../postgis.tar.gz --strip-components=1 -C . && \
    find /usr/local/pgsql -type f | sed 's|^/usr/local/pgsql/||' > /before.txt &&\
    ./autogen.sh && \
    ./configure --with-sfcgal=/usr/local/bin/sfcgal-config && \
-    make -j $(getconf _NPROCESSORS_ONLN) && \
    make -j $(getconf _NPROCESSORS_ONLN) install && \
    cd extensions/postgis && \
    make clean && \
@@ -185,27 +152,11 @@ RUN case "${PG_VERSION}" in \
    cp /usr/local/pgsql/share/extension/address_standardizer.control /extensions/postgis && \
    cp /usr/local/pgsql/share/extension/address_standardizer_data_us.control /extensions/postgis

-# Uses versioned libraries, i.e. libpgrouting-3.4
-# and may introduce function signature changes between releases
-# i.e. release 3.5.0 has new signature for pg_dijkstra function
-#
-# Use new version only for v17
-# last release v3.6.2 - Mar 30, 2024
-RUN case "${PG_VERSION}" in \
-    "v17") \
-        export PGROUTING_VERSION=3.6.2 \
-        export PGROUTING_CHECKSUM=f4a1ed79d6f714e52548eca3bb8e5593c6745f1bde92eb5fb858efd8984dffa2 \
-    ;; \
-    "v14" | "v15" | "v16") \
-        export PGROUTING_VERSION=3.4.2 \
-        export PGROUTING_CHECKSUM=cac297c07d34460887c4f3b522b35c470138760fe358e351ad1db4edb6ee306e \
-    ;; \
-    *) \
-        echo "unexpected PostgreSQL version" && exit 1 \
-    ;; \
+RUN case "${PG_VERSION}" in "v17") \
+    echo "v17 extensions are not supported yet. Quit" && exit 0;; \
    esac && \
-    wget https://github.com/pgRouting/pgrouting/archive/v${PGROUTING_VERSION}.tar.gz -O pgrouting.tar.gz && \
-    echo "${PGROUTING_CHECKSUM} pgrouting.tar.gz" | sha256sum --check && \
+    wget https://github.com/pgRouting/pgrouting/archive/v3.4.2.tar.gz -O pgrouting.tar.gz && \
+    echo "cac297c07d34460887c4f3b522b35c470138760fe358e351ad1db4edb6ee306e pgrouting.tar.gz" | sha256sum --check && \
    mkdir pgrouting-src && cd pgrouting-src && tar xzf ../pgrouting.tar.gz --strip-components=1 -C . && \
    mkdir build && cd build && \
    cmake -DCMAKE_BUILD_TYPE=Release .. && \
@@ -264,9 +215,10 @@ FROM build-deps AS h3-pg-build
 ARG PG_VERSION
 COPY --from=pg-build /usr/local/pgsql/ /usr/local/pgsql/

-# not version-specific
-# last release v4.1.0 - Jan 18, 2023
-RUN mkdir -p /h3/usr/ && \
+RUN case "${PG_VERSION}" in "v17") \
+        mkdir -p /h3/usr/ && \
+    echo "v17 extensions are not supported yet. Quit" && exit 0;; \
+    esac && \
    wget https://github.com/uber/h3/archive/refs/tags/v4.1.0.tar.gz -O h3.tar.gz && \
    echo "ec99f1f5974846bde64f4513cf8d2ea1b8d172d2218ab41803bf6a63532272bc h3.tar.gz" | sha256sum --check && \
    mkdir h3-src && cd h3-src && tar xzf ../h3.tar.gz --strip-components=1 -C . && \
@@ -277,9 +229,10 @@ RUN mkdir -p /h3/usr/ && \
    cp -R /h3/usr / && \
    rm -rf build

-# not version-specific
-# last release v4.1.3 - Jul 26, 2023
-RUN wget https://github.com/zachasme/h3-pg/archive/refs/tags/v4.1.3.tar.gz -O h3-pg.tar.gz && \
+RUN case "${PG_VERSION}" in "v17") \
+    echo "v17 extensions are not supported yet. Quit" && exit 0;; \
+    esac && \
+    wget https://github.com/zachasme/h3-pg/archive/refs/tags/v4.1.3.tar.gz -O h3-pg.tar.gz && \
    echo "5c17f09a820859ffe949f847bebf1be98511fb8f1bd86f94932512c00479e324 h3-pg.tar.gz" | sha256sum --check && \
    mkdir h3-pg-src && cd h3-pg-src && tar xzf ../h3-pg.tar.gz --strip-components=1 -C . && \
    export PATH="/usr/local/pgsql/bin:$PATH" && \
@@ -298,10 +251,11 @@ FROM build-deps AS unit-pg-build
 ARG PG_VERSION
 COPY --from=pg-build /usr/local/pgsql/ /usr/local/pgsql/

-# not version-specific
-# last release 7.9 - Sep 15, 2024
-RUN wget https://github.com/df7cb/postgresql-unit/archive/refs/tags/7.9.tar.gz -O postgresql-unit.tar.gz && \
-    echo "e46de6245dcc8b2c2ecf29873dbd43b2b346773f31dd5ce4b8315895a052b456 postgresql-unit.tar.gz" | sha256sum --check && \
+RUN case "${PG_VERSION}" in "v17") \
+    echo "v17 extensions are not supported yet. Quit" && exit 0;; \
+    esac && \
+    wget https://github.com/df7cb/postgresql-unit/archive/refs/tags/7.7.tar.gz -O postgresql-unit.tar.gz && \
+    echo "411d05beeb97e5a4abf17572bfcfbb5a68d98d1018918feff995f6ee3bb03e79 postgresql-unit.tar.gz" | sha256sum --check && \
    mkdir postgresql-unit-src && cd postgresql-unit-src && tar xzf ../postgresql-unit.tar.gz --strip-components=1 -C . && \
    make -j $(getconf _NPROCESSORS_ONLN) PG_CONFIG=/usr/local/pgsql/bin/pg_config && \
    make -j $(getconf _NPROCESSORS_ONLN) install PG_CONFIG=/usr/local/pgsql/bin/pg_config && \
@@ -348,10 +302,12 @@ FROM build-deps AS pgjwt-pg-build
 ARG PG_VERSION
 COPY --from=pg-build /usr/local/pgsql/ /usr/local/pgsql/

-# not version-specific
-# doesn't use releases, last commit f3d82fd - Mar 2, 2023 
-RUN wget https://github.com/michelp/pgjwt/archive/f3d82fd30151e754e19ce5d6a06c71c20689ce3d.tar.gz -O pgjwt.tar.gz && \
-    echo "dae8ed99eebb7593b43013f6532d772b12dfecd55548d2673f2dfd0163f6d2b9 pgjwt.tar.gz" | sha256sum --check && \
+# 9742dab1b2f297ad3811120db7b21451bca2d3c9 made on 13/11/2021
+RUN case "${PG_VERSION}" in "v17") \
+    echo "v17 extensions are not supported yet. Quit" && exit 0;; \
+    esac && \
+    wget https://github.com/michelp/pgjwt/archive/9742dab1b2f297ad3811120db7b21451bca2d3c9.tar.gz -O pgjwt.tar.gz && \
+    echo "cfdefb15007286f67d3d45510f04a6a7a495004be5b3aecb12cda667e774203f pgjwt.tar.gz" | sha256sum --check && \
    mkdir pgjwt-src && cd pgjwt-src && tar xzf ../pgjwt.tar.gz --strip-components=1 -C . && \
    make -j $(getconf _NPROCESSORS_ONLN) install PG_CONFIG=/usr/local/pgsql/bin/pg_config && \
    echo 'trusted = true' >> /usr/local/pgsql/share/extension/pgjwt.control
@@ -386,9 +342,10 @@ FROM build-deps AS pg-hashids-pg-build
 ARG PG_VERSION
 COPY --from=pg-build /usr/local/pgsql/ /usr/local/pgsql/

-# not version-specific
-# last release v1.2.1 -Jan 12, 2018
-RUN wget https://github.com/iCyberon/pg_hashids/archive/refs/tags/v1.2.1.tar.gz -O pg_hashids.tar.gz && \
+RUN case "${PG_VERSION}" in "v17") \
+    echo "v17 extensions are not supported yet. Quit" && exit 0;; \
+    esac && \
+    wget https://github.com/iCyberon/pg_hashids/archive/refs/tags/v1.2.1.tar.gz -O pg_hashids.tar.gz && \
    echo "74576b992d9277c92196dd8d816baa2cc2d8046fe102f3dcd7f3c3febed6822a pg_hashids.tar.gz" | sha256sum --check && \
    mkdir pg_hashids-src && cd pg_hashids-src && tar xzf ../pg_hashids.tar.gz --strip-components=1 -C . && \
    make -j $(getconf _NPROCESSORS_ONLN) PG_CONFIG=/usr/local/pgsql/bin/pg_config USE_PGXS=1 && \
@@ -448,9 +405,10 @@ FROM build-deps AS ip4r-pg-build
 ARG PG_VERSION
 COPY --from=pg-build /usr/local/pgsql/ /usr/local/pgsql/

-# not version-specific
-# last release v2.4.2 - Jul 29, 2023
-RUN wget https://github.com/RhodiumToad/ip4r/archive/refs/tags/2.4.2.tar.gz -O ip4r.tar.gz && \
+RUN case "${PG_VERSION}" in "v17") \
+    echo "v17 extensions are not supported yet. Quit" && exit 0;; \
+    esac && \
+    wget https://github.com/RhodiumToad/ip4r/archive/refs/tags/2.4.2.tar.gz -O ip4r.tar.gz && \
    echo "0f7b1f159974f49a47842a8ab6751aecca1ed1142b6d5e38d81b064b2ead1b4b ip4r.tar.gz" | sha256sum --check && \
    mkdir ip4r-src && cd ip4r-src && tar xzf ../ip4r.tar.gz --strip-components=1 -C . && \
    make -j $(getconf _NPROCESSORS_ONLN) PG_CONFIG=/usr/local/pgsql/bin/pg_config && \
@@ -467,9 +425,10 @@ FROM build-deps AS prefix-pg-build
 ARG PG_VERSION
 COPY --from=pg-build /usr/local/pgsql/ /usr/local/pgsql/

-# not version-specific
-# last release v1.2.10  - Jul 5, 2023
-RUN wget https://github.com/dimitri/prefix/archive/refs/tags/v1.2.10.tar.gz -O prefix.tar.gz && \
+RUN case "${PG_VERSION}" in "v17") \
+    echo "v17 extensions are not supported yet. Quit" && exit 0;; \
+    esac && \
+    wget https://github.com/dimitri/prefix/archive/refs/tags/v1.2.10.tar.gz -O prefix.tar.gz && \
    echo "4342f251432a5f6fb05b8597139d3ccde8dcf87e8ca1498e7ee931ca057a8575 prefix.tar.gz" | sha256sum --check && \
    mkdir prefix-src && cd prefix-src && tar xzf ../prefix.tar.gz --strip-components=1 -C . && \
    make -j $(getconf _NPROCESSORS_ONLN) PG_CONFIG=/usr/local/pgsql/bin/pg_config && \
@@ -486,9 +445,10 @@ FROM build-deps AS hll-pg-build
 ARG PG_VERSION
 COPY --from=pg-build /usr/local/pgsql/ /usr/local/pgsql/

-# not version-specific
-# last release v2.18 - Aug 29, 2023
-RUN wget https://github.com/citusdata/postgresql-hll/archive/refs/tags/v2.18.tar.gz -O hll.tar.gz && \
+RUN case "${PG_VERSION}" in "v17") \
+    echo "v17 extensions are not supported yet. Quit" && exit 0;; \
+    esac && \
+    wget https://github.com/citusdata/postgresql-hll/archive/refs/tags/v2.18.tar.gz -O hll.tar.gz && \
    echo "e2f55a6f4c4ab95ee4f1b4a2b73280258c5136b161fe9d059559556079694f0e hll.tar.gz" | sha256sum --check && \
    mkdir hll-src && cd hll-src && tar xzf ../hll.tar.gz --strip-components=1 -C . && \
    make -j $(getconf _NPROCESSORS_ONLN) PG_CONFIG=/usr/local/pgsql/bin/pg_config && \
@@ -699,10 +659,11 @@ FROM build-deps AS pg-roaringbitmap-pg-build
 ARG PG_VERSION
 COPY --from=pg-build /usr/local/pgsql/ /usr/local/pgsql/

-# not version-specific
-# last release v0.5.4 - Jun 28, 2022
 ENV PATH="/usr/local/pgsql/bin/:$PATH"
-RUN wget https://github.com/ChenHuajun/pg_roaringbitmap/archive/refs/tags/v0.5.4.tar.gz -O pg_roaringbitmap.tar.gz && \
+RUN case "${PG_VERSION}" in "v17") \
+    echo "v17 extensions is not supported yet by pg_roaringbitmap. Quit" && exit 0;; \
+    esac && \
+    wget https://github.com/ChenHuajun/pg_roaringbitmap/archive/refs/tags/v0.5.4.tar.gz -O pg_roaringbitmap.tar.gz && \
    echo "b75201efcb1c2d1b014ec4ae6a22769cc7a224e6e406a587f5784a37b6b5a2aa pg_roaringbitmap.tar.gz" | sha256sum --check && \
    mkdir pg_roaringbitmap-src && cd pg_roaringbitmap-src && tar xzf ../pg_roaringbitmap.tar.gz --strip-components=1 -C . && \
    make -j $(getconf _NPROCESSORS_ONLN) && \
@@ -719,27 +680,12 @@ FROM build-deps AS pg-semver-pg-build
 ARG PG_VERSION
 COPY --from=pg-build /usr/local/pgsql/ /usr/local/pgsql/

-# Release 0.40.0 breaks backward compatibility with previous versions
-# see release note https://github.com/theory/pg-semver/releases/tag/v0.40.0
-# Use new version only for v17
-#
-# last release v0.40.0 - Jul 22, 2024
 ENV PATH="/usr/local/pgsql/bin/:$PATH"
-RUN case "${PG_VERSION}" in \
-    "v17") \
-        export SEMVER_VERSION=0.40.0 \
-        export SEMVER_CHECKSUM=3e50bcc29a0e2e481e7b6d2bc937cadc5f5869f55d983b5a1aafeb49f5425cfc \
-    ;; \
-    "v14" | "v15" | "v16") \
-        export SEMVER_VERSION=0.32.1 \
-        export SEMVER_CHECKSUM=fbdaf7512026d62eec03fad8687c15ed509b6ba395bff140acd63d2e4fbe25d7 \
-    ;; \
-    *) \
-        echo "unexpected PostgreSQL version" && exit 1 \
-    ;; \
+RUN case "${PG_VERSION}" in "v17") \
+    echo "v17 is not supported yet by pg_semver. Quit" && exit 0;; \
    esac && \
-    wget https://github.com/theory/pg-semver/archive/refs/tags/v${SEMVER_VERSION}.tar.gz -O pg_semver.tar.gz && \
-    echo "${SEMVER_CHECKSUM} pg_semver.tar.gz" | sha256sum --check && \
+    wget https://github.com/theory/pg-semver/archive/refs/tags/v0.32.1.tar.gz -O pg_semver.tar.gz && \
+    echo "fbdaf7512026d62eec03fad8687c15ed509b6ba395bff140acd63d2e4fbe25d7 pg_semver.tar.gz" | sha256sum --check && \
    mkdir pg_semver-src && cd pg_semver-src && tar xzf ../pg_semver.tar.gz --strip-components=1 -C . && \
    make -j $(getconf _NPROCESSORS_ONLN) && \
    make -j $(getconf _NPROCESSORS_ONLN) install && \
@@ -929,11 +875,14 @@ ARG PG_VERSION
 RUN case "${PG_VERSION}" in "v17") \
    echo "pg_session_jwt does not yet have a release that supports pg17" && exit 0;; \
    esac && \
-    wget https://github.com/neondatabase/pg_session_jwt/archive/5aee2625af38213650e1a07ae038fdc427250ee4.tar.gz -O pg_session_jwt.tar.gz && \
-    echo "5d91b10bc1347d36cffc456cb87bec25047935d6503dc652ca046f04760828e7 pg_session_jwt.tar.gz" | sha256sum --check && \
+    wget https://github.com/neondatabase/pg_session_jwt/archive/ff0a72440e8ff584dab24b3f9b7c00c56c660b8e.tar.gz -O pg_session_jwt.tar.gz && \
+    echo "1fbb2b5a339263bcf6daa847fad8bccbc0b451cea6a62e6d3bf232b0087f05cb pg_session_jwt.tar.gz" | sha256sum --check && \
    mkdir pg_session_jwt-src && cd pg_session_jwt-src && tar xzf ../pg_session_jwt.tar.gz --strip-components=1 -C . && \
    sed -i 's/pgrx = "=0.11.3"/pgrx = { version = "=0.11.3", features = [ "unsafe-postgres" ] }/g' Cargo.toml && \
    cargo pgrx install --release
+    # it's needed to enable extension because it uses untrusted C language
+    # sed -i 's/superuser = false/superuser = true/g' /usr/local/pgsql/share/extension/pg_session_jwt.control && \
+    # echo "trusted = true" >> /usr/local/pgsql/share/extension/pg_session_jwt.control

 #########################################################################################
 #
@@ -1095,6 +1044,7 @@ RUN cd compute_tools && mold -run cargo build --locked --profile release-line-de
 #########################################################################################

 FROM debian:$DEBIAN_FLAVOR AS compute-tools-image
+ARG DEBIAN_FLAVOR

 COPY --from=compute-tools /home/nonroot/target/release-line-debug-size-lto/compute_ctl /usr/local/bin/compute_ctl

@@ -1105,6 +1055,7 @@ COPY --from=compute-tools /home/nonroot/target/release-line-debug-size-lto/compu
 #########################################################################################

 FROM debian:$DEBIAN_FLAVOR AS pgbouncer
+ARG DEBIAN_FLAVOR
 RUN set -e \
    && apt-get update \
    && apt-get install --no-install-recommends -y \
@@ -1127,20 +1078,6 @@ RUN set -e \
    && make -j $(nproc) dist_man_MANS= \
    && make install dist_man_MANS=

-#########################################################################################
-#
-# Compile the Neon-specific `local_proxy` binary
-#
-#########################################################################################
-FROM $REPOSITORY/$IMAGE:$TAG AS local_proxy
-ARG BUILD_TAG
-ENV BUILD_TAG=$BUILD_TAG
-
-USER nonroot
-# Copy entire project to get Cargo.* files with proper dependencies for the whole project
-COPY --chown=nonroot . .
-RUN mold -run cargo build --locked --profile release-line-debug-size-lto --bin local_proxy
-
 #########################################################################################
 #
 # Layers "postgres-exporter" and "sql-exporter"
@@ -1259,7 +1196,7 @@ ENV PGDATABASE=postgres
 #
 #########################################################################################
 FROM debian:$DEBIAN_FLAVOR
-ARG DEBIAN_VERSION
+ARG DEBIAN_FLAVOR
 # Add user postgres
 RUN mkdir /var/db && useradd -m -d /var/db/postgres postgres && \
    echo "postgres:test_console_pass" | chpasswd && \
@@ -1279,10 +1216,6 @@ COPY --from=compute-tools --chown=postgres /home/nonroot/target/release-line-deb
 COPY --from=pgbouncer         /usr/local/pgbouncer/bin/pgbouncer /usr/local/bin/pgbouncer
 COPY --chmod=0666 --chown=postgres compute/etc/pgbouncer.ini /etc/pgbouncer.ini

-# local_proxy and its config
-COPY --from=local_proxy --chown=postgres /home/nonroot/target/release-line-debug-size-lto/local_proxy /usr/local/bin/local_proxy
-RUN mkdir -p /etc/local_proxy && chown postgres:postgres /etc/local_proxy
-
 # Metrics exporter binaries and  configuration files
 COPY --from=postgres-exporter /bin/postgres_exporter /bin/postgres_exporter
 COPY --from=sql-exporter      /bin/sql_exporter      /bin/sql_exporter
@@ -1307,22 +1240,19 @@ RUN mkdir /usr/local/download_extensions && chown -R postgres:postgres /usr/loca


 RUN apt update && \
-    case $DEBIAN_VERSION in \
+    case $DEBIAN_FLAVOR in \
      # Version-specific installs for Bullseye (PG14-PG16):
      # libicu67, locales for collations (including ICU and plpgsql_check)
      # libgdal28, libproj19 for PostGIS
-      bullseye) \
+      bullseye*) \
        VERSION_INSTALLS="libicu67 libgdal28 libproj19"; \
      ;; \
      # Version-specific installs for Bookworm (PG17):
      # libicu72, locales for collations (including ICU and plpgsql_check)
      # libgdal32, libproj25 for PostGIS
-      bookworm) \
+      bookworm*) \
        VERSION_INSTALLS="libicu72 libgdal32 libproj25"; \
      ;; \
-      *) \
-        echo "Unknown Debian version ${DEBIAN_VERSION}" && exit 1 \
-      ;; \
    esac && \
    apt install --no-install-recommends -y \
        gdb \
--- a/compute/vm-image-spec-bookworm.yaml
+++ b/compute/vm-image-spec-bookworm.yaml
@@ -1,126 +0,0 @@
-# Supplemental file for neondatabase/autoscaling's vm-builder, for producing the VM compute image.
---
-commands:
-  - name: cgconfigparser
-    user: root
-    sysvInitAction: sysinit
-    shell: 'cgconfigparser -l /etc/cgconfig.conf -s 1664'
-  # restrict permissions on /neonvm/bin/resize-swap, because we grant access to compute_ctl for
-  # running it as root.
-  - name: chmod-resize-swap
-    user: root
-    sysvInitAction: sysinit
-    shell: 'chmod 711 /neonvm/bin/resize-swap'
-  - name: chmod-set-disk-quota
-    user: root
-    sysvInitAction: sysinit
-    shell: 'chmod 711 /neonvm/bin/set-disk-quota'
-  - name: pgbouncer
-    user: postgres
-    sysvInitAction: respawn
-    shell: '/usr/local/bin/pgbouncer /etc/pgbouncer.ini'
-  - name: local_proxy
-    user: postgres
-    sysvInitAction: respawn
-    shell: '/usr/local/bin/local_proxy --config-path /etc/local_proxy/config.json --pid-path /etc/local_proxy/pid --http 0.0.0.0:10432'
-  - name: postgres-exporter
-    user: nobody
-    sysvInitAction: respawn
-    shell: 'DATA_SOURCE_NAME="user=cloud_admin sslmode=disable dbname=postgres application_name=postgres-exporter" /bin/postgres_exporter'
-  - name: sql-exporter
-    user: nobody
-    sysvInitAction: respawn
-    shell: '/bin/sql_exporter -config.file=/etc/sql_exporter.yml -web.listen-address=:9399'
-  - name: sql-exporter-autoscaling
-    user: nobody
-    sysvInitAction: respawn
-    shell: '/bin/sql_exporter -config.file=/etc/sql_exporter_autoscaling.yml -web.listen-address=:9499'
-shutdownHook: |
-  su -p postgres --session-command '/usr/local/bin/pg_ctl stop -D /var/db/postgres/compute/pgdata -m fast --wait -t 10'
-files:
-  - filename: compute_ctl-sudoers
-    content: |
-      # Allow postgres user (which is what compute_ctl runs as) to run /neonvm/bin/resize-swap
-      # and /neonvm/bin/set-disk-quota as root without requiring entering a password (NOPASSWD),
-      # regardless of hostname (ALL)
-      postgres ALL=(root) NOPASSWD: /neonvm/bin/resize-swap, /neonvm/bin/set-disk-quota
-  - filename: cgconfig.conf
-    content: |
-      # Configuration for cgroups in VM compute nodes
-      group neon-postgres {
-          perm {
-              admin {
-                  uid = postgres;
-              }
-              task {
-                  gid = users;
-              }
-          }
-          memory {}
-      }
-build: |
-  # Build cgroup-tools
-  #
-  # At time of writing (2023-03-14), debian bullseye has a version of cgroup-tools (technically
-  # libcgroup) that doesn't support cgroup v2 (version 0.41-11). Unfortunately, the vm-monitor
-  # requires cgroup v2, so we'll build cgroup-tools ourselves.
-  #
-  # At time of migration to bookworm (2024-10-09), debian has a version of libcgroup/cgroup-tools 2.0.2,
-  # and it _probably_ can be used as-is. However, we'll build it ourselves to minimise the changeset
-  # for debian version migration.
-  #
-  FROM debian:bookworm-slim as libcgroup-builder
-  ENV LIBCGROUP_VERSION=v2.0.3
-
-  RUN set -exu \
-      && apt update \
-      && apt install --no-install-recommends -y \
-          git \
-          ca-certificates \
-          automake \
-          cmake \
-          make \
-          gcc \
-          byacc \
-          flex \
-          libtool \
-          libpam0g-dev \
-      && git clone --depth 1 -b $LIBCGROUP_VERSION https://github.com/libcgroup/libcgroup \
-      && INSTALL_DIR="/libcgroup-install" \
-      && mkdir -p "$INSTALL_DIR/bin" "$INSTALL_DIR/include" \
-      && cd libcgroup \
-      # extracted from bootstrap.sh, with modified flags:
-      && (test -d m4 || mkdir m4) \
-      && autoreconf -fi \
-      && rm -rf autom4te.cache \
-      && CFLAGS="-O3" ./configure --prefix="$INSTALL_DIR" --sysconfdir=/etc --localstatedir=/var --enable-opaque-hierarchy="name=systemd" \
-      # actually build the thing...
-      && make install
-merge: |
-  # tweak nofile limits
-  RUN set -e \
-      && echo 'fs.file-max = 1048576' >>/etc/sysctl.conf \
-      && test ! -e /etc/security || ( \
-         echo '*    - nofile 1048576' >>/etc/security/limits.conf \
-      && echo 'root - nofile 1048576' >>/etc/security/limits.conf \
-         )
-
-  # Allow postgres user (compute_ctl) to run swap resizer.
-  # Need to install sudo in order to allow this.
-  #
-  # Also, remove the 'read' permission from group/other on /neonvm/bin/resize-swap, just to be safe.
-  RUN set -e \
-      && apt update \
-      && apt install --no-install-recommends -y \
-             sudo \
-      && rm -rf /var/lib/apt/lists/* /tmp/* /var/tmp/*
-  COPY compute_ctl-sudoers /etc/sudoers.d/compute_ctl-sudoers
-
-  COPY cgconfig.conf /etc/cgconfig.conf
-
-  RUN set -e \
-      && chmod 0644 /etc/cgconfig.conf
-
-  COPY --from=libcgroup-builder /libcgroup-install/bin/*  /usr/bin/
-  COPY --from=libcgroup-builder /libcgroup-install/lib/*  /usr/lib/
-  COPY --from=libcgroup-builder /libcgroup-install/sbin/* /usr/sbin/
--- a/compute/vm-image-spec-bullseye.yaml
+++ b/compute/vm-image-spec-bullseye.yaml
@@ -19,10 +19,6 @@ commands:
    user: postgres
    sysvInitAction: respawn
    shell: '/usr/local/bin/pgbouncer /etc/pgbouncer.ini'
-  - name: local_proxy
-    user: postgres
-    sysvInitAction: respawn
-    shell: '/usr/local/bin/local_proxy --config-path /etc/local_proxy/config.json --pid-path /etc/local_proxy/pid --http 0.0.0.0:10432'
  - name: postgres-exporter
    user: nobody
    sysvInitAction: respawn
--- a/compute_tools/Cargo.toml
+++ b/compute_tools/Cargo.toml
@@ -11,13 +11,12 @@ testing = []

 [dependencies]
 anyhow.workspace = true
-camino.workspace = true
 chrono.workspace = true
 cfg-if.workspace = true
 clap.workspace = true
 flate2.workspace = true
 futures.workspace = true
-hyper0 = { workspace = true, features = ["full"] }
+hyper = { workspace = true, features = ["full"] }
 nix.workspace = true
 notify.workspace = true
 num_cpus.workspace = true
--- a/compute_tools/src/bin/compute_ctl.rs
+++ b/compute_tools/src/bin/compute_ctl.rs
@@ -402,7 +402,8 @@ fn start_postgres(
 ) -> Result<(Option<PostgresHandle>, StartPostgresResult)> {
    // We got all we need, update the state.
    let mut state = compute.state.lock().unwrap();
-    state.set_status(ComputeStatus::Init, &compute.state_changed);
+    state.status = ComputeStatus::Init;
+    compute.state_changed.notify_all();

    info!(
        "running compute with features: {:?}",
--- a/compute_tools/src/compute.rs
+++ b/compute_tools/src/compute.rs
@@ -34,7 +34,6 @@ use nix::sys::signal::{kill, Signal};
 use remote_storage::{DownloadError, RemotePath};

 use crate::checker::create_availability_check_data;
-use crate::local_proxy;
 use crate::logger::inlinify;
 use crate::pg_helpers::*;
 use crate::spec::*;
@@ -109,18 +108,6 @@ impl ComputeState {
            metrics: ComputeMetrics::default(),
        }
    }
-
-    pub fn set_status(&mut self, status: ComputeStatus, state_changed: &Condvar) {
-        let prev = self.status;
-        info!("Changing compute status from {} to {}", prev, status);
-        self.status = status;
-        state_changed.notify_all();
-    }
-
-    pub fn set_failed_status(&mut self, err: anyhow::Error, state_changed: &Condvar) {
-        self.error = Some(format!("{err:?}"));
-        self.set_status(ComputeStatus::Failed, state_changed);
-    }
 }

 impl Default for ComputeState {
@@ -315,12 +302,15 @@ impl ComputeNode {

    pub fn set_status(&self, status: ComputeStatus) {
        let mut state = self.state.lock().unwrap();
-        state.set_status(status, &self.state_changed);
+        state.status = status;
+        self.state_changed.notify_all();
    }

    pub fn set_failed_status(&self, err: anyhow::Error) {
        let mut state = self.state.lock().unwrap();
-        state.set_failed_status(err, &self.state_changed);
+        state.error = Some(format!("{err:?}"));
+        state.status = ComputeStatus::Failed;
+        self.state_changed.notify_all();
    }

    pub fn get_status(&self) -> ComputeStatus {
@@ -896,11 +886,6 @@ impl ComputeNode {
        // 'Close' connection
        drop(client);

-        if let Some(ref local_proxy) = spec.local_proxy_config {
-            info!("configuring local_proxy");
-            local_proxy::configure(local_proxy).context("apply_config local_proxy")?;
-        }
-
        // Run migrations separately to not hold up cold starts
        thread::spawn(move || {
            let mut connstr = connstr.clone();
@@ -951,19 +936,6 @@ impl ComputeNode {
            });
        }

-        if let Some(ref local_proxy) = spec.local_proxy_config {
-            info!("configuring local_proxy");
-
-            // Spawn a thread to do the configuration,
-            // so that we don't block the main thread that starts Postgres.
-            let local_proxy = local_proxy.clone();
-            let _handle = Some(thread::spawn(move || {
-                if let Err(err) = local_proxy::configure(&local_proxy) {
-                    error!("error while configuring local_proxy: {err:?}");
-                }
-            }));
-        }
-
        // Write new config
        let pgdata_path = Path::new(&self.pgdata);
        let postgresql_conf_path = pgdata_path.join("postgresql.conf");
@@ -1051,19 +1023,6 @@ impl ComputeNode {
            });
        }

-        if let Some(local_proxy) = &pspec.spec.local_proxy_config {
-            info!("configuring local_proxy");
-
-            // Spawn a thread to do the configuration,
-            // so that we don't block the main thread that starts Postgres.
-            let local_proxy = local_proxy.clone();
-            let _handle = thread::spawn(move || {
-                if let Err(err) = local_proxy::configure(&local_proxy) {
-                    error!("error while configuring local_proxy: {err:?}");
-                }
-            });
-        }
-
        info!(
            "start_compute spec.remote_extensions {:?}",
            pspec.spec.remote_extensions
@@ -1484,28 +1443,6 @@ LIMIT 100",
            info!("Pageserver config changed");
        }
    }
-
-    // Gather info about installed extensions
-    pub fn get_installed_extensions(&self) -> Result<()> {
-        let connstr = self.connstr.clone();
-
-        let rt = tokio::runtime::Builder::new_current_thread()
-            .enable_all()
-            .build()
-            .expect("failed to create runtime");
-        let result = rt
-            .block_on(crate::installed_extensions::get_installed_extensions(
-                connstr,
-            ))
-            .expect("failed to get installed extensions");
-
-        info!(
-            "{}",
-            serde_json::to_string(&result).expect("failed to serialize extensions list")
-        );
-
-        Ok(())
-    }
 }

 pub fn forward_termination_signal() {
--- a/compute_tools/src/configurator.rs
+++ b/compute_tools/src/configurator.rs
@@ -24,7 +24,8 @@ fn configurator_main_loop(compute: &Arc<ComputeNode>) {
        // Re-check the status after waking up
        if state.status == ComputeStatus::ConfigurationPending {
            info!("got configuration request");
-            state.set_status(ComputeStatus::Configuration, &compute.state_changed);
+            state.status = ComputeStatus::Configuration;
+            compute.state_changed.notify_all();
            drop(state);

            let mut new_status = ComputeStatus::Failed;
--- a/compute_tools/src/http/api.rs
+++ b/compute_tools/src/http/api.rs
@@ -165,32 +165,6 @@ async fn routes(req: Request<Body>, compute: &Arc<ComputeNode>) -> Response<Body
            }
        }

-        // get the list of installed extensions
-        // currently only used in python tests
-        // TODO: call it from cplane
-        (&Method::GET, "/installed_extensions") => {
-            info!("serving /installed_extensions GET request");
-            let status = compute.get_status();
-            if status != ComputeStatus::Running {
-                let msg = format!(
-                    "invalid compute status for extensions request: {:?}",
-                    status
-                );
-                error!(msg);
-                return Response::new(Body::from(msg));
-            }
-
-            let connstr = compute.connstr.clone();
-            let res = crate::installed_extensions::get_installed_extensions(connstr).await;
-            match res {
-                Ok(res) => render_json(Body::from(serde_json::to_string(&res).unwrap())),
-                Err(e) => render_json_error(
-                    &format!("could not get list of installed extensions: {}", e),
-                    StatusCode::INTERNAL_SERVER_ERROR,
-                ),
-            }
-        }
-
        // download extension files from remote extension storage on demand
        (&Method::POST, route) if route.starts_with("/extension_server/") => {
            info!("serving {:?} POST request", route);
@@ -314,7 +288,8 @@ async fn handle_configure_request(
                return Err((msg, StatusCode::PRECONDITION_FAILED));
            }
            state.pspec = Some(parsed_spec);
-            state.set_status(ComputeStatus::ConfigurationPending, &compute.state_changed);
+            state.status = ComputeStatus::ConfigurationPending;
+            compute.state_changed.notify_all();
            drop(state);
            info!("set new spec and notified waiters");
        }
@@ -387,15 +362,15 @@ async fn handle_terminate_request(compute: &Arc<ComputeNode>) -> Result<(), (Str
        }
        if state.status != ComputeStatus::Empty && state.status != ComputeStatus::Running {
            let msg = format!(
-                "invalid compute status for termination request: {}",
-                state.status
+                "invalid compute status for termination request: {:?}",
+                state.status.clone()
            );
            return Err((msg, StatusCode::PRECONDITION_FAILED));
        }
-        state.set_status(ComputeStatus::TerminationPending, &compute.state_changed);
+        state.status = ComputeStatus::TerminationPending;
+        compute.state_changed.notify_all();
        drop(state);
    }
-
    forward_termination_signal();
    info!("sent signal and notified waiters");

@@ -409,8 +384,7 @@ async fn handle_terminate_request(compute: &Arc<ComputeNode>) -> Result<(), (Str
        while state.status != ComputeStatus::Terminated {
            state = c.state_changed.wait(state).unwrap();
            info!(
-                "waiting for compute to become {}, current status: {:?}",
-                ComputeStatus::Terminated,
+                "waiting for compute to become Terminated, current status: {:?}",
                state.status
            );
        }
--- a/compute_tools/src/http/openapi_spec.yaml
+++ b/compute_tools/src/http/openapi_spec.yaml
@@ -53,20 +53,6 @@ paths:
              schema:
                $ref: "#/components/schemas/ComputeInsights"

-  /installed_extensions:
-    get:
-      tags:
-      - Info
-      summary: Get installed extensions.
-      description: ""
-      operationId: getInstalledExtensions
-      responses:
-        200:
-          description: List of installed extensions
-          content:
-            application/json:
-              schema:
-                $ref: "#/components/schemas/InstalledExtensions"
  /info:
    get:
      tags:
@@ -409,24 +395,6 @@ components:
        - configuration
      example: running

-    InstalledExtensions:
-      type: object
-      properties:
-        extensions:
-          description: Contains list of installed extensions.
-          type: array
-          items:
-            type: object
-            properties:
-              extname:
-                type: string
-              versions:
-                type: array
-                items:
-                  type: string
-              n_databases:
-                type: integer
-
    #
    # Errors
    #
--- a/compute_tools/src/installed_extensions.rs
+++ b/compute_tools/src/installed_extensions.rs
@@ -1,80 +0,0 @@
-use compute_api::responses::{InstalledExtension, InstalledExtensions};
-use std::collections::HashMap;
-use std::collections::HashSet;
-use url::Url;
-
-use anyhow::Result;
-use postgres::{Client, NoTls};
-use tokio::task;
-
-/// We don't reuse get_existing_dbs() just for code clarity
-/// and to make database listing query here more explicit.
-///
-/// Limit the number of databases to 500 to avoid excessive load.
-fn list_dbs(client: &mut Client) -> Result<Vec<String>> {
-    // `pg_database.datconnlimit = -2` means that the database is in the
-    // invalid state
-    let databases = client
-        .query(
-            "SELECT datname FROM pg_catalog.pg_database
-                WHERE datallowconn
-                AND datconnlimit <> - 2
-                LIMIT 500",
-            &[],
-        )?
-        .iter()
-        .map(|row| {
-            let db: String = row.get("datname");
-            db
-        })
-        .collect();
-
-    Ok(databases)
-}
-
-/// Connect to every database (see list_dbs above) and get the list of installed extensions.
-/// Same extension can be installed in multiple databases with different versions,
-/// we only keep the highest and lowest version across all databases.
-pub async fn get_installed_extensions(connstr: Url) -> Result<InstalledExtensions> {
-    let mut connstr = connstr.clone();
-
-    task::spawn_blocking(move || {
-        let mut client = Client::connect(connstr.as_str(), NoTls)?;
-        let databases: Vec<String> = list_dbs(&mut client)?;
-
-        let mut extensions_map: HashMap<String, InstalledExtension> = HashMap::new();
-        for db in databases.iter() {
-            connstr.set_path(db);
-            let mut db_client = Client::connect(connstr.as_str(), NoTls)?;
-            let extensions: Vec<(String, String)> = db_client
-                .query(
-                    "SELECT extname, extversion FROM pg_catalog.pg_extension;",
-                    &[],
-                )?
-                .iter()
-                .map(|row| (row.get("extname"), row.get("extversion")))
-                .collect();
-
-            for (extname, v) in extensions.iter() {
-                let version = v.to_string();
-                extensions_map
-                    .entry(extname.to_string())
-                    .and_modify(|e| {
-                        e.versions.insert(version.clone());
-                        // count the number of databases where the extension is installed
-                        e.n_databases += 1;
-                    })
-                    .or_insert(InstalledExtension {
-                        extname: extname.to_string(),
-                        versions: HashSet::from([version.clone()]),
-                        n_databases: 1,
-                    });
-            }
-        }
-
-        Ok(InstalledExtensions {
-            extensions: extensions_map.values().cloned().collect(),
-        })
-    })
-    .await?
-}
--- a/compute_tools/src/lib.rs
+++ b/compute_tools/src/lib.rs
@@ -2,9 +2,6 @@
 //! configuration.
 #![deny(unsafe_code)]
 #![deny(clippy::undocumented_unsafe_blocks)]
-
-extern crate hyper0 as hyper;
-
 pub mod checker;
 pub mod config;
 pub mod configurator;
@@ -15,8 +12,6 @@ pub mod catalog;
 pub mod compute;
 pub mod disk_quota;
 pub mod extension_server;
-pub mod installed_extensions;
-pub mod local_proxy;
 pub mod lsn_lease;
 mod migration;
 pub mod monitor;
--- a/compute_tools/src/local_proxy.rs
+++ b/compute_tools/src/local_proxy.rs
@@ -1,56 +0,0 @@
-//! Local Proxy is a feature of our BaaS Neon Authorize project.
-//!
-//! Local Proxy validates JWTs and manages the pg_session_jwt extension.
-//! It also maintains a connection pool to postgres.
-
-use anyhow::{Context, Result};
-use camino::Utf8Path;
-use compute_api::spec::LocalProxySpec;
-use nix::sys::signal::Signal;
-use utils::pid_file::{self, PidFileRead};
-
-pub fn configure(local_proxy: &LocalProxySpec) -> Result<()> {
-    write_local_proxy_conf("/etc/local_proxy/config.json".as_ref(), local_proxy)?;
-    notify_local_proxy("/etc/local_proxy/pid".as_ref())?;
-
-    Ok(())
-}
-
-/// Create or completely rewrite configuration file specified by `path`
-fn write_local_proxy_conf(path: &Utf8Path, local_proxy: &LocalProxySpec) -> Result<()> {
-    let config =
-        serde_json::to_string_pretty(local_proxy).context("serializing LocalProxySpec to json")?;
-    std::fs::write(path, config).with_context(|| format!("writing {path}"))?;
-
-    Ok(())
-}
-
-/// Notify local proxy about a new config file.
-fn notify_local_proxy(path: &Utf8Path) -> Result<()> {
-    match pid_file::read(path)? {
-        // if the file doesn't exist, or isn't locked, local_proxy isn't running
-        // and will naturally pick up our config later
-        PidFileRead::NotExist | PidFileRead::NotHeldByAnyProcess(_) => {}
-        PidFileRead::LockedByOtherProcess(pid) => {
-            // From the pid_file docs:
-            //
-            // > 1. The other process might exit at any time, turning the given PID stale.
-            // > 2. There is a small window in which `claim_for_current_process` has already
-            // >    locked the file but not yet updates its contents. [`read`] will return
-            // >    this variant here, but with the old file contents, i.e., a stale PID.
-            // >
-            // > The kernel is free to recycle PID once it has been `wait(2)`ed upon by
-            // > its creator. Thus, acting upon a stale PID, e.g., by issuing a `kill`
-            // > system call on it, bears the risk of killing an unrelated process.
-            // > This is an inherent limitation of using pidfiles.
-            // > The only race-free solution is to have a supervisor-process with a lifetime
-            // > that exceeds that of all of its child-processes (e.g., `runit`, `supervisord`).
-            //
-            // This is an ok risk as we only send a SIGHUP which likely won't actually
-            // kill the process, only reload config.
-            nix::sys::signal::kill(pid, Signal::SIGHUP).context("sending signal to local_proxy")?;
-        }
-    }
-
-    Ok(())
-}
--- a/compute_tools/src/spec.rs
+++ b/compute_tools/src/spec.rs
@@ -1,4 +1,3 @@
-use std::collections::HashSet;
 use std::fs::File;
 use std::path::Path;
 use std::str::FromStr;
@@ -190,15 +189,6 @@ pub fn handle_roles(spec: &ComputeSpec, client: &mut Client) -> Result<()> {
    let mut xact = client.transaction()?;
    let existing_roles: Vec<Role> = get_existing_roles(&mut xact)?;

-    let mut jwks_roles = HashSet::new();
-    if let Some(local_proxy) = &spec.local_proxy_config {
-        for jwks_setting in local_proxy.jwks.iter().flatten() {
-            for role_name in &jwks_setting.role_names {
-                jwks_roles.insert(role_name.clone());
-            }
-        }
-    }
-
    // Print a list of existing Postgres roles (only in debug mode)
    if span_enabled!(Level::INFO) {
        let mut vec = Vec::new();
@@ -318,9 +308,6 @@ pub fn handle_roles(spec: &ComputeSpec, client: &mut Client) -> Result<()> {
                    "CREATE ROLE {} INHERIT CREATEROLE CREATEDB BYPASSRLS REPLICATION IN ROLE neon_superuser",
                    name.pg_quote()
                );
-                if jwks_roles.contains(name.as_str()) {
-                    query = format!("CREATE ROLE {}", name.pg_quote());
-                }
                info!("running role create query: '{}'", &query);
                query.push_str(&role.to_pg_options());
                xact.execute(query.as_str(), &[])?;
--- a/control_plane/Cargo.toml
+++ b/control_plane/Cargo.toml
@@ -14,7 +14,7 @@ humantime.workspace = true
 nix.workspace = true
 once_cell.workspace = true
 humantime-serde.workspace = true
-hyper0.workspace = true
+hyper.workspace = true
 regex.workspace = true
 reqwest = { workspace = true, features = ["blocking", "json"] }
 scopeguard.workspace = true
--- a/control_plane/src/endpoint.rs
+++ b/control_plane/src/endpoint.rs
@@ -97,21 +97,7 @@ impl ComputeControlPlane {
        for endpoint_dir in std::fs::read_dir(env.endpoints_path())
            .with_context(|| format!("failed to list {}", env.endpoints_path().display()))?
        {
-            let ep_res = Endpoint::from_dir_entry(endpoint_dir?, &env);
-            let ep = match ep_res {
-                Ok(ep) => ep,
-                Err(e) => match e.downcast::<std::io::Error>() {
-                    Ok(e) => {
-                        // A parallel task could delete an endpoint while we have just scanned the directory
-                        if e.kind() == std::io::ErrorKind::NotFound {
-                            continue;
-                        } else {
-                            Err(e)?
-                        }
-                    }
-                    Err(e) => Err(e)?,
-                },
-            };
+            let ep = Endpoint::from_dir_entry(endpoint_dir?, &env)?;
            endpoints.insert(ep.endpoint_id.clone(), Arc::new(ep));
        }

@@ -613,7 +599,6 @@ impl Endpoint {
            remote_extensions,
            pgbouncer_settings: None,
            shard_stripe_size: Some(shard_stripe_size),
-            local_proxy_config: None,
        };
        let spec_path = self.endpoint_path().join("spec.json");
        std::fs::write(spec_path, serde_json::to_string_pretty(&spec)?)?;
--- a/control_plane/src/storage_controller.rs
+++ b/control_plane/src/storage_controller.rs
@@ -3,7 +3,7 @@ use crate::{
    local_env::{LocalEnv, NeonStorageControllerConf},
 };
 use camino::{Utf8Path, Utf8PathBuf};
-use hyper0::Uri;
+use hyper::Uri;
 use nix::unistd::Pid;
 use pageserver_api::{
    controller_api::{
--- a/docs/docker.md
+++ b/docs/docker.md
@@ -5,7 +5,7 @@
 Currently we build two main images:

 - [neondatabase/neon](https://hub.docker.com/repository/docker/neondatabase/neon) — image with pre-built `pageserver`, `safekeeper` and `proxy` binaries and all the required runtime dependencies. Built from [/Dockerfile](/Dockerfile).
- [neondatabase/compute-node-v16](https://hub.docker.com/repository/docker/neondatabase/compute-node-v16) — compute node image with pre-built Postgres binaries from [neondatabase/postgres](https://github.com/neondatabase/postgres). Similar images exist for v15 and v14. Built from [/compute-node/Dockerfile](/compute/Dockerfile.compute-node).
+- [neondatabase/compute-node-v16](https://hub.docker.com/repository/docker/neondatabase/compute-node-v16) — compute node image with pre-built Postgres binaries from [neondatabase/postgres](https://github.com/neondatabase/postgres). Similar images exist for v15 and v14.

 And additional intermediate image:

--- a/libs/compute_api/src/responses.rs
+++ b/libs/compute_api/src/responses.rs
@@ -1,8 +1,5 @@
 //! Structs representing the JSON formats used in the compute_ctl's HTTP API.

-use std::collections::HashSet;
-use std::fmt::Display;
-
 use chrono::{DateTime, Utc};
 use serde::{Deserialize, Serialize, Serializer};

@@ -61,21 +58,6 @@ pub enum ComputeStatus {
    Terminated,
 }

-impl Display for ComputeStatus {
-    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
-        match self {
-            ComputeStatus::Empty => f.write_str("empty"),
-            ComputeStatus::ConfigurationPending => f.write_str("configuration-pending"),
-            ComputeStatus::Init => f.write_str("init"),
-            ComputeStatus::Running => f.write_str("running"),
-            ComputeStatus::Configuration => f.write_str("configuration"),
-            ComputeStatus::Failed => f.write_str("failed"),
-            ComputeStatus::TerminationPending => f.write_str("termination-pending"),
-            ComputeStatus::Terminated => f.write_str("terminated"),
-        }
-    }
-}
-
 fn rfc3339_serialize<S>(x: &Option<DateTime<Utc>>, s: S) -> Result<S::Ok, S::Error>
 where
    S: Serializer,
@@ -156,15 +138,3 @@ pub enum ControlPlaneComputeStatus {
    // should be able to start with provided spec.
    Attached,
 }
-
-#[derive(Clone, Debug, Default, Serialize)]
-pub struct InstalledExtension {
-    pub extname: String,
-    pub versions: HashSet<String>,
-    pub n_databases: u32, // Number of databases using this extension
-}
-
-#[derive(Clone, Debug, Default, Serialize)]
-pub struct InstalledExtensions {
-    pub extensions: Vec<InstalledExtension>,
-}
--- a/libs/compute_api/src/spec.rs
+++ b/libs/compute_api/src/spec.rs
@@ -106,10 +106,6 @@ pub struct ComputeSpec {
    // Stripe size for pageserver sharding, in pages
    #[serde(default)]
    pub shard_stripe_size: Option<usize>,
-
-    /// Local Proxy configuration used for JWT authentication
-    #[serde(default)]
-    pub local_proxy_config: Option<LocalProxySpec>,
 }

 /// Feature flag to signal `compute_ctl` to enable certain experimental functionality.
@@ -282,13 +278,11 @@ pub struct GenericOption {
 /// declare a `trait` on it.
 pub type GenericOptions = Option<Vec<GenericOption>>;

-/// Configured the local_proxy application with the relevant JWKS and roles it should
+/// Configured the local-proxy application with the relevant JWKS and roles it should
 /// use for authorizing connect requests using JWT.
 #[derive(Clone, Debug, Deserialize, Serialize)]
 pub struct LocalProxySpec {
-    #[serde(default)]
-    #[serde(skip_serializing_if = "Option::is_none")]
-    pub jwks: Option<Vec<JwksSettings>>,
+    pub jwks: Vec<JwksSettings>,
 }

 #[derive(Clone, Debug, Deserialize, Serialize)]
--- a/libs/pageserver_api/src/config.rs
+++ b/libs/pageserver_api/src/config.rs
@@ -104,7 +104,8 @@ pub struct ConfigToml {
    pub image_compression: ImageCompressionAlgorithm,
    pub ephemeral_bytes_per_memory_kb: usize,
    pub l0_flush: Option<crate::models::L0FlushConfig>,
-    pub virtual_file_io_mode: Option<crate::models::virtual_file::IoMode>,
+    pub virtual_file_direct_io: crate::models::virtual_file::DirectIoMode,
+    pub io_buffer_alignment: usize,
 }

 #[derive(Debug, Clone, PartialEq, Eq, serde::Serialize, serde::Deserialize)]
@@ -295,14 +296,7 @@ pub mod defaults {

    pub const DEFAULT_INGEST_BATCH_SIZE: u64 = 100;

-    /// Soft limit for the maximum size of a vectored read.
-    ///
-    /// This is determined by the largest NeonWalRecord that can exist (minus dbdir and reldir keys
-    /// which are bounded by the blob io limits only). As of this writing, that is a `NeonWalRecord::ClogSetCommitted` record,
-    /// with 32k xids. That's the max number of XIDS on a single CLOG page. The size of such a record
-    /// is `sizeof(Transactionid) * 32768 + (some fixed overhead from 'timestamp`, the Vec length and whatever extra serde serialization adds)`.
-    /// That is, slightly above 128 kB.
-    pub const DEFAULT_MAX_VECTORED_READ_BYTES: usize = 130 * 1024; // 130 KiB
+    pub const DEFAULT_MAX_VECTORED_READ_BYTES: usize = 128 * 1024; // 128 KiB

    pub const DEFAULT_IMAGE_COMPRESSION: ImageCompressionAlgorithm =
        ImageCompressionAlgorithm::Zstd { level: Some(1) };
@@ -387,7 +381,10 @@ impl Default for ConfigToml {
            image_compression: (DEFAULT_IMAGE_COMPRESSION),
            ephemeral_bytes_per_memory_kb: (DEFAULT_EPHEMERAL_BYTES_PER_MEMORY_KB),
            l0_flush: None,
-            virtual_file_io_mode: None,
+            virtual_file_direct_io: crate::models::virtual_file::DirectIoMode::default(),
+
+            io_buffer_alignment: DEFAULT_IO_BUFFER_ALIGNMENT,
+
            tenant_config: TenantConfigToml::default(),
        }
    }
--- a/libs/pageserver_api/src/key.rs
+++ b/libs/pageserver_api/src/key.rs
@@ -748,16 +748,6 @@ impl Key {
        self.field1 == 0x00 && self.field4 != 0 && self.field6 != 0xffffffff
    }

-    #[inline(always)]
-    pub fn is_rel_dir_key(&self) -> bool {
-        self.field1 == 0x00
-            && self.field2 != 0
-            && self.field3 != 0
-            && self.field4 == 0
-            && self.field5 == 0
-            && self.field6 == 1
-    }
-
    /// Guaranteed to return `Ok()` if [`Self::is_rel_block_key`] returns `true` for `key`.
    #[inline(always)]
    pub fn to_rel_block(self) -> anyhow::Result<(RelTag, BlockNumber)> {
--- a/libs/pageserver_api/src/lib.rs
+++ b/libs/pageserver_api/src/lib.rs
@@ -5,11 +5,9 @@ pub mod controller_api;
 pub mod key;
 pub mod keyspace;
 pub mod models;
-pub mod record;
 pub mod reltag;
 pub mod shard;
 /// Public API types
 pub mod upcall_api;
-pub mod value;

 pub mod config;
--- a/libs/pageserver_api/src/models.rs
+++ b/libs/pageserver_api/src/models.rs
@@ -972,6 +972,8 @@ pub struct TopTenantShardsResponse {
 }

 pub mod virtual_file {
+    use std::path::PathBuf;
+
    #[derive(
        Copy,
        Clone,
@@ -992,45 +994,50 @@ pub mod virtual_file {
    }

    /// Direct IO modes for a pageserver.
-    #[derive(
-        Copy,
-        Clone,
-        PartialEq,
-        Eq,
-        Hash,
-        strum_macros::EnumString,
-        strum_macros::Display,
-        serde_with::DeserializeFromStr,
-        serde_with::SerializeDisplay,
-        Debug,
-    )]
-    #[strum(serialize_all = "kebab-case")]
-    #[repr(u8)]
-    pub enum IoMode {
-        /// Uses buffered IO.
-        Buffered,
-        /// Uses direct IO, error out if the operation fails.
-        #[cfg(target_os = "linux")]
-        Direct,
+    #[derive(Debug, PartialEq, Eq, Clone, serde::Deserialize, serde::Serialize, Default)]
+    #[serde(tag = "mode", rename_all = "kebab-case", deny_unknown_fields)]
+    pub enum DirectIoMode {
+        /// Direct IO disabled (uses usual buffered IO).
+        #[default]
+        Disabled,
+        /// Direct IO disabled (performs checks and perf simulations).
+        Evaluate {
+            /// Alignment check level
+            alignment_check: DirectIoAlignmentCheckLevel,
+            /// Latency padded for performance simulation.
+            latency_padding: DirectIoLatencyPadding,
+        },
+        /// Direct IO enabled.
+        Enabled {
+            /// Actions to perform on alignment error.
+            on_alignment_error: DirectIoOnAlignmentErrorAction,
+        },
    }

-    impl IoMode {
-        pub const fn preferred() -> Self {
-            Self::Buffered
-        }
+    #[derive(Debug, PartialEq, Eq, Clone, serde::Deserialize, serde::Serialize, Default)]
+    #[serde(rename_all = "kebab-case")]
+    pub enum DirectIoAlignmentCheckLevel {
+        #[default]
+        Error,
+        Log,
+        None,
    }

-    impl TryFrom<u8> for IoMode {
-        type Error = u8;
+    #[derive(Debug, PartialEq, Eq, Clone, serde::Deserialize, serde::Serialize, Default)]
+    #[serde(rename_all = "kebab-case")]
+    pub enum DirectIoOnAlignmentErrorAction {
+        Error,
+        #[default]
+        FallbackToBuffered,
+    }

-        fn try_from(value: u8) -> Result<Self, Self::Error> {
-            Ok(match value {
-                v if v == (IoMode::Buffered as u8) => IoMode::Buffered,
-                #[cfg(target_os = "linux")]
-                v if v == (IoMode::Direct as u8) => IoMode::Direct,
-                x => return Err(x),
-            })
-        }
+    #[derive(Debug, PartialEq, Eq, Clone, serde::Deserialize, serde::Serialize, Default)]
+    #[serde(tag = "type", rename_all = "kebab-case")]
+    pub enum DirectIoLatencyPadding {
+        /// Pad virtual file operations with IO to a fake file.
+        FakeFileRW { path: PathBuf },
+        #[default]
+        None,
    }
 }

--- a/libs/pageserver_api/src/record.rs
+++ b/libs/pageserver_api/src/record.rs
@@ -1,113 +0,0 @@
-//! This module defines the WAL record format used within the pageserver.
-
-use bytes::Bytes;
-use postgres_ffi::record::{describe_postgres_wal_record, MultiXactMember};
-use postgres_ffi::{MultiXactId, MultiXactOffset, TimestampTz, TransactionId};
-use serde::{Deserialize, Serialize};
-use utils::bin_ser::DeserializeError;
-
-/// Each update to a page is represented by a NeonWalRecord. It can be a wrapper
-/// around a PostgreSQL WAL record, or a custom neon-specific "record".
-#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)]
-pub enum NeonWalRecord {
-    /// Native PostgreSQL WAL record
-    Postgres { will_init: bool, rec: Bytes },
-
-    /// Clear bits in heap visibility map. ('flags' is bitmap of bits to clear)
-    ClearVisibilityMapFlags {
-        new_heap_blkno: Option<u32>,
-        old_heap_blkno: Option<u32>,
-        flags: u8,
-    },
-    /// Mark transaction IDs as committed on a CLOG page
-    ClogSetCommitted {
-        xids: Vec<TransactionId>,
-        timestamp: TimestampTz,
-    },
-    /// Mark transaction IDs as aborted on a CLOG page
-    ClogSetAborted { xids: Vec<TransactionId> },
-    /// Extend multixact offsets SLRU
-    MultixactOffsetCreate {
-        mid: MultiXactId,
-        moff: MultiXactOffset,
-    },
-    /// Extend multixact members SLRU.
-    MultixactMembersCreate {
-        moff: MultiXactOffset,
-        members: Vec<MultiXactMember>,
-    },
-    /// Update the map of AUX files, either writing or dropping an entry
-    AuxFile {
-        file_path: String,
-        content: Option<Bytes>,
-    },
-
-    /// A testing record for unit testing purposes. It supports append data to an existing image, or clear it.
-    #[cfg(feature = "testing")]
-    Test {
-        /// Append a string to the image.
-        append: String,
-        /// Clear the image before appending.
-        clear: bool,
-        /// Treat this record as an init record. `clear` should be set to true if this field is set
-        /// to true. This record does not need the history WALs to reconstruct. See [`NeonWalRecord::will_init`] and
-        /// its references in `timeline.rs`.
-        will_init: bool,
-    },
-}
-
-impl NeonWalRecord {
-    /// Does replaying this WAL record initialize the page from scratch, or does
-    /// it need to be applied over the previous image of the page?
-    pub fn will_init(&self) -> bool {
-        // If you change this function, you'll also need to change ValueBytes::will_init
-        match self {
-            NeonWalRecord::Postgres { will_init, rec: _ } => *will_init,
-            #[cfg(feature = "testing")]
-            NeonWalRecord::Test { will_init, .. } => *will_init,
-            // None of the special neon record types currently initialize the page
-            _ => false,
-        }
-    }
-
-    #[cfg(feature = "testing")]
-    pub fn wal_append(s: impl AsRef<str>) -> Self {
-        Self::Test {
-            append: s.as_ref().to_string(),
-            clear: false,
-            will_init: false,
-        }
-    }
-
-    #[cfg(feature = "testing")]
-    pub fn wal_clear() -> Self {
-        Self::Test {
-            append: "".to_string(),
-            clear: true,
-            will_init: false,
-        }
-    }
-
-    #[cfg(feature = "testing")]
-    pub fn wal_init() -> Self {
-        Self::Test {
-            append: "".to_string(),
-            clear: true,
-            will_init: true,
-        }
-    }
-}
-
-/// Build a human-readable string to describe a WAL record
-///
-/// For debugging purposes
-pub fn describe_wal_record(rec: &NeonWalRecord) -> Result<String, DeserializeError> {
-    match rec {
-        NeonWalRecord::Postgres { will_init, rec } => Ok(format!(
-            "will_init: {}, {}",
-            will_init,
-            describe_postgres_wal_record(rec)?
-        )),
-        _ => Ok(format!("{:?}", rec)),
-    }
-}
--- a/libs/postgres_ffi/Cargo.toml
+++ b/libs/postgres_ffi/Cargo.toml
@@ -15,7 +15,6 @@ memoffset.workspace = true
 thiserror.workspace = true
 serde.workspace = true
 utils.workspace = true
-tracing.workspace = true

 [dev-dependencies]
 env_logger.workspace = true
--- a/libs/postgres_ffi/src/lib.rs
+++ b/libs/postgres_ffi/src/lib.rs
@@ -216,7 +216,6 @@ macro_rules! enum_pgversion {
 }

 pub mod pg_constants;
-pub mod record;
 pub mod relfile_utils;

 // Export some widely used datatypes that are unlikely to change across Postgres versions
--- a/libs/remote_storage/Cargo.toml
+++ b/libs/remote_storage/Cargo.toml
@@ -16,7 +16,7 @@ aws-sdk-s3.workspace = true
 bytes.workspace = true
 camino = { workspace = true, features = ["serde1"] }
 humantime-serde.workspace = true
-hyper0 = { workspace = true, features = ["stream"] }
+hyper = { workspace = true, features = ["stream"] }
 futures.workspace = true
 serde.workspace = true
 serde_json.workspace = true
--- a/libs/remote_storage/src/azure_blob.rs
+++ b/libs/remote_storage/src/azure_blob.rs
@@ -14,7 +14,7 @@ use std::time::SystemTime;

 use super::REMOTE_STORAGE_PREFIX_SEPARATOR;
 use anyhow::Result;
-use azure_core::request_options::{IfMatchCondition, MaxResults, Metadata, Range};
+use azure_core::request_options::{MaxResults, Metadata, Range};
 use azure_core::{Continuable, RetryOptions};
 use azure_identity::DefaultAzureCredential;
 use azure_storage::StorageCredentials;
@@ -33,10 +33,10 @@ use tracing::debug;
 use utils::backoff;

 use crate::metrics::{start_measuring_requests, AttemptOutcome, RequestKind};
+use crate::ListingObject;
 use crate::{
-    config::AzureConfig, error::Cancelled, ConcurrencyLimiter, Download, DownloadError,
-    DownloadOpts, Listing, ListingMode, ListingObject, RemotePath, RemoteStorage, StorageMetadata,
-    TimeTravelError, TimeoutOrCancel,
+    config::AzureConfig, error::Cancelled, ConcurrencyLimiter, Download, DownloadError, Listing,
+    ListingMode, RemotePath, RemoteStorage, StorageMetadata, TimeTravelError, TimeoutOrCancel,
 };

 pub struct AzureBlobStorage {
@@ -259,7 +259,6 @@ fn to_download_error(error: azure_core::Error) -> DownloadError {
    if let Some(http_err) = error.as_http_error() {
        match http_err.status() {
            StatusCode::NotFound => DownloadError::NotFound,
-            StatusCode::NotModified => DownloadError::Unmodified,
            StatusCode::BadRequest => DownloadError::BadInput(anyhow::Error::new(error)),
            _ => DownloadError::Other(anyhow::Error::new(error)),
        }
@@ -485,23 +484,32 @@ impl RemoteStorage for AzureBlobStorage {
    async fn download(
        &self,
        from: &RemotePath,
-        opts: &DownloadOpts,
+        cancel: &CancellationToken,
+    ) -> Result<Download, DownloadError> {
+        let blob_client = self.client.blob_client(self.relative_path_to_name(from));
+
+        let builder = blob_client.get();
+
+        self.download_for_builder(builder, cancel).await
+    }
+
+    async fn download_byte_range(
+        &self,
+        from: &RemotePath,
+        start_inclusive: u64,
+        end_exclusive: Option<u64>,
        cancel: &CancellationToken,
    ) -> Result<Download, DownloadError> {
        let blob_client = self.client.blob_client(self.relative_path_to_name(from));

        let mut builder = blob_client.get();

-        if let Some(ref etag) = opts.etag {
-            builder = builder.if_match(IfMatchCondition::NotMatch(etag.to_string()))
-        }
-
-        if let Some((start, end)) = opts.byte_range() {
-            builder = builder.range(match end {
-                Some(end) => Range::Range(start..end),
-                None => Range::RangeFrom(start..),
-            });
-        }
+        let range: Range = if let Some(end_exclusive) = end_exclusive {
+            (start_inclusive..end_exclusive).into()
+        } else {
+            (start_inclusive..).into()
+        };
+        builder = builder.range(range);

        self.download_for_builder(builder, cancel).await
    }
--- a/libs/remote_storage/src/error.rs
+++ b/libs/remote_storage/src/error.rs
@@ -5,8 +5,6 @@ pub enum DownloadError {
    BadInput(anyhow::Error),
    /// The file was not found in the remote storage.
    NotFound,
-    /// The caller provided an ETag, and the file was not modified.
-    Unmodified,
    /// A cancellation token aborted the download, typically during
    /// tenant detach or process shutdown.
    Cancelled,
@@ -26,7 +24,6 @@ impl std::fmt::Display for DownloadError {
                write!(f, "Failed to download a remote file due to user input: {e}")
            }
            DownloadError::NotFound => write!(f, "No file found for the remote object id given"),
-            DownloadError::Unmodified => write!(f, "File was not modified"),
            DownloadError::Cancelled => write!(f, "Cancelled, shutting down"),
            DownloadError::Timeout => write!(f, "timeout"),
            DownloadError::Other(e) => write!(f, "Failed to download a remote file: {e:?}"),
@@ -41,7 +38,7 @@ impl DownloadError {
    pub fn is_permanent(&self) -> bool {
        use DownloadError::*;
        match self {
-            BadInput(_) | NotFound | Unmodified | Cancelled => true,
+            BadInput(_) | NotFound | Cancelled => true,
            Timeout | Other(_) => false,
        }
    }
--- a/libs/remote_storage/src/lib.rs
+++ b/libs/remote_storage/src/lib.rs
@@ -19,8 +19,7 @@ mod simulate_failures;
 mod support;

 use std::{
-    collections::HashMap, fmt::Debug, num::NonZeroU32, ops::Bound, pin::Pin, sync::Arc,
-    time::SystemTime,
+    collections::HashMap, fmt::Debug, num::NonZeroU32, pin::Pin, sync::Arc, time::SystemTime,
 };

 use anyhow::Context;
@@ -162,63 +161,6 @@ pub struct Listing {
    pub keys: Vec<ListingObject>,
 }

-/// Options for downloads. The default value is a plain GET.
-pub struct DownloadOpts {
-    /// If given, returns [`DownloadError::Unmodified`] if the object still has
-    /// the same ETag (using If-None-Match).
-    pub etag: Option<Etag>,
-    /// The start of the byte range to download, or unbounded.
-    pub byte_start: Bound<u64>,
-    /// The end of the byte range to download, or unbounded. Must be after the
-    /// start bound.
-    pub byte_end: Bound<u64>,
-}
-
-impl Default for DownloadOpts {
-    fn default() -> Self {
-        Self {
-            etag: Default::default(),
-            byte_start: Bound::Unbounded,
-            byte_end: Bound::Unbounded,
-        }
-    }
-}
-
-impl DownloadOpts {
-    /// Returns the byte range with inclusive start and exclusive end, or None
-    /// if unbounded.
-    pub fn byte_range(&self) -> Option<(u64, Option<u64>)> {
-        if self.byte_start == Bound::Unbounded && self.byte_end == Bound::Unbounded {
-            return None;
-        }
-        let start = match self.byte_start {
-            Bound::Excluded(i) => i + 1,
-            Bound::Included(i) => i,
-            Bound::Unbounded => 0,
-        };
-        let end = match self.byte_end {
-            Bound::Excluded(i) => Some(i),
-            Bound::Included(i) => Some(i + 1),
-            Bound::Unbounded => None,
-        };
-        if let Some(end) = end {
-            assert!(start < end, "range end {end} at or before start {start}");
-        }
-        Some((start, end))
-    }
-
-    /// Returns the byte range as an RFC 2616 Range header value with inclusive
-    /// bounds, or None if unbounded.
-    pub fn byte_range_header(&self) -> Option<String> {
-        self.byte_range()
-            .map(|(start, end)| (start, end.map(|end| end - 1))) // make end inclusive
-            .map(|(start, end)| match end {
-                Some(end) => format!("bytes={start}-{end}"),
-                None => format!("bytes={start}-"),
-            })
-    }
-}
-
 /// Storage (potentially remote) API to manage its state.
 /// This storage tries to be unaware of any layered repository context,
 /// providing basic CRUD operations for storage files.
@@ -303,7 +245,21 @@ pub trait RemoteStorage: Send + Sync + 'static {
    async fn download(
        &self,
        from: &RemotePath,
-        opts: &DownloadOpts,
+        cancel: &CancellationToken,
+    ) -> Result<Download, DownloadError>;
+
+    /// Streams a given byte range of the remote storage entry contents.
+    ///
+    /// The returned download stream will obey initial timeout and cancellation signal by erroring
+    /// on whichever happens first. Only one of the reasons will fail the stream, which is usually
+    /// enough for `tokio::io::copy_buf` usage. If needed the error can be filtered out.
+    ///
+    /// Returns the metadata, if any was stored with the file previously.
+    async fn download_byte_range(
+        &self,
+        from: &RemotePath,
+        start_inclusive: u64,
+        end_exclusive: Option<u64>,
        cancel: &CancellationToken,
    ) -> Result<Download, DownloadError>;

@@ -445,18 +401,43 @@ impl<Other: RemoteStorage> GenericRemoteStorage<Arc<Other>> {
        }
    }

-    /// See [`RemoteStorage::download`]
    pub async fn download(
        &self,
        from: &RemotePath,
-        opts: &DownloadOpts,
        cancel: &CancellationToken,
    ) -> Result<Download, DownloadError> {
        match self {
-            Self::LocalFs(s) => s.download(from, opts, cancel).await,
-            Self::AwsS3(s) => s.download(from, opts, cancel).await,
-            Self::AzureBlob(s) => s.download(from, opts, cancel).await,
-            Self::Unreliable(s) => s.download(from, opts, cancel).await,
+            Self::LocalFs(s) => s.download(from, cancel).await,
+            Self::AwsS3(s) => s.download(from, cancel).await,
+            Self::AzureBlob(s) => s.download(from, cancel).await,
+            Self::Unreliable(s) => s.download(from, cancel).await,
+        }
+    }
+
+    pub async fn download_byte_range(
+        &self,
+        from: &RemotePath,
+        start_inclusive: u64,
+        end_exclusive: Option<u64>,
+        cancel: &CancellationToken,
+    ) -> Result<Download, DownloadError> {
+        match self {
+            Self::LocalFs(s) => {
+                s.download_byte_range(from, start_inclusive, end_exclusive, cancel)
+                    .await
+            }
+            Self::AwsS3(s) => {
+                s.download_byte_range(from, start_inclusive, end_exclusive, cancel)
+                    .await
+            }
+            Self::AzureBlob(s) => {
+                s.download_byte_range(from, start_inclusive, end_exclusive, cancel)
+                    .await
+            }
+            Self::Unreliable(s) => {
+                s.download_byte_range(from, start_inclusive, end_exclusive, cancel)
+                    .await
+            }
        }
    }

@@ -581,6 +562,20 @@ impl GenericRemoteStorage {
            })
    }

+    /// Downloads the storage object into the `to_path` provided.
+    /// `byte_range` could be specified to dowload only a part of the file, if needed.
+    pub async fn download_storage_object(
+        &self,
+        byte_range: Option<(u64, Option<u64>)>,
+        from: &RemotePath,
+        cancel: &CancellationToken,
+    ) -> Result<Download, DownloadError> {
+        match byte_range {
+            Some((start, end)) => self.download_byte_range(from, start, end, cancel).await,
+            None => self.download(from, cancel).await,
+        }
+    }
+
    /// The name of the bucket/container/etc.
    pub fn bucket_name(&self) -> Option<&str> {
        match self {
@@ -654,76 +649,6 @@ impl ConcurrencyLimiter {
 mod tests {
    use super::*;

-    /// DownloadOpts::byte_range() should generate (inclusive, exclusive) ranges
-    /// with optional end bound, or None when unbounded.
-    #[test]
-    fn download_opts_byte_range() {
-        // Consider using test_case or a similar table-driven test framework.
-        let cases = [
-            // (byte_start, byte_end, expected)
-            (Bound::Unbounded, Bound::Unbounded, None),
-            (Bound::Unbounded, Bound::Included(7), Some((0, Some(8)))),
-            (Bound::Unbounded, Bound::Excluded(7), Some((0, Some(7)))),
-            (Bound::Included(3), Bound::Unbounded, Some((3, None))),
-            (Bound::Included(3), Bound::Included(7), Some((3, Some(8)))),
-            (Bound::Included(3), Bound::Excluded(7), Some((3, Some(7)))),
-            (Bound::Excluded(3), Bound::Unbounded, Some((4, None))),
-            (Bound::Excluded(3), Bound::Included(7), Some((4, Some(8)))),
-            (Bound::Excluded(3), Bound::Excluded(7), Some((4, Some(7)))),
-            // 1-sized ranges are fine, 0 aren't and will panic (separate test).
-            (Bound::Included(3), Bound::Included(3), Some((3, Some(4)))),
-            (Bound::Included(3), Bound::Excluded(4), Some((3, Some(4)))),
-        ];
-
-        for (byte_start, byte_end, expect) in cases {
-            let opts = DownloadOpts {
-                byte_start,
-                byte_end,
-                ..Default::default()
-            };
-            let result = opts.byte_range();
-            assert_eq!(
-                result, expect,
-                "byte_start={byte_start:?} byte_end={byte_end:?}"
-            );
-
-            // Check generated HTTP header, which uses an inclusive range.
-            let expect_header = expect.map(|(start, end)| match end {
-                Some(end) => format!("bytes={start}-{}", end - 1), // inclusive end
-                None => format!("bytes={start}-"),
-            });
-            assert_eq!(
-                opts.byte_range_header(),
-                expect_header,
-                "byte_start={byte_start:?} byte_end={byte_end:?}"
-            );
-        }
-    }
-
-    /// DownloadOpts::byte_range() zero-sized byte range should panic.
-    #[test]
-    #[should_panic]
-    fn download_opts_byte_range_zero() {
-        DownloadOpts {
-            byte_start: Bound::Included(3),
-            byte_end: Bound::Excluded(3),
-            ..Default::default()
-        }
-        .byte_range();
-    }
-
-    /// DownloadOpts::byte_range() negative byte range should panic.
-    #[test]
-    #[should_panic]
-    fn download_opts_byte_range_negative() {
-        DownloadOpts {
-            byte_start: Bound::Included(3),
-            byte_end: Bound::Included(2),
-            ..Default::default()
-        }
-        .byte_range();
-    }
-
    #[test]
    fn test_object_name() {
        let k = RemotePath::new(Utf8Path::new("a/b/c")).unwrap();
--- a/libs/remote_storage/src/local_fs.rs
+++ b/libs/remote_storage/src/local_fs.rs
@@ -23,8 +23,8 @@ use tokio_util::{io::ReaderStream, sync::CancellationToken};
 use utils::crashsafe::path_with_suffix_extension;

 use crate::{
-    Download, DownloadError, DownloadOpts, Listing, ListingMode, ListingObject, RemotePath,
-    TimeTravelError, TimeoutOrCancel, REMOTE_STORAGE_PREFIX_SEPARATOR,
+    Download, DownloadError, Listing, ListingMode, ListingObject, RemotePath, TimeTravelError,
+    TimeoutOrCancel, REMOTE_STORAGE_PREFIX_SEPARATOR,
 };

 use super::{RemoteStorage, StorageMetadata};
@@ -494,41 +494,22 @@ impl RemoteStorage for LocalFs {
    async fn download(
        &self,
        from: &RemotePath,
-        opts: &DownloadOpts,
        cancel: &CancellationToken,
    ) -> Result<Download, DownloadError> {
        let target_path = from.with_base(&self.storage_root);

        let file_metadata = file_metadata(&target_path).await?;
-        let etag = mock_etag(&file_metadata);

-        if opts.etag.as_ref() == Some(&etag) {
-            return Err(DownloadError::Unmodified);
-        }
-
-        let mut file = fs::OpenOptions::new()
-            .read(true)
-            .open(&target_path)
-            .await
-            .with_context(|| {
-                format!("Failed to open source file {target_path:?} to use in the download")
-            })
-            .map_err(DownloadError::Other)?;
-
-        let mut take = file_metadata.len();
-        if let Some((start, end)) = opts.byte_range() {
-            if start > 0 {
-                file.seek(io::SeekFrom::Start(start))
-                    .await
-                    .context("Failed to seek to the range start in a local storage file")
-                    .map_err(DownloadError::Other)?;
-            }
-            if let Some(end) = end {
-                take = end - start;
-            }
-        }
-
-        let source = ReaderStream::new(file.take(take));
+        let source = ReaderStream::new(
+            fs::OpenOptions::new()
+                .read(true)
+                .open(&target_path)
+                .await
+                .with_context(|| {
+                    format!("Failed to open source file {target_path:?} to use in the download")
+                })
+                .map_err(DownloadError::Other)?,
+        );

        let metadata = self
            .read_storage_metadata(&target_path)
@@ -538,6 +519,69 @@ impl RemoteStorage for LocalFs {
        let cancel_or_timeout = crate::support::cancel_or_timeout(self.timeout, cancel.clone());
        let source = crate::support::DownloadStream::new(cancel_or_timeout, source);

+        let etag = mock_etag(&file_metadata);
+        Ok(Download {
+            metadata,
+            last_modified: file_metadata
+                .modified()
+                .map_err(|e| DownloadError::Other(anyhow::anyhow!(e).context("Reading mtime")))?,
+            etag,
+            download_stream: Box::pin(source),
+        })
+    }
+
+    async fn download_byte_range(
+        &self,
+        from: &RemotePath,
+        start_inclusive: u64,
+        end_exclusive: Option<u64>,
+        cancel: &CancellationToken,
+    ) -> Result<Download, DownloadError> {
+        if let Some(end_exclusive) = end_exclusive {
+            if end_exclusive <= start_inclusive {
+                return Err(DownloadError::Other(anyhow::anyhow!("Invalid range, start ({start_inclusive}) is not less than end_exclusive ({end_exclusive:?})")));
+            };
+            if start_inclusive == end_exclusive.saturating_sub(1) {
+                return Err(DownloadError::Other(anyhow::anyhow!("Invalid range, start ({start_inclusive}) and end_exclusive ({end_exclusive:?}) difference is zero bytes")));
+            }
+        }
+
+        let target_path = from.with_base(&self.storage_root);
+        let file_metadata = file_metadata(&target_path).await?;
+        let mut source = tokio::fs::OpenOptions::new()
+            .read(true)
+            .open(&target_path)
+            .await
+            .with_context(|| {
+                format!("Failed to open source file {target_path:?} to use in the download")
+            })
+            .map_err(DownloadError::Other)?;
+
+        let len = source
+            .metadata()
+            .await
+            .context("query file length")
+            .map_err(DownloadError::Other)?
+            .len();
+
+        source
+            .seek(io::SeekFrom::Start(start_inclusive))
+            .await
+            .context("Failed to seek to the range start in a local storage file")
+            .map_err(DownloadError::Other)?;
+
+        let metadata = self
+            .read_storage_metadata(&target_path)
+            .await
+            .map_err(DownloadError::Other)?;
+
+        let source = source.take(end_exclusive.unwrap_or(len) - start_inclusive);
+        let source = ReaderStream::new(source);
+
+        let cancel_or_timeout = crate::support::cancel_or_timeout(self.timeout, cancel.clone());
+        let source = crate::support::DownloadStream::new(cancel_or_timeout, source);
+
+        let etag = mock_etag(&file_metadata);
        Ok(Download {
            metadata,
            last_modified: file_metadata
@@ -639,7 +683,7 @@ mod fs_tests {
    use super::*;

    use camino_tempfile::tempdir;
-    use std::{collections::HashMap, io::Write, ops::Bound};
+    use std::{collections::HashMap, io::Write};

    async fn read_and_check_metadata(
        storage: &LocalFs,
@@ -648,7 +692,7 @@ mod fs_tests {
    ) -> anyhow::Result<String> {
        let cancel = CancellationToken::new();
        let download = storage
-            .download(remote_storage_path, &DownloadOpts::default(), &cancel)
+            .download(remote_storage_path, &cancel)
            .await
            .map_err(|e| anyhow::anyhow!("Download failed: {e}"))?;
        ensure!(
@@ -729,8 +773,8 @@ mod fs_tests {
            "We should upload and download the same contents"
        );

-        let non_existing_path = RemotePath::new(Utf8Path::new("somewhere/else"))?;
-        match storage.download(&non_existing_path, &DownloadOpts::default(), &cancel).await {
+        let non_existing_path = "somewhere/else";
+        match storage.download(&RemotePath::new(Utf8Path::new(non_existing_path))?, &cancel).await {
            Err(DownloadError::NotFound) => {} // Should get NotFound for non existing keys
            other => panic!("Should get a NotFound error when downloading non-existing storage files, but got: {other:?}"),
        }
@@ -755,12 +799,10 @@ mod fs_tests {
        let (first_part_local, second_part_local) = uploaded_bytes.split_at(3);

        let first_part_download = storage
-            .download(
+            .download_byte_range(
                &upload_target,
-                &DownloadOpts {
-                    byte_end: Bound::Excluded(first_part_local.len() as u64),
-                    ..Default::default()
-                },
+                0,
+                Some(first_part_local.len() as u64),
                &cancel,
            )
            .await?;
@@ -776,15 +818,10 @@ mod fs_tests {
        );

        let second_part_download = storage
-            .download(
+            .download_byte_range(
                &upload_target,
-                &DownloadOpts {
-                    byte_start: Bound::Included(first_part_local.len() as u64),
-                    byte_end: Bound::Excluded(
-                        (first_part_local.len() + second_part_local.len()) as u64,
-                    ),
-                    ..Default::default()
-                },
+                first_part_local.len() as u64,
+                Some((first_part_local.len() + second_part_local.len()) as u64),
                &cancel,
            )
            .await?;
@@ -800,14 +837,7 @@ mod fs_tests {
        );

        let suffix_bytes = storage
-            .download(
-                &upload_target,
-                &DownloadOpts {
-                    byte_start: Bound::Included(13),
-                    ..Default::default()
-                },
-                &cancel,
-            )
+            .download_byte_range(&upload_target, 13, None, &cancel)
            .await?
            .download_stream;
        let suffix_bytes = aggregate(suffix_bytes).await?;
@@ -815,7 +845,7 @@ mod fs_tests {
        assert_eq!(upload_name, suffix);

        let all_bytes = storage
-            .download(&upload_target, &DownloadOpts::default(), &cancel)
+            .download_byte_range(&upload_target, 0, None, &cancel)
            .await?
            .download_stream;
        let all_bytes = aggregate(all_bytes).await?;
@@ -826,26 +856,48 @@ mod fs_tests {
    }

    #[tokio::test]
-    #[should_panic(expected = "at or before start")]
-    async fn download_file_range_negative() {
-        let (storage, cancel) = create_storage().unwrap();
+    async fn download_file_range_negative() -> anyhow::Result<()> {
+        let (storage, cancel) = create_storage()?;
        let upload_name = "upload_1";
-        let upload_target = upload_dummy_file(&storage, upload_name, None, &cancel)
-            .await
-            .unwrap();
+        let upload_target = upload_dummy_file(&storage, upload_name, None, &cancel).await?;

-        storage
-            .download(
+        let start = 1_000_000_000;
+        let end = start + 1;
+        match storage
+            .download_byte_range(
                &upload_target,
-                &DownloadOpts {
-                    byte_start: Bound::Included(10),
-                    byte_end: Bound::Excluded(10),
-                    ..Default::default()
-                },
+                start,
+                Some(end), // exclusive end
                &cancel,
            )
            .await
-            .unwrap();
+        {
+            Ok(_) => panic!("Should not allow downloading wrong ranges"),
+            Err(e) => {
+                let error_string = e.to_string();
+                assert!(error_string.contains("zero bytes"));
+                assert!(error_string.contains(&start.to_string()));
+                assert!(error_string.contains(&end.to_string()));
+            }
+        }
+
+        let start = 10000;
+        let end = 234;
+        assert!(start > end, "Should test an incorrect range");
+        match storage
+            .download_byte_range(&upload_target, start, Some(end), &cancel)
+            .await
+        {
+            Ok(_) => panic!("Should not allow downloading wrong ranges"),
+            Err(e) => {
+                let error_string = e.to_string();
+                assert!(error_string.contains("Invalid range"));
+                assert!(error_string.contains(&start.to_string()));
+                assert!(error_string.contains(&end.to_string()));
+            }
+        }
+
+        Ok(())
    }

    #[tokio::test]
@@ -888,12 +940,10 @@ mod fs_tests {
        let (first_part_local, _) = uploaded_bytes.split_at(3);

        let partial_download_with_metadata = storage
-            .download(
+            .download_byte_range(
                &upload_target,
-                &DownloadOpts {
-                    byte_end: Bound::Excluded(first_part_local.len() as u64),
-                    ..Default::default()
-                },
+                0,
+                Some(first_part_local.len() as u64),
                &cancel,
            )
            .await?;
@@ -1051,13 +1101,7 @@ mod fs_tests {
            storage.upload(body, len, &path, None, &cancel).await?;
        }

-        let read = aggregate(
-            storage
-                .download(&path, &DownloadOpts::default(), &cancel)
-                .await?
-                .download_stream,
-        )
-        .await?;
+        let read = aggregate(storage.download(&path, &cancel).await?.download_stream).await?;
        assert_eq!(body, read);

        let shorter = Bytes::from_static(b"shorter body");
@@ -1068,13 +1112,7 @@ mod fs_tests {
            storage.upload(body, len, &path, None, &cancel).await?;
        }

-        let read = aggregate(
-            storage
-                .download(&path, &DownloadOpts::default(), &cancel)
-                .await?
-                .download_stream,
-        )
-        .await?;
+        let read = aggregate(storage.download(&path, &cancel).await?.download_stream).await?;
        assert_eq!(shorter, read);
        Ok(())
    }
@@ -1107,13 +1145,7 @@ mod fs_tests {
            storage.upload(body, len, &path, None, &cancel).await?;
        }

-        let read = aggregate(
-            storage
-                .download(&path, &DownloadOpts::default(), &cancel)
-                .await?
-                .download_stream,
-        )
-        .await?;
+        let read = aggregate(storage.download(&path, &cancel).await?.download_stream).await?;
        assert_eq!(body, read);

        Ok(())
--- a/libs/remote_storage/src/s3_bucket.rs
+++ b/libs/remote_storage/src/s3_bucket.rs
@@ -28,13 +28,12 @@ use aws_sdk_s3::{
    Client,
 };
 use aws_smithy_async::rt::sleep::TokioSleep;
-use http_types::StatusCode;

 use aws_smithy_types::{body::SdkBody, DateTime};
 use aws_smithy_types::{byte_stream::ByteStream, date_time::ConversionError};
 use bytes::Bytes;
 use futures::stream::Stream;
-use hyper0::Body;
+use hyper::Body;
 use scopeguard::ScopeGuard;
 use tokio_util::sync::CancellationToken;
 use utils::backoff;
@@ -45,8 +44,8 @@ use crate::{
    error::Cancelled,
    metrics::{start_counting_cancelled_wait, start_measuring_requests},
    support::PermitCarrying,
-    ConcurrencyLimiter, Download, DownloadError, DownloadOpts, Listing, ListingMode, ListingObject,
-    RemotePath, RemoteStorage, TimeTravelError, TimeoutOrCancel, MAX_KEYS_PER_DELETE,
+    ConcurrencyLimiter, Download, DownloadError, Listing, ListingMode, ListingObject, RemotePath,
+    RemoteStorage, TimeTravelError, TimeoutOrCancel, MAX_KEYS_PER_DELETE,
    REMOTE_STORAGE_PREFIX_SEPARATOR,
 };

@@ -68,7 +67,6 @@ pub struct S3Bucket {
 struct GetObjectRequest {
    bucket: String,
    key: String,
-    etag: Option<String>,
    range: Option<String>,
 }
 impl S3Bucket {
@@ -250,18 +248,13 @@ impl S3Bucket {

        let started_at = start_measuring_requests(kind);

-        let mut builder = self
+        let get_object = self
            .client
            .get_object()
            .bucket(request.bucket)
            .key(request.key)
-            .set_range(request.range);
-
-        if let Some(etag) = request.etag {
-            builder = builder.if_none_match(etag);
-        }
-
-        let get_object = builder.send();
+            .set_range(request.range)
+            .send();

        let get_object = tokio::select! {
            res = get_object => res,
@@ -284,20 +277,6 @@ impl S3Bucket {
                );
                return Err(DownloadError::NotFound);
            }
-            Err(SdkError::ServiceError(e))
-                // aws_smithy_runtime_api::http::response::StatusCode isn't
-                // re-exported by any aws crates, so just check the numeric
-                // status against http_types::StatusCode instead of pulling it.
-                if e.raw().status().as_u16() == StatusCode::NotModified =>
-            {
-                // Count an unmodified file as a success.
-                crate::metrics::BUCKET_METRICS.req_seconds.observe_elapsed(
-                    kind,
-                    AttemptOutcome::Ok,
-                    started_at,
-                );
-                return Err(DownloadError::Unmodified);
-            }
            Err(e) => {
                crate::metrics::BUCKET_METRICS.req_seconds.observe_elapsed(
                    kind,
@@ -794,7 +773,6 @@ impl RemoteStorage for S3Bucket {
    async fn download(
        &self,
        from: &RemotePath,
-        opts: &DownloadOpts,
        cancel: &CancellationToken,
    ) -> Result<Download, DownloadError> {
        // if prefix is not none then download file `prefix/from`
@@ -803,8 +781,33 @@ impl RemoteStorage for S3Bucket {
            GetObjectRequest {
                bucket: self.bucket_name.clone(),
                key: self.relative_path_to_s3_object(from),
-                etag: opts.etag.as_ref().map(|e| e.to_string()),
-                range: opts.byte_range_header(),
+                range: None,
+            },
+            cancel,
+        )
+        .await
+    }
+
+    async fn download_byte_range(
+        &self,
+        from: &RemotePath,
+        start_inclusive: u64,
+        end_exclusive: Option<u64>,
+        cancel: &CancellationToken,
+    ) -> Result<Download, DownloadError> {
+        // S3 accepts ranges as https://www.w3.org/Protocols/rfc2616/rfc2616-sec14.html#sec14.35
+        // and needs both ends to be exclusive
+        let end_inclusive = end_exclusive.map(|end| end.saturating_sub(1));
+        let range = Some(match end_inclusive {
+            Some(end_inclusive) => format!("bytes={start_inclusive}-{end_inclusive}"),
+            None => format!("bytes={start_inclusive}-"),
+        });
+
+        self.download_object(
+            GetObjectRequest {
+                bucket: self.bucket_name.clone(),
+                key: self.relative_path_to_s3_object(from),
+                range,
            },
            cancel,
        )
--- a/libs/remote_storage/src/simulate_failures.rs
+++ b/libs/remote_storage/src/simulate_failures.rs
@@ -12,8 +12,8 @@ use std::{collections::hash_map::Entry, sync::Arc};
 use tokio_util::sync::CancellationToken;

 use crate::{
-    Download, DownloadError, DownloadOpts, GenericRemoteStorage, Listing, ListingMode, RemotePath,
-    RemoteStorage, StorageMetadata, TimeTravelError,
+    Download, DownloadError, GenericRemoteStorage, Listing, ListingMode, RemotePath, RemoteStorage,
+    StorageMetadata, TimeTravelError,
 };

 pub struct UnreliableWrapper {
@@ -167,14 +167,28 @@ impl RemoteStorage for UnreliableWrapper {
    async fn download(
        &self,
        from: &RemotePath,
-        opts: &DownloadOpts,
        cancel: &CancellationToken,
    ) -> Result<Download, DownloadError> {
-        // Note: We treat any byte range as an "attempt" of the same operation.
-        // We don't pay attention to the ranges. That's good enough for now.
        self.attempt(RemoteOp::Download(from.clone()))
            .map_err(DownloadError::Other)?;
-        self.inner.download(from, opts, cancel).await
+        self.inner.download(from, cancel).await
+    }
+
+    async fn download_byte_range(
+        &self,
+        from: &RemotePath,
+        start_inclusive: u64,
+        end_exclusive: Option<u64>,
+        cancel: &CancellationToken,
+    ) -> Result<Download, DownloadError> {
+        // Note: We treat any download_byte_range as an "attempt" of the same
+        // operation. We don't pay attention to the ranges. That's good enough
+        // for now.
+        self.attempt(RemoteOp::Download(from.clone()))
+            .map_err(DownloadError::Other)?;
+        self.inner
+            .download_byte_range(from, start_inclusive, end_exclusive, cancel)
+            .await
    }

    async fn delete(&self, path: &RemotePath, cancel: &CancellationToken) -> anyhow::Result<()> {
--- a/libs/remote_storage/tests/common/tests.rs
+++ b/libs/remote_storage/tests/common/tests.rs
@@ -1,8 +1,8 @@
 use anyhow::Context;
 use camino::Utf8Path;
 use futures::StreamExt;
-use remote_storage::{DownloadError, DownloadOpts, ListingMode, ListingObject, RemotePath};
-use std::ops::Bound;
+use remote_storage::ListingMode;
+use remote_storage::RemotePath;
 use std::sync::Arc;
 use std::{collections::HashSet, num::NonZeroU32};
 use test_context::test_context;
@@ -284,25 +284,14 @@ async fn upload_download_works(ctx: &mut MaybeEnabledStorage) -> anyhow::Result<
    ctx.client.upload(data, len, &path, None, &cancel).await?;

    // Normal download request
-    let dl = ctx
-        .client
-        .download(&path, &DownloadOpts::default(), &cancel)
-        .await?;
+    let dl = ctx.client.download(&path, &cancel).await?;
    let buf = download_to_vec(dl).await?;
    assert_eq!(&buf, &orig);

    // Full range (end specified)
    let dl = ctx
        .client
-        .download(
-            &path,
-            &DownloadOpts {
-                byte_start: Bound::Included(0),
-                byte_end: Bound::Excluded(len as u64),
-                ..Default::default()
-            },
-            &cancel,
-        )
+        .download_byte_range(&path, 0, Some(len as u64), &cancel)
        .await?;
    let buf = download_to_vec(dl).await?;
    assert_eq!(&buf, &orig);
@@ -310,15 +299,7 @@ async fn upload_download_works(ctx: &mut MaybeEnabledStorage) -> anyhow::Result<
    // partial range (end specified)
    let dl = ctx
        .client
-        .download(
-            &path,
-            &DownloadOpts {
-                byte_start: Bound::Included(4),
-                byte_end: Bound::Excluded(10),
-                ..Default::default()
-            },
-            &cancel,
-        )
+        .download_byte_range(&path, 4, Some(10), &cancel)
        .await?;
    let buf = download_to_vec(dl).await?;
    assert_eq!(&buf, &orig[4..10]);
@@ -326,15 +307,7 @@ async fn upload_download_works(ctx: &mut MaybeEnabledStorage) -> anyhow::Result<
    // partial range (end beyond real end)
    let dl = ctx
        .client
-        .download(
-            &path,
-            &DownloadOpts {
-                byte_start: Bound::Included(8),
-                byte_end: Bound::Excluded(len as u64 * 100),
-                ..Default::default()
-            },
-            &cancel,
-        )
+        .download_byte_range(&path, 8, Some(len as u64 * 100), &cancel)
        .await?;
    let buf = download_to_vec(dl).await?;
    assert_eq!(&buf, &orig[8..]);
@@ -342,14 +315,7 @@ async fn upload_download_works(ctx: &mut MaybeEnabledStorage) -> anyhow::Result<
    // Partial range (end unspecified)
    let dl = ctx
        .client
-        .download(
-            &path,
-            &DownloadOpts {
-                byte_start: Bound::Included(4),
-                ..Default::default()
-            },
-            &cancel,
-        )
+        .download_byte_range(&path, 4, None, &cancel)
        .await?;
    let buf = download_to_vec(dl).await?;
    assert_eq!(&buf, &orig[4..]);
@@ -357,14 +323,7 @@ async fn upload_download_works(ctx: &mut MaybeEnabledStorage) -> anyhow::Result<
    // Full range (end unspecified)
    let dl = ctx
        .client
-        .download(
-            &path,
-            &DownloadOpts {
-                byte_start: Bound::Included(0),
-                ..Default::default()
-            },
-            &cancel,
-        )
+        .download_byte_range(&path, 0, None, &cancel)
        .await?;
    let buf = download_to_vec(dl).await?;
    assert_eq!(&buf, &orig);
@@ -378,54 +337,6 @@ async fn upload_download_works(ctx: &mut MaybeEnabledStorage) -> anyhow::Result<
    Ok(())
 }

-/// Tests that conditional downloads work properly, by returning
-/// DownloadError::Unmodified when the object ETag matches the given ETag.
-#[test_context(MaybeEnabledStorage)]
-#[tokio::test]
-async fn download_conditional(ctx: &mut MaybeEnabledStorage) -> anyhow::Result<()> {
-    let MaybeEnabledStorage::Enabled(ctx) = ctx else {
-        return Ok(());
-    };
-    let cancel = CancellationToken::new();
-
-    // Create a file.
-    let path = RemotePath::new(Utf8Path::new(format!("{}/file", ctx.base_prefix).as_str()))?;
-    let data = bytes::Bytes::from_static("foo".as_bytes());
-    let (stream, len) = wrap_stream(data);
-    ctx.client.upload(stream, len, &path, None, &cancel).await?;
-
-    // Download it to obtain its etag.
-    let mut opts = DownloadOpts::default();
-    let download = ctx.client.download(&path, &opts, &cancel).await?;
-
-    // Download with the etag yields DownloadError::Unmodified.
-    opts.etag = Some(download.etag);
-    let result = ctx.client.download(&path, &opts, &cancel).await;
-    assert!(
-        matches!(result, Err(DownloadError::Unmodified)),
-        "expected DownloadError::Unmodified, got {result:?}"
-    );
-
-    // Replace the file contents.
-    let data = bytes::Bytes::from_static("bar".as_bytes());
-    let (stream, len) = wrap_stream(data);
-    ctx.client.upload(stream, len, &path, None, &cancel).await?;
-
-    // A download with the old etag should yield the new file.
-    let download = ctx.client.download(&path, &opts, &cancel).await?;
-    assert_ne!(download.etag, opts.etag.unwrap(), "ETag did not change");
-
-    // A download with the new etag should yield Unmodified again.
-    opts.etag = Some(download.etag);
-    let result = ctx.client.download(&path, &opts, &cancel).await;
-    assert!(
-        matches!(result, Err(DownloadError::Unmodified)),
-        "expected DownloadError::Unmodified, got {result:?}"
-    );
-
-    Ok(())
-}
-
 #[test_context(MaybeEnabledStorage)]
 #[tokio::test]
 async fn copy_works(ctx: &mut MaybeEnabledStorage) -> anyhow::Result<()> {
@@ -453,10 +364,7 @@ async fn copy_works(ctx: &mut MaybeEnabledStorage) -> anyhow::Result<()> {
    // Normal download request
    ctx.client.copy_object(&path, &path_dest, &cancel).await?;

-    let dl = ctx
-        .client
-        .download(&path_dest, &DownloadOpts::default(), &cancel)
-        .await?;
+    let dl = ctx.client.download(&path_dest, &cancel).await?;
    let buf = download_to_vec(dl).await?;
    assert_eq!(&buf, &orig);

@@ -468,56 +376,3 @@ async fn copy_works(ctx: &mut MaybeEnabledStorage) -> anyhow::Result<()> {

    Ok(())
 }
-
-/// Tests that head_object works properly.
-#[test_context(MaybeEnabledStorage)]
-#[tokio::test]
-async fn head_object(ctx: &mut MaybeEnabledStorage) -> anyhow::Result<()> {
-    let MaybeEnabledStorage::Enabled(ctx) = ctx else {
-        return Ok(());
-    };
-    let cancel = CancellationToken::new();
-
-    let path = RemotePath::new(Utf8Path::new(format!("{}/file", ctx.base_prefix).as_str()))?;
-
-    // Errors on missing file.
-    let result = ctx.client.head_object(&path, &cancel).await;
-    assert!(
-        matches!(result, Err(DownloadError::NotFound)),
-        "expected NotFound, got {result:?}"
-    );
-
-    // Create the file.
-    let data = bytes::Bytes::from_static("foo".as_bytes());
-    let (stream, len) = wrap_stream(data);
-    ctx.client.upload(stream, len, &path, None, &cancel).await?;
-
-    // Fetch the head metadata.
-    let object = ctx.client.head_object(&path, &cancel).await?;
-    assert_eq!(
-        object,
-        ListingObject {
-            key: path.clone(),
-            last_modified: object.last_modified, // ignore
-            size: 3
-        }
-    );
-
-    // Wait for a couple of seconds, and then update the file to check the last
-    // modified timestamp.
-    tokio::time::sleep(std::time::Duration::from_secs(2)).await;
-
-    let data = bytes::Bytes::from_static("bar".as_bytes());
-    let (stream, len) = wrap_stream(data);
-    ctx.client.upload(stream, len, &path, None, &cancel).await?;
-    let new = ctx.client.head_object(&path, &cancel).await?;
-
-    assert!(
-        !new.last_modified
-            .duration_since(object.last_modified)?
-            .is_zero(),
-        "last_modified did not advance"
-    );
-
-    Ok(())
-}
--- a/libs/remote_storage/tests/test_real_s3.rs
+++ b/libs/remote_storage/tests/test_real_s3.rs
@@ -12,8 +12,8 @@ use anyhow::Context;
 use camino::Utf8Path;
 use futures_util::StreamExt;
 use remote_storage::{
-    DownloadError, DownloadOpts, GenericRemoteStorage, ListingMode, RemotePath,
-    RemoteStorageConfig, RemoteStorageKind, S3Config,
+    DownloadError, GenericRemoteStorage, ListingMode, RemotePath, RemoteStorageConfig,
+    RemoteStorageKind, S3Config,
 };
 use test_context::test_context;
 use test_context::AsyncTestContext;
@@ -121,8 +121,7 @@ async fn s3_time_travel_recovery_works(ctx: &mut MaybeEnabledStorage) -> anyhow:

    // A little check to ensure that our clock is not too far off from the S3 clock
    {
-        let opts = DownloadOpts::default();
-        let dl = retry(|| ctx.client.download(&path2, &opts, &cancel)).await?;
+        let dl = retry(|| ctx.client.download(&path2, &cancel)).await?;
        let last_modified = dl.last_modified;
        let half_wt = WAIT_TIME.mul_f32(0.5);
        let t0_hwt = t0 + half_wt;
@@ -160,12 +159,7 @@ async fn s3_time_travel_recovery_works(ctx: &mut MaybeEnabledStorage) -> anyhow:
    let t2_files_recovered = list_files(&ctx.client, &cancel).await?;
    println!("after recovery to t2: {t2_files_recovered:?}");
    assert_eq!(t2_files, t2_files_recovered);
-    let path2_recovered_t2 = download_to_vec(
-        ctx.client
-            .download(&path2, &DownloadOpts::default(), &cancel)
-            .await?,
-    )
-    .await?;
+    let path2_recovered_t2 = download_to_vec(ctx.client.download(&path2, &cancel).await?).await?;
    assert_eq!(path2_recovered_t2, new_data.as_bytes());

    // after recovery to t1: path1 is back, path2 has the old content
@@ -176,12 +170,7 @@ async fn s3_time_travel_recovery_works(ctx: &mut MaybeEnabledStorage) -> anyhow:
    let t1_files_recovered = list_files(&ctx.client, &cancel).await?;
    println!("after recovery to t1: {t1_files_recovered:?}");
    assert_eq!(t1_files, t1_files_recovered);
-    let path2_recovered_t1 = download_to_vec(
-        ctx.client
-            .download(&path2, &DownloadOpts::default(), &cancel)
-            .await?,
-    )
-    .await?;
+    let path2_recovered_t1 = download_to_vec(ctx.client.download(&path2, &cancel).await?).await?;
    assert_eq!(path2_recovered_t1, old_data.as_bytes());

    // after recovery to t0: everything is gone except for path1
@@ -427,7 +416,7 @@ async fn download_is_timeouted(ctx: &mut MaybeEnabledStorage) {
    let started_at = std::time::Instant::now();
    let mut stream = ctx
        .client
-        .download(&path, &DownloadOpts::default(), &cancel)
+        .download(&path, &cancel)
        .await
        .expect("download succeeds")
        .download_stream;
@@ -502,7 +491,7 @@ async fn download_is_cancelled(ctx: &mut MaybeEnabledStorage) {
    {
        let stream = ctx
            .client
-            .download(&path, &DownloadOpts::default(), &cancel)
+            .download(&path, &cancel)
            .await
            .expect("download succeeds")
            .download_stream;
--- a/libs/tracing-utils/Cargo.toml
+++ b/libs/tracing-utils/Cargo.toml
@@ -5,7 +5,7 @@ edition.workspace = true
 license.workspace = true

 [dependencies]
-hyper0.workspace = true
+hyper.workspace = true
 opentelemetry = { workspace = true, features = ["trace"] }
 opentelemetry_sdk = { workspace = true, features = ["rt-tokio"] }
 opentelemetry-otlp = { workspace = true, default-features = false, features = ["http-proto", "trace", "http", "reqwest-client"] }
--- a/libs/tracing-utils/src/http.rs
+++ b/libs/tracing-utils/src/http.rs
@@ -1,7 +1,7 @@
 //! Tracing wrapper for Hyper HTTP server

-use hyper0::HeaderMap;
-use hyper0::{Body, Request, Response};
+use hyper::HeaderMap;
+use hyper::{Body, Request, Response};
 use std::future::Future;
 use tracing::Instrument;
 use tracing_opentelemetry::OpenTelemetrySpanExt;
--- a/libs/utils/Cargo.toml
+++ b/libs/utils/Cargo.toml
@@ -22,7 +22,7 @@ chrono.workspace = true
 git-version.workspace = true
 hex = { workspace = true, features = ["serde"] }
 humantime.workspace = true
-hyper0 = { workspace = true, features = ["full"] }
+hyper = { workspace = true, features = ["full"] }
 fail.workspace = true
 futures = { workspace = true}
 jsonwebtoken.workspace = true
--- a/libs/utils/src/auth.rs
+++ b/libs/utils/src/auth.rs
@@ -31,12 +31,9 @@ pub enum Scope {
    /// The scope used by pageservers in upcalls to storage controller and cloud control plane
    #[serde(rename = "generations_api")]
    GenerationsApi,
-    /// Allows access to control plane managment API and all storage controller endpoints.
+    /// Allows access to control plane managment API and some storage controller endpoints.
    Admin,

-    /// Allows access to control plane & storage controller endpoints used in infrastructure automation (e.g. node registration)
-    Infra,
-
    /// Allows access to storage controller APIs used by the scrubber, to interrogate the state
    /// of a tenant & post scrub results.
    Scrubber,
--- a/libs/utils/src/http/error.rs
+++ b/libs/utils/src/http/error.rs
@@ -28,9 +28,6 @@ pub enum ApiError {
    #[error("Resource temporarily unavailable: {0}")]
    ResourceUnavailable(Cow<'static, str>),

-    #[error("Too many requests: {0}")]
-    TooManyRequests(Cow<'static, str>),
-
    #[error("Shutting down")]
    ShuttingDown,

@@ -76,10 +73,6 @@ impl ApiError {
                err.to_string(),
                StatusCode::SERVICE_UNAVAILABLE,
            ),
-            ApiError::TooManyRequests(err) => HttpErrorBody::response_from_msg_and_status(
-                err.to_string(),
-                StatusCode::TOO_MANY_REQUESTS,
-            ),
            ApiError::Timeout(err) => HttpErrorBody::response_from_msg_and_status(
                err.to_string(),
                StatusCode::REQUEST_TIMEOUT,
--- a/libs/utils/src/lib.rs
+++ b/libs/utils/src/lib.rs
@@ -2,8 +2,6 @@
 //! between other crates in this repository.
 #![deny(clippy::undocumented_unsafe_blocks)]

-extern crate hyper0 as hyper;
-
 pub mod backoff;

 /// `Lsn` type implements common tasks on Log Sequence Numbers
--- a/libs/vm_monitor/src/lib.rs
+++ b/libs/vm_monitor/src/lib.rs
@@ -7,13 +7,11 @@ use axum::{
    extract::{ws::WebSocket, State, WebSocketUpgrade},
    response::Response,
 };
-use axum::{routing::get, Router};
+use axum::{routing::get, Router, Server};
 use clap::Parser;
 use futures::Future;
-use std::net::SocketAddr;
 use std::{fmt::Debug, time::Duration};
 use sysinfo::{RefreshKind, System, SystemExt};
-use tokio::net::TcpListener;
 use tokio::{sync::broadcast, task::JoinHandle};
 use tokio_util::sync::CancellationToken;
 use tracing::{error, info};
@@ -134,14 +132,14 @@ pub async fn start(args: &'static Args, token: CancellationToken) -> anyhow::Res
            args,
        });

-    let addr_str = args.addr();
-    let addr: SocketAddr = addr_str.parse().expect("parsing address should not fail");
-
-    let listener = TcpListener::bind(&addr)
-        .await
+    let addr = args.addr();
+    let bound = Server::try_bind(&addr.parse().expect("parsing address should not fail"))
        .with_context(|| format!("failed to bind to {addr}"))?;
-    info!(addr_str, "server bound");
-    axum::serve(listener, app.into_make_service())
+
+    info!(addr, "server bound");
+
+    bound
+        .serve(app.into_make_service())
        .await
        .context("server exited")?;

--- a/libs/vm_monitor/src/runner.rs
+++ b/libs/vm_monitor/src/runner.rs
@@ -79,7 +79,8 @@ pub struct Config {
    /// memory.
    ///
    /// The default value of `0.15` means that we *guarantee* sending upscale requests if the
-    /// cgroup is using more than 85% of total memory.
+    /// cgroup is using more than 85% of total memory (even if we're *not* separately reserving
+    /// memory for the file cache).
    cgroup_min_overhead_fraction: f64,

    cgroup_downscale_threshold_buffer_bytes: u64,
@@ -96,12 +97,24 @@ impl Default for Config {
 }

 impl Config {
-    fn cgroup_threshold(&self, total_mem: u64) -> u64 {
-        // We want our threshold to be met gracefully instead of letting postgres get OOM-killed
-        // (or if there's room, spilling to swap).
+    fn cgroup_threshold(&self, total_mem: u64, file_cache_disk_size: u64) -> u64 {
+        // If the file cache is in tmpfs, then it will count towards shmem usage of the cgroup,
+        // and thus be non-reclaimable, so we should allow for additional memory usage.
+        //
+        // If the file cache sits on disk, our desired stable system state is for it to be fully
+        // page cached (its contents should only be paged to/from disk in situations where we can't
+        // upscale fast enough). Page-cached memory is reclaimable, so we need to lower the
+        // threshold for non-reclaimable memory so we scale up *before* the kernel starts paging
+        // out the file cache.
+        let memory_remaining_for_cgroup = total_mem.saturating_sub(file_cache_disk_size);
+
+        // Even if we're not separately making room for the file cache (if it's in tmpfs), we still
+        // want our threshold to be met gracefully instead of letting postgres get OOM-killed.
        // So we guarantee that there's at least `cgroup_min_overhead_fraction` of total memory
        // remaining above the threshold.
-        (total_mem as f64 * (1.0 - self.cgroup_min_overhead_fraction)) as u64
+        let max_threshold = (total_mem as f64 * (1.0 - self.cgroup_min_overhead_fraction)) as u64;
+
+        memory_remaining_for_cgroup.min(max_threshold)
    }
 }

@@ -136,6 +149,11 @@ impl Runner {

        let mem = get_total_system_memory();

+        let mut file_cache_disk_size = 0;
+
+        // We need to process file cache initialization before cgroup initialization, so that the memory
+        // allocated to the file cache is appropriately taken into account when we decide the cgroup's
+        // memory limits.
        if let Some(connstr) = &args.pgconnstr {
            info!("initializing file cache");
            let config = FileCacheConfig::default();
@@ -166,6 +184,7 @@ impl Runner {
                info!("file cache size actually got set to {actual_size}")
            }

+            file_cache_disk_size = actual_size;
            state.filecache = Some(file_cache);
        }

@@ -188,7 +207,7 @@ impl Runner {
                cgroup.watch(hist_tx).await
            });

-            let threshold = state.config.cgroup_threshold(mem);
+            let threshold = state.config.cgroup_threshold(mem, file_cache_disk_size);
            info!(threshold, "set initial cgroup threshold",);

            state.cgroup = Some(CgroupState {
@@ -240,7 +259,9 @@ impl Runner {
                return Ok((false, status.to_owned()));
            }

-            let new_threshold = self.config.cgroup_threshold(usable_system_memory);
+            let new_threshold = self
+                .config
+                .cgroup_threshold(usable_system_memory, expected_file_cache_size);

            let current = last_history.avg_non_reclaimable;

@@ -261,11 +282,13 @@ impl Runner {

        // The downscaling has been approved. Downscale the file cache, then the cgroup.
        let mut status = vec![];
+        let mut file_cache_disk_size = 0;
        if let Some(file_cache) = &mut self.filecache {
            let actual_usage = file_cache
                .set_file_cache_size(expected_file_cache_size)
                .await
                .context("failed to set file cache size")?;
+            file_cache_disk_size = actual_usage;
            let message = format!(
                "set file cache size to {} MiB",
                bytes_to_mebibytes(actual_usage),
@@ -275,7 +298,9 @@ impl Runner {
        }

        if let Some(cgroup) = &mut self.cgroup {
-            let new_threshold = self.config.cgroup_threshold(usable_system_memory);
+            let new_threshold = self
+                .config
+                .cgroup_threshold(usable_system_memory, file_cache_disk_size);

            let message = format!(
                "set cgroup memory threshold from {} MiB to {} MiB, of new total {} MiB",
@@ -304,6 +329,7 @@ impl Runner {
        let new_mem = resources.mem;
        let usable_system_memory = new_mem.saturating_sub(self.config.sys_buffer_bytes);

+        let mut file_cache_disk_size = 0;
        if let Some(file_cache) = &mut self.filecache {
            let expected_usage = file_cache.config.calculate_cache_size(usable_system_memory);
            info!(
@@ -316,6 +342,7 @@ impl Runner {
                .set_file_cache_size(expected_usage)
                .await
                .context("failed to set file cache size")?;
+            file_cache_disk_size = actual_usage;

            if actual_usage != expected_usage {
                warn!(
@@ -327,7 +354,9 @@ impl Runner {
        }

        if let Some(cgroup) = &mut self.cgroup {
-            let new_threshold = self.config.cgroup_threshold(usable_system_memory);
+            let new_threshold = self
+                .config
+                .cgroup_threshold(usable_system_memory, file_cache_disk_size);

            info!(
                "set cgroup memory threshold from {} MiB to {} MiB of new total {} MiB",
--- a/libs/wal_decoder/Cargo.toml
+++ b/libs/wal_decoder/Cargo.toml
@@ -1,14 +0,0 @@
-[package]
-name = "wal_decoder"
-version = "0.1.0"
-edition.workspace = true
-license.workspace = true
-
-[dependencies]
-anyhow.workspace = true
-bytes.workspace = true
-pageserver_api.workspace = true
-postgres_ffi.workspace = true
-serde.workspace = true
-tracing.workspace = true
-utils.workspace = true
--- a/libs/wal_decoder/src/decoder.rs
+++ b/libs/wal_decoder/src/decoder.rs
--- a/libs/wal_decoder/src/lib.rs
+++ b/libs/wal_decoder/src/lib.rs
@@ -1,2 +0,0 @@
-pub mod decoder;
-pub mod models;
--- a/libs/wal_decoder/src/models.rs
+++ b/libs/wal_decoder/src/models.rs
@@ -1,177 +0,0 @@
-//! This module houses types which represent decoded PG WAL records
-//! ready for the pageserver to interpret.
-
-use bytes::Bytes;
-use pageserver_api::key::CompactKey;
-use pageserver_api::reltag::{BlockNumber, RelTag, SlruKind};
-use pageserver_api::value::Value;
-use postgres_ffi::record::{
-    XlMultiXactCreate, XlMultiXactTruncate, XlRelmapUpdate, XlReploriginDrop, XlReploriginSet,
-    XlXactParsedRecord,
-};
-use postgres_ffi::{Oid, TransactionId};
-use utils::lsn::Lsn;
-
-pub enum FlushUncommittedRecords {
-    Yes,
-    No,
-}
-
-pub struct InterpretedWalRecord {
-    pub metadata_record: Option<MetadataRecord>,
-    pub blocks: Vec<(CompactKey, Option<Value>)>,
-    pub lsn: Lsn,
-    pub flush_uncommitted: FlushUncommittedRecords,
-    pub xid: TransactionId,
-}
-
-pub enum MetadataRecord {
-    Heapam(HeapamRecord),
-    Neonrmgr(NeonrmgrRecord),
-    Smgr(SmgrRecord),
-    Dbase(DbaseRecord),
-    Clog(ClogRecord),
-    Xact(XactRecord),
-    MultiXact(MultiXactRecord),
-    Relmap(RelmapRecord),
-    Xlog(XlogRecord),
-    LogicalMessage(LogicalMessageRecord),
-    Standby(StandbyRecord),
-    Replorigin(ReploriginRecord),
-}
-
-pub enum HeapamRecord {
-    ClearVmBits(ClearVmBits),
-}
-
-pub struct ClearVmBits {
-    pub new_heap_blkno: Option<u32>,
-    pub old_heap_blkno: Option<u32>,
-    pub vm_rel: RelTag,
-    pub flags: u8,
-}
-
-pub enum NeonrmgrRecord {
-    ClearVmBits(ClearVmBits),
-}
-
-pub enum SmgrRecord {
-    Create(SmgrCreate),
-    Truncate(SmgrTruncate),
-}
-
-pub struct SmgrCreate {
-    pub rel: RelTag,
-}
-
-pub struct SmgrTruncate {
-    pub rel: RelTag,
-    pub to: BlockNumber,
-}
-
-pub enum DbaseRecord {
-    Create(DbaseCreate),
-    Drop(DbaseDrop),
-}
-
-pub struct DbaseCreate {
-    pub db_id: Oid,
-    pub tablespace_id: Oid,
-    pub src_db_id: Oid,
-    pub src_tablespace_id: Oid,
-}
-
-pub struct DbaseDrop {
-    pub db_id: Oid,
-    pub tablespace_ids: Vec<Oid>,
-}
-
-pub enum ClogRecord {
-    ZeroPage(ClogZeroPage),
-    Truncate(ClogTruncate),
-}
-
-pub struct ClogZeroPage {
-    pub segno: u32,
-    pub rpageno: u32,
-}
-
-pub struct ClogTruncate {
-    pub pageno: u32,
-    pub oldest_xid: TransactionId,
-    pub oldest_xid_db: Oid,
-}
-
-pub enum XactRecord {
-    Commit(XactCommon),
-    Abort(XactCommon),
-    CommitPrepared(XactCommon),
-    AbortPrepared(XactCommon),
-    Prepare(XactPrepare),
-}
-
-pub struct XactCommon {
-    pub parsed: XlXactParsedRecord,
-    pub origin_id: u16,
-    // Fields below are only used for logging
-    pub xl_xid: TransactionId,
-    pub lsn: Lsn,
-}
-
-pub struct XactPrepare {
-    pub xl_xid: TransactionId,
-    pub data: Bytes,
-}
-
-pub enum MultiXactRecord {
-    ZeroPage(MultiXactZeroPage),
-    Create(XlMultiXactCreate),
-    Truncate(XlMultiXactTruncate),
-}
-
-pub struct MultiXactZeroPage {
-    pub slru_kind: SlruKind,
-    pub segno: u32,
-    pub rpageno: u32,
-}
-
-pub enum RelmapRecord {
-    Update(RelmapUpdate),
-}
-
-pub struct RelmapUpdate {
-    pub update: XlRelmapUpdate,
-    pub buf: Bytes,
-}
-
-pub enum XlogRecord {
-    Raw(RawXlogRecord),
-}
-
-pub struct RawXlogRecord {
-    pub info: u8,
-    pub lsn: Lsn,
-    pub buf: Bytes,
-}
-
-pub enum LogicalMessageRecord {
-    Put(PutLogicalMessage),
-}
-
-pub struct PutLogicalMessage {
-    pub buf: Bytes,
-    pub prefix_size: usize,
-}
-
-pub enum StandbyRecord {
-    RunningXacts(StandbyRunningXacts),
-}
-
-pub struct StandbyRunningXacts {
-    pub oldest_running_xid: TransactionId,
-}
-
-pub enum ReploriginRecord {
-    Set(XlReploriginSet),
-    Drop(XlReploriginDrop),
-}
--- a/pageserver/Cargo.toml
+++ b/pageserver/Cargo.toml
@@ -30,7 +30,7 @@ futures.workspace = true
 hex.workspace = true
 humantime.workspace = true
 humantime-serde.workspace = true
-hyper0.workspace = true
+hyper.workspace = true
 itertools.workspace = true
 md5.workspace = true
 nix.workspace = true
@@ -83,7 +83,6 @@ enum-map.workspace = true
 enumset = { workspace = true, features = ["serde"]}
 strum.workspace = true
 strum_macros.workspace = true
-wal_decoder.workspace = true

 [target.'cfg(target_os = "linux")'.dependencies]
 procfs.workspace = true
@@ -93,7 +92,6 @@ criterion.workspace = true
 hex-literal.workspace = true
 tokio = { workspace = true, features = ["process", "sync", "fs", "rt", "io-util", "time", "test-util"] }
 indoc.workspace = true
-# pageserver_api = { workspace = true, features = ["testing"] }

 [[bench]]
 name = "bench_layer_map"
--- a/pageserver/benches/bench_ingest.rs
+++ b/pageserver/benches/bench_ingest.rs
@@ -6,15 +6,15 @@ use criterion::{criterion_group, criterion_main, Criterion};
 use pageserver::{
    config::PageServerConf,
    context::{DownloadBehavior, RequestContext},
-    gc_result::Value,
    l0_flush::{L0FlushConfig, L0FlushGlobalState},
    page_cache,
+    repository::Value,
    task_mgr::TaskKind,
    tenant::storage_layer::inmemory_layer::SerializedBatch,
    tenant::storage_layer::InMemoryLayer,
    virtual_file,
 };
-use pageserver_api::{key::Key, shard::TenantShardId, value::Value};
+use pageserver_api::{key::Key, shard::TenantShardId};
 use utils::{
    bin_ser::BeSer,
    id::{TenantId, TimelineId},
@@ -164,7 +164,11 @@ fn criterion_benchmark(c: &mut Criterion) {
    let conf: &'static PageServerConf = Box::leak(Box::new(
        pageserver::config::PageServerConf::dummy_conf(temp_dir.path().to_path_buf()),
    ));
-    virtual_file::init(16384, virtual_file::io_engine_for_bench());
+    virtual_file::init(
+        16384,
+        virtual_file::io_engine_for_bench(),
+        pageserver_api::config::defaults::DEFAULT_IO_BUFFER_ALIGNMENT,
+    );
    page_cache::init(conf.page_cache_size);

    {
--- a/pageserver/benches/bench_layer_map.rs
+++ b/pageserver/benches/bench_layer_map.rs
@@ -1,9 +1,9 @@
 use criterion::measurement::WallTime;
 use pageserver::keyspace::{KeyPartitioning, KeySpace};
+use pageserver::repository::Key;
 use pageserver::tenant::layer_map::LayerMap;
 use pageserver::tenant::storage_layer::LayerName;
 use pageserver::tenant::storage_layer::PersistentLayerDesc;
-use pageserver_api::key::Key;
 use pageserver_api::shard::TenantShardId;
 use rand::prelude::{SeedableRng, SliceRandom, StdRng};
 use std::cmp::{max, min};
--- a/pageserver/benches/bench_walredo.rs
+++ b/pageserver/benches/bench_walredo.rs
@@ -60,8 +60,7 @@ use anyhow::Context;
 use bytes::{Buf, Bytes};
 use criterion::{BenchmarkId, Criterion};
 use once_cell::sync::Lazy;
-use pageserver::{config::PageServerConf, walredo::PostgresRedoManager};
-use pageserver_api::record::NeonWalRecord;
+use pageserver::{config::PageServerConf, walrecord::NeonWalRecord, walredo::PostgresRedoManager};
 use pageserver_api::{key::Key, shard::TenantShardId};
 use std::{
    future::Future,
--- a/pageserver/client/src/mgmt_api.rs
+++ b/pageserver/client/src/mgmt_api.rs
@@ -540,13 +540,10 @@ impl Client {
            .map_err(Error::ReceiveBody)
    }

-    /// Configs io mode at runtime.
-    pub async fn put_io_mode(
-        &self,
-        mode: &pageserver_api::models::virtual_file::IoMode,
-    ) -> Result<()> {
-        let uri = format!("{}/v1/io_mode", self.mgmt_api_endpoint);
-        self.request(Method::PUT, uri, mode)
+    /// Configs io buffer alignment at runtime.
+    pub async fn put_io_alignment(&self, align: usize) -> Result<()> {
+        let uri = format!("{}/v1/io_alignment", self.mgmt_api_endpoint);
+        self.request(Method::PUT, uri, align)
            .await?
            .json()
            .await
--- a/pageserver/ctl/src/draw_timeline_dir.rs
+++ b/pageserver/ctl/src/draw_timeline_dir.rs
@@ -51,7 +51,7 @@
 //!

 use anyhow::{Context, Result};
-use pageserver_api::key::Key;
+use pageserver::repository::Key;
 use std::cmp::Ordering;
 use std::io::{self, BufRead};
 use std::path::PathBuf;
--- a/pageserver/ctl/src/layer_map_analyzer.rs
+++ b/pageserver/ctl/src/layer_map_analyzer.rs
@@ -13,12 +13,12 @@ use std::ops::Range;
 use std::{fs, str};

 use pageserver::page_cache::{self, PAGE_SZ};
+use pageserver::repository::{Key, KEY_SIZE};
 use pageserver::tenant::block_io::FileBlockReader;
 use pageserver::tenant::disk_btree::{DiskBtreeReader, VisitDirection};
 use pageserver::tenant::storage_layer::delta_layer::{Summary, DELTA_KEY_SIZE};
 use pageserver::tenant::storage_layer::range_overlaps;
 use pageserver::virtual_file::{self, VirtualFile};
-use pageserver_api::key::{Key, KEY_SIZE};

 use utils::{bin_ser::BeSer, lsn::Lsn};

@@ -152,7 +152,11 @@ pub(crate) async fn main(cmd: &AnalyzeLayerMapCmd) -> Result<()> {
    let ctx = RequestContext::new(TaskKind::DebugTool, DownloadBehavior::Error);

    // Initialize virtual_file (file desriptor cache) and page cache which are needed to access layer persistent B-Tree.
-    pageserver::virtual_file::init(10, virtual_file::api::IoEngineKind::StdFs);
+    pageserver::virtual_file::init(
+        10,
+        virtual_file::api::IoEngineKind::StdFs,
+        pageserver_api::config::defaults::DEFAULT_IO_BUFFER_ALIGNMENT,
+    );
    pageserver::page_cache::init(100);

    let mut total_delta_layers = 0usize;
--- a/pageserver/ctl/src/layers.rs
+++ b/pageserver/ctl/src/layers.rs
@@ -13,13 +13,13 @@ use pageserver::tenant::storage_layer::{DeltaLayer, ImageLayer};
 use pageserver::tenant::{TENANTS_SEGMENT_NAME, TIMELINES_SEGMENT_NAME};
 use pageserver::{page_cache, virtual_file};
 use pageserver::{
+    repository::{Key, KEY_SIZE},
    tenant::{
        block_io::FileBlockReader, disk_btree::VisitDirection,
        storage_layer::delta_layer::DELTA_KEY_SIZE,
    },
    virtual_file::VirtualFile,
 };
-use pageserver_api::key::{Key, KEY_SIZE};
 use std::fs;
 use utils::bin_ser::BeSer;
 use utils::id::{TenantId, TimelineId};
@@ -59,7 +59,7 @@ pub(crate) enum LayerCmd {

 async fn read_delta_file(path: impl AsRef<Path>, ctx: &RequestContext) -> Result<()> {
    let path = Utf8Path::from_path(path.as_ref()).expect("non-Unicode path");
-    virtual_file::init(10, virtual_file::api::IoEngineKind::StdFs);
+    virtual_file::init(10, virtual_file::api::IoEngineKind::StdFs, 1);
    page_cache::init(100);
    let file = VirtualFile::open(path, ctx).await?;
    let file_id = page_cache::next_file_id();
@@ -190,7 +190,11 @@ pub(crate) async fn main(cmd: &LayerCmd) -> Result<()> {
            new_tenant_id,
            new_timeline_id,
        } => {
-            pageserver::virtual_file::init(10, virtual_file::api::IoEngineKind::StdFs);
+            pageserver::virtual_file::init(
+                10,
+                virtual_file::api::IoEngineKind::StdFs,
+                pageserver_api::config::defaults::DEFAULT_IO_BUFFER_ALIGNMENT,
+            );
            pageserver::page_cache::init(100);

            let ctx = RequestContext::new(TaskKind::DebugTool, DownloadBehavior::Error);
--- a/pageserver/ctl/src/main.rs
+++ b/pageserver/ctl/src/main.rs
@@ -26,7 +26,7 @@ use pageserver::{
    tenant::{dump_layerfile_from_path, metadata::TimelineMetadata},
    virtual_file,
 };
-use pageserver_api::shard::TenantShardId;
+use pageserver_api::{config::defaults::DEFAULT_IO_BUFFER_ALIGNMENT, shard::TenantShardId};
 use postgres_ffi::ControlFileData;
 use remote_storage::{RemotePath, RemoteStorageConfig};
 use tokio_util::sync::CancellationToken;
@@ -205,7 +205,11 @@ fn read_pg_control_file(control_file_path: &Utf8Path) -> anyhow::Result<()> {

 async fn print_layerfile(path: &Utf8Path) -> anyhow::Result<()> {
    // Basic initialization of things that don't change after startup
-    virtual_file::init(10, virtual_file::api::IoEngineKind::StdFs);
+    virtual_file::init(
+        10,
+        virtual_file::api::IoEngineKind::StdFs,
+        DEFAULT_IO_BUFFER_ALIGNMENT,
+    );
    page_cache::init(100);
    let ctx = RequestContext::new(TaskKind::DebugTool, DownloadBehavior::Error);
    dump_layerfile_from_path(path, true, &ctx).await
--- a/pageserver/pagebench/src/cmd/getpage_latest_lsn.rs
+++ b/pageserver/pagebench/src/cmd/getpage_latest_lsn.rs
@@ -59,9 +59,9 @@ pub(crate) struct Args {
    #[clap(long)]
    set_io_engine: Option<pageserver_api::models::virtual_file::IoEngineKind>,

-    /// Before starting the benchmark, live-reconfigure the pageserver to use specified io mode (buffered vs. direct).
+    /// Before starting the benchmark, live-reconfigure the pageserver to use specified alignment for io buffers.
    #[clap(long)]
-    set_io_mode: Option<pageserver_api::models::virtual_file::IoMode>,
+    set_io_alignment: Option<usize>,

    targets: Option<Vec<TenantTimelineId>>,
 }
@@ -129,8 +129,8 @@ async fn main_impl(
        mgmt_api_client.put_io_engine(engine_str).await?;
    }

-    if let Some(mode) = &args.set_io_mode {
-        mgmt_api_client.put_io_mode(mode).await?;
+    if let Some(align) = args.set_io_alignment {
+        mgmt_api_client.put_io_alignment(align).await?;
    }

    // discover targets
--- a/pageserver/src/auth.rs
+++ b/pageserver/src/auth.rs
@@ -14,19 +14,14 @@ pub fn check_permission(claims: &Claims, tenant_id: Option<TenantId>) -> Result<
        }
        (Scope::PageServerApi, None) => Ok(()), // access to management api for PageServerApi scope
        (Scope::PageServerApi, Some(_)) => Ok(()), // access to tenant api using PageServerApi scope
-        (
-            Scope::Admin
-            | Scope::SafekeeperData
-            | Scope::GenerationsApi
-            | Scope::Infra
-            | Scope::Scrubber,
-            _,
-        ) => Err(AuthError(
-            format!(
-                "JWT scope '{:?}' is ineligible for Pageserver auth",
-                claims.scope
-            )
-            .into(),
-        )),
+        (Scope::Admin | Scope::SafekeeperData | Scope::GenerationsApi | Scope::Scrubber, _) => {
+            Err(AuthError(
+                format!(
+                    "JWT scope '{:?}' is ineligible for Pageserver auth",
+                    claims.scope
+                )
+                .into(),
+            ))
+        }
    }
 }
--- a/pageserver/src/bin/pageserver.rs
+++ b/pageserver/src/bin/pageserver.rs
@@ -125,7 +125,8 @@ fn main() -> anyhow::Result<()> {

    // after setting up logging, log the effective IO engine choice and read path implementations
    info!(?conf.virtual_file_io_engine, "starting with virtual_file IO engine");
-    info!(?conf.virtual_file_io_mode, "starting with virtual_file IO mode");
+    info!(?conf.virtual_file_direct_io, "starting with virtual_file Direct IO settings");
+    info!(?conf.io_buffer_alignment, "starting with setting for IO buffer alignment");

    // The tenants directory contains all the pageserver local disk state.
    // Create if not exists and make sure all the contents are durable before proceeding.
@@ -167,7 +168,11 @@ fn main() -> anyhow::Result<()> {
    let scenario = failpoint_support::init();

    // Basic initialization of things that don't change after startup
-    virtual_file::init(conf.max_file_descriptors, conf.virtual_file_io_engine);
+    virtual_file::init(
+        conf.max_file_descriptors,
+        conf.virtual_file_io_engine,
+        conf.io_buffer_alignment,
+    );
    page_cache::init(conf.page_cache_size);

    start_pageserver(launch_ts, conf).context("Failed to start pageserver")?;
@@ -570,7 +575,7 @@ fn start_pageserver(
            .build()
            .map_err(|err| anyhow!(err))?;
        let service = utils::http::RouterService::new(router).unwrap();
-        let server = hyper0::Server::from_tcp(http_listener)?
+        let server = hyper::Server::from_tcp(http_listener)?
            .serve(service)
            .with_graceful_shutdown({
                let cancel = cancel.clone();
--- a/pageserver/src/config.rs
+++ b/pageserver/src/config.rs
@@ -174,7 +174,9 @@ pub struct PageServerConf {
    pub l0_flush: crate::l0_flush::L0FlushConfig,

    /// Direct IO settings
-    pub virtual_file_io_mode: virtual_file::IoMode,
+    pub virtual_file_direct_io: virtual_file::DirectIoMode,
+
+    pub io_buffer_alignment: usize,
 }

 /// Token for authentication to safekeepers
@@ -323,10 +325,11 @@ impl PageServerConf {
            image_compression,
            ephemeral_bytes_per_memory_kb,
            l0_flush,
-            virtual_file_io_mode,
+            virtual_file_direct_io,
            concurrent_tenant_warmup,
            concurrent_tenant_size_logical_size_queries,
            virtual_file_io_engine,
+            io_buffer_alignment,
            tenant_config,
        } = config_toml;

@@ -365,6 +368,8 @@ impl PageServerConf {
            max_vectored_read_bytes,
            image_compression,
            ephemeral_bytes_per_memory_kb,
+            virtual_file_direct_io,
+            io_buffer_alignment,

            // ------------------------------------------------------------
            // fields that require additional validation or custom handling
@@ -403,7 +408,6 @@ impl PageServerConf {
            l0_flush: l0_flush
                .map(crate::l0_flush::L0FlushConfig::from)
                .unwrap_or_default(),
-            virtual_file_io_mode: virtual_file_io_mode.unwrap_or(virtual_file::IoMode::preferred()),
        };

        // ------------------------------------------------------------
--- a/pageserver/src/deletion_queue.rs
+++ b/pageserver/src/deletion_queue.rs
@@ -696,7 +696,7 @@ impl DeletionQueue {
 mod test {
    use camino::Utf8Path;
    use hex_literal::hex;
-    use pageserver_api::{key::Key, shard::ShardIndex, upcall_api::ReAttachResponseTenant};
+    use pageserver_api::{shard::ShardIndex, upcall_api::ReAttachResponseTenant};
    use std::{io::ErrorKind, time::Duration};
    use tracing::info;

@@ -705,6 +705,7 @@ mod test {

    use crate::{
        controller_upcall_client::RetryForeverError,
+        repository::Key,
        tenant::{harness::TenantHarness, storage_layer::DeltaLayerName},
    };

--- a/pageserver/src/gc_result.rs
+++ b/pageserver/src/gc_result.rs
@@ -1,57 +0,0 @@
-use anyhow::Result;
-use serde::Serialize;
-use std::ops::AddAssign;
-use std::time::Duration;
-
-///
-/// Result of performing GC
-///
-#[derive(Default, Serialize, Debug)]
-pub struct GcResult {
-    pub layers_total: u64,
-    pub layers_needed_by_cutoff: u64,
-    pub layers_needed_by_pitr: u64,
-    pub layers_needed_by_branches: u64,
-    pub layers_needed_by_leases: u64,
-    pub layers_not_updated: u64,
-    pub layers_removed: u64, // # of layer files removed because they have been made obsolete by newer ondisk files.
-
-    #[serde(serialize_with = "serialize_duration_as_millis")]
-    pub elapsed: Duration,
-
-    /// The layers which were garbage collected.
-    ///
-    /// Used in `/v1/tenant/:tenant_id/timeline/:timeline_id/do_gc` to wait for the layers to be
-    /// dropped in tests.
-    #[cfg(feature = "testing")]
-    #[serde(skip)]
-    pub(crate) doomed_layers: Vec<crate::tenant::storage_layer::Layer>,
-}
-
-// helper function for `GcResult`, serializing a `Duration` as an integer number of milliseconds
-fn serialize_duration_as_millis<S>(d: &Duration, serializer: S) -> Result<S::Ok, S::Error>
-where
-    S: serde::Serializer,
-{
-    d.as_millis().serialize(serializer)
-}
-
-impl AddAssign for GcResult {
-    fn add_assign(&mut self, other: Self) {
-        self.layers_total += other.layers_total;
-        self.layers_needed_by_pitr += other.layers_needed_by_pitr;
-        self.layers_needed_by_cutoff += other.layers_needed_by_cutoff;
-        self.layers_needed_by_branches += other.layers_needed_by_branches;
-        self.layers_needed_by_leases += other.layers_needed_by_leases;
-        self.layers_not_updated += other.layers_not_updated;
-        self.layers_removed += other.layers_removed;
-
-        self.elapsed += other.elapsed;
-
-        #[cfg(feature = "testing")]
-        {
-            let mut other = other;
-            self.doomed_layers.append(&mut other.doomed_layers);
-        }
-    }
-}
--- a/pageserver/src/http/routes.rs
+++ b/pageserver/src/http/routes.rs
@@ -17,7 +17,6 @@ use hyper::header;
 use hyper::StatusCode;
 use hyper::{Body, Request, Response, Uri};
 use metrics::launch_timestamp::LaunchTimestamp;
-use pageserver_api::models::virtual_file::IoMode;
 use pageserver_api::models::AuxFilePolicy;
 use pageserver_api::models::DownloadRemoteLayersTaskSpawnRequest;
 use pageserver_api::models::IngestAuxFilesRequest;
@@ -77,7 +76,6 @@ use crate::tenant::secondary::SecondaryController;
 use crate::tenant::size::ModelInputs;
 use crate::tenant::storage_layer::LayerAccessStatsReset;
 use crate::tenant::storage_layer::LayerName;
-use crate::tenant::timeline::offload::offload_timeline;
 use crate::tenant::timeline::CompactFlags;
 use crate::tenant::timeline::CompactionError;
 use crate::tenant::timeline::Timeline;
@@ -326,7 +324,6 @@ impl From<crate::tenant::TimelineArchivalError> for ApiError {
        match value {
            NotFound => ApiError::NotFound(anyhow::anyhow!("timeline not found").into()),
            Timeout => ApiError::Timeout("hit pageserver internal timeout".into()),
-            Cancelled => ApiError::ShuttingDown,
            e @ HasArchivedParent(_) => {
                ApiError::PreconditionFailed(e.to_string().into_boxed_str())
            }
@@ -706,8 +703,6 @@ async fn timeline_archival_config_handler(
    let tenant_shard_id: TenantShardId = parse_request_param(&request, "tenant_shard_id")?;
    let timeline_id: TimelineId = parse_request_param(&request, "timeline_id")?;

-    let ctx = RequestContext::new(TaskKind::MgmtRequest, DownloadBehavior::Warn);
-
    let request_data: TimelineArchivalConfigRequest = json_request(&mut request).await?;
    check_permission(&request, Some(tenant_shard_id.tenant_id))?;
    let state = get_state(&request);
@@ -717,10 +712,8 @@ async fn timeline_archival_config_handler(
            .tenant_manager
            .get_attached_tenant_shard(tenant_shard_id)?;

-        tenant.wait_to_become_active(ACTIVE_TENANT_TIMEOUT).await?;
-
        tenant
-            .apply_timeline_archival_config(timeline_id, request_data.state, ctx)
+            .apply_timeline_archival_config(timeline_id, request_data.state)
            .await?;
        Ok::<_, ApiError>(())
    }
@@ -1787,49 +1780,6 @@ async fn timeline_compact_handler(
    .await
 }

-// Run offload immediately on given timeline.
-async fn timeline_offload_handler(
-    request: Request<Body>,
-    _cancel: CancellationToken,
-) -> Result<Response<Body>, ApiError> {
-    let tenant_shard_id: TenantShardId = parse_request_param(&request, "tenant_shard_id")?;
-    let timeline_id: TimelineId = parse_request_param(&request, "timeline_id")?;
-    check_permission(&request, Some(tenant_shard_id.tenant_id))?;
-
-    let state = get_state(&request);
-
-    async {
-        let tenant = state
-            .tenant_manager
-            .get_attached_tenant_shard(tenant_shard_id)?;
-
-        if tenant.get_offloaded_timeline(timeline_id).is_ok() {
-            return json_response(StatusCode::OK, ());
-        }
-        let timeline =
-            active_timeline_of_active_tenant(&state.tenant_manager, tenant_shard_id, timeline_id)
-                .await?;
-
-        if !tenant.timeline_has_no_attached_children(timeline_id) {
-            return Err(ApiError::PreconditionFailed(
-                "timeline has attached children".into(),
-            ));
-        }
-        if !timeline.can_offload() {
-            return Err(ApiError::PreconditionFailed(
-                "Timeline::can_offload() returned false".into(),
-            ));
-        }
-        offload_timeline(&tenant, &timeline)
-            .await
-            .map_err(ApiError::InternalServerError)?;
-
-        json_response(StatusCode::OK, ())
-    }
-    .instrument(info_span!("manual_timeline_offload", tenant_id = %tenant_shard_id.tenant_id, shard_id = %tenant_shard_id.shard_slug(), %timeline_id))
-    .await
-}
-
 // Run checkpoint immediately on given timeline.
 async fn timeline_checkpoint_handler(
    request: Request<Body>,
@@ -2050,13 +2000,13 @@ async fn getpage_at_lsn_handler(
    check_permission(&request, Some(tenant_shard_id.tenant_id))?;
    let state = get_state(&request);

-    struct Key(pageserver_api::key::Key);
+    struct Key(crate::repository::Key);

    impl std::str::FromStr for Key {
        type Err = anyhow::Error;

        fn from_str(s: &str) -> std::result::Result<Self, Self::Err> {
-            pageserver_api::key::Key::from_hex(s).map(Key)
+            crate::repository::Key::from_hex(s).map(Key)
        }
    }

@@ -2429,13 +2379,17 @@ async fn put_io_engine_handler(
    json_response(StatusCode::OK, ())
 }

-async fn put_io_mode_handler(
+async fn put_io_alignment_handler(
    mut r: Request<Body>,
    _cancel: CancellationToken,
 ) -> Result<Response<Body>, ApiError> {
    check_permission(&r, None)?;
-    let mode: IoMode = json_request(&mut r).await?;
-    crate::virtual_file::set_io_mode(mode);
+    let align: usize = json_request(&mut r).await?;
+    crate::virtual_file::set_io_buffer_alignment(align).map_err(|align| {
+        ApiError::PreconditionFailed(
+            format!("Requested io alignment ({align}) is not a power of two").into(),
+        )
+    })?;
    json_response(StatusCode::OK, ())
 }

@@ -3053,10 +3007,6 @@ pub fn make_router(
            "/v1/tenant/:tenant_shard_id/timeline/:timeline_id/compact",
            |r| api_handler(r, timeline_compact_handler),
        )
-        .put(
-            "/v1/tenant/:tenant_shard_id/timeline/:timeline_id/offload",
-            |r| testing_api_handler("attempt timeline offload", r, timeline_offload_handler),
-        )
        .put(
            "/v1/tenant/:tenant_shard_id/timeline/:timeline_id/checkpoint",
            |r| testing_api_handler("run timeline checkpoint", r, timeline_checkpoint_handler),
@@ -3130,7 +3080,9 @@ pub fn make_router(
            |r| api_handler(r, timeline_collect_keyspace),
        )
        .put("/v1/io_engine", |r| api_handler(r, put_io_engine_handler))
-        .put("/v1/io_mode", |r| api_handler(r, put_io_mode_handler))
+        .put("/v1/io_alignment", |r| {
+            api_handler(r, put_io_alignment_handler)
+        })
        .put(
            "/v1/tenant/:tenant_shard_id/timeline/:timeline_id/force_aux_policy_switch",
            |r| api_handler(r, force_aux_policy_switch_handler),
--- a/pageserver/src/import_datadir.rs
+++ b/pageserver/src/import_datadir.rs
@@ -12,7 +12,6 @@ use pageserver_api::key::rel_block_to_key;
 use tokio::io::{AsyncRead, AsyncReadExt};
 use tokio_tar::Archive;
 use tracing::*;
-use wal_decoder::models::InterpretedWalRecord;
 use walkdir::WalkDir;

 use crate::context::RequestContext;
@@ -20,6 +19,8 @@ use crate::metrics::WAL_INGEST;
 use crate::pgdatadir_mapping::*;
 use crate::tenant::Timeline;
 use crate::walingest::WalIngest;
+use crate::walrecord::decode_wal_record;
+use crate::walrecord::DecodedWALRecord;
 use pageserver_api::reltag::{RelTag, SlruKind};
 use postgres_ffi::pg_constants;
 use postgres_ffi::relfile_utils::*;
@@ -312,15 +313,11 @@ async fn import_wal(
        let mut modification = tline.begin_modification(last_lsn);
        while last_lsn <= endpoint {
            if let Some((lsn, recdata)) = waldecoder.poll_decode()? {
-                let interpreted = InterpretedWalRecord::from_bytes(
-                    recdata,
-                    tline.get_shard_identity(),
-                    lsn,
-                    tline.pg_version,
-                )?;
+                let mut decoded = DecodedWALRecord::default();
+                decode_wal_record(recdata, &mut decoded, tline.pg_version)?;

                walingest
-                    .ingest_record(interpreted, &mut modification, ctx)
+                    .ingest_record(decoded, lsn, &mut modification, ctx)
                    .await?;
                WAL_INGEST.records_committed.inc();

@@ -457,15 +454,10 @@ pub async fn import_wal_from_tar(
        let mut modification = tline.begin_modification(last_lsn);
        while last_lsn <= end_lsn {
            if let Some((lsn, recdata)) = waldecoder.poll_decode()? {
-                let interpreted = InterpretedWalRecord::from_bytes(
-                    recdata,
-                    tline.get_shard_identity(),
-                    lsn,
-                    tline.pg_version,
-                )?;
-
+                let mut decoded = DecodedWALRecord::default();
+                decode_wal_record(recdata, &mut decoded, tline.pg_version)?;
                walingest
-                    .ingest_record(interpreted, &mut modification, ctx)
+                    .ingest_record(decoded, lsn, &mut modification, ctx)
                    .await?;
                modification.commit(ctx).await?;
                last_lsn = lsn;
--- a/pageserver/src/lib.rs
+++ b/pageserver/src/lib.rs
@@ -13,18 +13,16 @@ pub mod http;
 pub mod import_datadir;
 pub mod l0_flush;

-extern crate hyper0 as hyper;
-
 use futures::{stream::FuturesUnordered, StreamExt};
 pub use pageserver_api::keyspace;
 use tokio_util::sync::CancellationToken;
 mod assert_u64_eq_usize;
 pub mod aux_file;
-pub mod gc_result;
 pub mod metrics;
 pub mod page_cache;
 pub mod page_service;
 pub mod pgdatadir_mapping;
+pub mod repository;
 pub mod span;
 pub(crate) mod statvfs;
 pub mod task_mgr;
@@ -32,6 +30,7 @@ pub mod tenant;
 pub mod utilization;
 pub mod virtual_file;
 pub mod walingest;
+pub mod walrecord;
 pub mod walredo;

 use camino::Utf8Path;
--- a/pageserver/src/pgdatadir_mapping.rs
+++ b/pageserver/src/pgdatadir_mapping.rs
@@ -7,14 +7,14 @@
 //! Clarify that)
 //!
 use super::tenant::{PageReconstructError, Timeline};
-use crate::aux_file;
 use crate::context::RequestContext;
 use crate::keyspace::{KeySpace, KeySpaceAccum};
 use crate::span::debug_assert_current_span_has_tenant_and_timeline_id_no_shard_id;
+use crate::walrecord::NeonWalRecord;
+use crate::{aux_file, repository::*};
 use anyhow::{ensure, Context};
 use bytes::{Buf, Bytes, BytesMut};
 use enum_map::Enum;
-use pageserver_api::key::Key;
 use pageserver_api::key::{
    dbdir_key_range, rel_block_to_key, rel_dir_to_key, rel_key_range, rel_size_to_key,
    relmap_file_key, repl_origin_key, repl_origin_key_range, slru_block_to_key, slru_dir_to_key,
@@ -23,9 +23,7 @@ use pageserver_api::key::{
 };
 use pageserver_api::keyspace::SparseKeySpace;
 use pageserver_api::models::AuxFilePolicy;
-use pageserver_api::record::NeonWalRecord;
 use pageserver_api::reltag::{BlockNumber, RelTag, SlruKind};
-use pageserver_api::value::Value;
 use postgres_ffi::relfile_utils::{FSM_FORKNUM, VISIBILITYMAP_FORKNUM};
 use postgres_ffi::BLCKSZ;
 use postgres_ffi::{Oid, RepOriginId, TimestampTz, TransactionId};
--- a/libs/pageserver_api/src/value.rs
+++ b/libs/pageserver_api/src/value.rs
@@ -1,9 +1,13 @@
-//! This module defines the value type used by the storage engine.
-
-use crate::record::NeonWalRecord;
+use crate::walrecord::NeonWalRecord;
+use anyhow::Result;
 use bytes::Bytes;
 use serde::{Deserialize, Serialize};
+use std::ops::AddAssign;
+use std::time::Duration;

+pub use pageserver_api::key::{Key, KEY_SIZE};
+
+/// A 'value' stored for a one Key.
 #[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)]
 pub enum Value {
    /// An Image value contains a full copy of the value
@@ -29,17 +33,17 @@ impl Value {
 }

 #[derive(Debug, PartialEq)]
-pub enum InvalidInput {
+pub(crate) enum InvalidInput {
    TooShortValue,
    TooShortPostgresRecord,
 }

 /// We could have a ValueRef where everything is `serde(borrow)`. Before implementing that, lets
 /// use this type for querying if a slice looks some particular way.
-pub struct ValueBytes;
+pub(crate) struct ValueBytes;

 impl ValueBytes {
-    pub fn will_init(raw: &[u8]) -> Result<bool, InvalidInput> {
+    pub(crate) fn will_init(raw: &[u8]) -> Result<bool, InvalidInput> {
        if raw.len() < 12 {
            return Err(InvalidInput::TooShortValue);
        }
@@ -75,7 +79,6 @@ impl ValueBytes {
 mod test {
    use super::*;

-    use bytes::Bytes;
    use utils::bin_ser::BeSer;

    macro_rules! roundtrip {
@@ -226,3 +229,56 @@ mod test {
        assert!(!ValueBytes::will_init(&expected).unwrap());
    }
 }
+
+///
+/// Result of performing GC
+///
+#[derive(Default, Serialize, Debug)]
+pub struct GcResult {
+    pub layers_total: u64,
+    pub layers_needed_by_cutoff: u64,
+    pub layers_needed_by_pitr: u64,
+    pub layers_needed_by_branches: u64,
+    pub layers_needed_by_leases: u64,
+    pub layers_not_updated: u64,
+    pub layers_removed: u64, // # of layer files removed because they have been made obsolete by newer ondisk files.
+
+    #[serde(serialize_with = "serialize_duration_as_millis")]
+    pub elapsed: Duration,
+
+    /// The layers which were garbage collected.
+    ///
+    /// Used in `/v1/tenant/:tenant_id/timeline/:timeline_id/do_gc` to wait for the layers to be
+    /// dropped in tests.
+    #[cfg(feature = "testing")]
+    #[serde(skip)]
+    pub(crate) doomed_layers: Vec<crate::tenant::storage_layer::Layer>,
+}
+
+// helper function for `GcResult`, serializing a `Duration` as an integer number of milliseconds
+fn serialize_duration_as_millis<S>(d: &Duration, serializer: S) -> Result<S::Ok, S::Error>
+where
+    S: serde::Serializer,
+{
+    d.as_millis().serialize(serializer)
+}
+
+impl AddAssign for GcResult {
+    fn add_assign(&mut self, other: Self) {
+        self.layers_total += other.layers_total;
+        self.layers_needed_by_pitr += other.layers_needed_by_pitr;
+        self.layers_needed_by_cutoff += other.layers_needed_by_cutoff;
+        self.layers_needed_by_branches += other.layers_needed_by_branches;
+        self.layers_needed_by_leases += other.layers_needed_by_leases;
+        self.layers_not_updated += other.layers_not_updated;
+        self.layers_removed += other.layers_removed;
+
+        self.elapsed += other.elapsed;
+
+        #[cfg(feature = "testing")]
+        {
+            let mut other = other;
+            self.doomed_layers.append(&mut other.doomed_layers);
+        }
+    }
+}
--- a/pageserver/src/tenant.rs
+++ b/pageserver/src/tenant.rs
@@ -38,7 +38,6 @@ use std::future::Future;
 use std::sync::Weak;
 use std::time::SystemTime;
 use storage_broker::BrokerClientChannel;
-use timeline::offload::offload_timeline;
 use tokio::io::BufReader;
 use tokio::sync::watch;
 use tokio::task::JoinSet;
@@ -79,7 +78,6 @@ use crate::config::PageServerConf;
 use crate::context::{DownloadBehavior, RequestContext};
 use crate::deletion_queue::DeletionQueueClient;
 use crate::deletion_queue::DeletionQueueError;
-use crate::gc_result::GcResult;
 use crate::import_datadir;
 use crate::is_uninit_mark;
 use crate::l0_flush::L0FlushGlobalState;
@@ -88,6 +86,7 @@ use crate::metrics::{
    remove_tenant_metrics, BROKEN_TENANTS_SET, CIRCUIT_BREAKERS_BROKEN, CIRCUIT_BREAKERS_UNBROKEN,
    TENANT_STATE_METRIC, TENANT_SYNTHETIC_SIZE_METRIC,
 };
+use crate::repository::GcResult;
 use crate::task_mgr;
 use crate::task_mgr::TaskKind;
 use crate::tenant::config::LocationMode;
@@ -288,13 +287,9 @@ pub struct Tenant {

    /// During timeline creation, we first insert the TimelineId to the
    /// creating map, then `timelines`, then remove it from the creating map.
-    /// **Lock order**: if acquiring both, acquire`timelines` before `timelines_creating`
+    /// **Lock order**: if acquring both, acquire`timelines` before `timelines_creating`
    timelines_creating: std::sync::Mutex<HashSet<TimelineId>>,

-    /// Possibly offloaded and archived timelines
-    /// **Lock order**: if acquiring both, acquire`timelines` before `timelines_offloaded`
-    timelines_offloaded: Mutex<HashMap<TimelineId, Arc<OffloadedTimeline>>>,
-
    // This mutex prevents creation of new timelines during GC.
    // Adding yet another mutex (in addition to `timelines`) is needed because holding
    // `timelines` mutex during all GC iteration
@@ -461,10 +456,10 @@ impl WalRedoManager {
    /// This method is cancellation-safe.
    pub async fn request_redo(
        &self,
-        key: pageserver_api::key::Key,
+        key: crate::repository::Key,
        lsn: Lsn,
        base_img: Option<(Lsn, bytes::Bytes)>,
-        records: Vec<(Lsn, pageserver_api::record::NeonWalRecord)>,
+        records: Vec<(Lsn, crate::walrecord::NeonWalRecord)>,
        pg_version: u32,
    ) -> Result<bytes::Bytes, walredo::Error> {
        match self {
@@ -489,77 +484,6 @@ impl WalRedoManager {
    }
 }

-pub struct OffloadedTimeline {
-    pub tenant_shard_id: TenantShardId,
-    pub timeline_id: TimelineId,
-    pub ancestor_timeline_id: Option<TimelineId>,
-    /// Whether to retain the branch lsn at the ancestor or not
-    pub ancestor_retain_lsn: Option<Lsn>,
-
-    // TODO: once we persist offloaded state, make this lazily constructed
-    pub remote_client: Arc<RemoteTimelineClient>,
-
-    /// Prevent two tasks from deleting the timeline at the same time. If held, the
-    /// timeline is being deleted. If 'true', the timeline has already been deleted.
-    pub delete_progress: Arc<tokio::sync::Mutex<DeleteTimelineFlow>>,
-}
-
-impl OffloadedTimeline {
-    fn from_timeline(timeline: &Timeline) -> Self {
-        let ancestor_retain_lsn = timeline
-            .get_ancestor_timeline_id()
-            .map(|_timeline_id| timeline.get_ancestor_lsn());
-        Self {
-            tenant_shard_id: timeline.tenant_shard_id,
-            timeline_id: timeline.timeline_id,
-            ancestor_timeline_id: timeline.get_ancestor_timeline_id(),
-            ancestor_retain_lsn,
-
-            remote_client: timeline.remote_client.clone(),
-            delete_progress: timeline.delete_progress.clone(),
-        }
-    }
-}
-
-#[derive(Copy, Clone, PartialEq, Eq, Hash, Debug)]
-pub enum MaybeOffloaded {
-    Yes,
-    No,
-}
-
-#[derive(Clone)]
-pub enum TimelineOrOffloaded {
-    Timeline(Arc<Timeline>),
-    Offloaded(Arc<OffloadedTimeline>),
-}
-
-impl TimelineOrOffloaded {
-    pub fn tenant_shard_id(&self) -> TenantShardId {
-        match self {
-            TimelineOrOffloaded::Timeline(timeline) => timeline.tenant_shard_id,
-            TimelineOrOffloaded::Offloaded(offloaded) => offloaded.tenant_shard_id,
-        }
-    }
-    pub fn timeline_id(&self) -> TimelineId {
-        match self {
-            TimelineOrOffloaded::Timeline(timeline) => timeline.timeline_id,
-            TimelineOrOffloaded::Offloaded(offloaded) => offloaded.timeline_id,
-        }
-    }
-    pub fn delete_progress(&self) -> &Arc<tokio::sync::Mutex<DeleteTimelineFlow>> {
-        match self {
-            TimelineOrOffloaded::Timeline(timeline) => &timeline.delete_progress,
-            TimelineOrOffloaded::Offloaded(offloaded) => &offloaded.delete_progress,
-        }
-    }
-    pub fn remote_client(&self) -> &Arc<RemoteTimelineClient> {
-        match self {
-            TimelineOrOffloaded::Timeline(timeline) => &timeline.remote_client,
-            TimelineOrOffloaded::Offloaded(offloaded) => &offloaded.remote_client,
-        }
-    }
-}
-
 #[derive(Debug, thiserror::Error, PartialEq, Eq)]
 pub enum GetTimelineError {
    #[error("Timeline is shutting down")]
@@ -619,9 +543,6 @@ pub enum TimelineArchivalError {
    #[error("Timeout")]
    Timeout,

-    #[error("Cancelled")]
-    Cancelled,
-
    #[error("ancestor is archived: {}", .0)]
    HasArchivedParent(TimelineId),

@@ -640,7 +561,6 @@ impl Debug for TimelineArchivalError {
        match self {
            Self::NotFound => write!(f, "NotFound"),
            Self::Timeout => write!(f, "Timeout"),
-            Self::Cancelled => write!(f, "Cancelled"),
            Self::HasArchivedParent(p) => f.debug_tuple("HasArchivedParent").field(p).finish(),
            Self::HasUnarchivedChildren(c) => {
                f.debug_tuple("HasUnarchivedChildren").field(c).finish()
@@ -1486,195 +1406,52 @@ impl Tenant {
        }
    }

-    fn check_to_be_archived_has_no_unarchived_children(
-        timeline_id: TimelineId,
-        timelines: &std::sync::MutexGuard<'_, HashMap<TimelineId, Arc<Timeline>>>,
-    ) -> Result<(), TimelineArchivalError> {
-        let children: Vec<TimelineId> = timelines
-            .iter()
-            .filter_map(|(id, entry)| {
-                if entry.get_ancestor_timeline_id() != Some(timeline_id) {
-                    return None;
-                }
-                if entry.is_archived() == Some(true) {
-                    return None;
-                }
-                Some(*id)
-            })
-            .collect();
-
-        if !children.is_empty() {
-            return Err(TimelineArchivalError::HasUnarchivedChildren(children));
-        }
-        Ok(())
-    }
-
-    fn check_ancestor_of_to_be_unarchived_is_not_archived(
-        ancestor_timeline_id: TimelineId,
-        timelines: &std::sync::MutexGuard<'_, HashMap<TimelineId, Arc<Timeline>>>,
-        offloaded_timelines: &std::sync::MutexGuard<
-            '_,
-            HashMap<TimelineId, Arc<OffloadedTimeline>>,
-        >,
-    ) -> Result<(), TimelineArchivalError> {
-        let has_archived_parent =
-            if let Some(ancestor_timeline) = timelines.get(&ancestor_timeline_id) {
-                ancestor_timeline.is_archived() == Some(true)
-            } else if offloaded_timelines.contains_key(&ancestor_timeline_id) {
-                true
-            } else {
-                error!("ancestor timeline {ancestor_timeline_id} not found");
-                if cfg!(debug_assertions) {
-                    panic!("ancestor timeline {ancestor_timeline_id} not found");
-                }
-                return Err(TimelineArchivalError::NotFound);
-            };
-        if has_archived_parent {
-            return Err(TimelineArchivalError::HasArchivedParent(
-                ancestor_timeline_id,
-            ));
-        }
-        Ok(())
-    }
-
-    fn check_to_be_unarchived_timeline_has_no_archived_parent(
-        timeline: &Arc<Timeline>,
-    ) -> Result<(), TimelineArchivalError> {
-        if let Some(ancestor_timeline) = timeline.ancestor_timeline() {
-            if ancestor_timeline.is_archived() == Some(true) {
-                return Err(TimelineArchivalError::HasArchivedParent(
-                    ancestor_timeline.timeline_id,
-                ));
-            }
-        }
-        Ok(())
-    }
-
-    /// Loads the specified (offloaded) timeline from S3 and attaches it as a loaded timeline
-    async fn unoffload_timeline(
-        self: &Arc<Self>,
-        timeline_id: TimelineId,
-        ctx: RequestContext,
-    ) -> Result<Arc<Timeline>, TimelineArchivalError> {
-        info!("unoffloading timeline");
-        let cancel = self.cancel.clone();
-        let timeline_preload = self
-            .load_timeline_metadata(timeline_id, self.remote_storage.clone(), cancel)
-            .await;
-
-        let index_part = match timeline_preload.index_part {
-            Ok(index_part) => {
-                debug!("remote index part exists for timeline {timeline_id}");
-                index_part
-            }
-            Err(DownloadError::NotFound) => {
-                error!(%timeline_id, "index_part not found on remote");
-                return Err(TimelineArchivalError::NotFound);
-            }
-            Err(DownloadError::Cancelled) => return Err(TimelineArchivalError::Cancelled),
-            Err(e) => {
-                // Some (possibly ephemeral) error happened during index_part download.
-                warn!(%timeline_id, "Failed to load index_part from remote storage, failed creation? ({e})");
-                return Err(TimelineArchivalError::Other(
-                    anyhow::Error::new(e).context("downloading index_part from remote storage"),
-                ));
-            }
-        };
-        let index_part = match index_part {
-            MaybeDeletedIndexPart::IndexPart(index_part) => index_part,
-            MaybeDeletedIndexPart::Deleted(_index_part) => {
-                info!("timeline is deleted according to index_part.json");
-                return Err(TimelineArchivalError::NotFound);
-            }
-        };
-        let remote_metadata = index_part.metadata.clone();
-        let timeline_resources = self.build_timeline_resources(timeline_id);
-        self.load_remote_timeline(
-            timeline_id,
-            index_part,
-            remote_metadata,
-            timeline_resources,
-            &ctx,
-        )
-        .await
-        .with_context(|| {
-            format!(
-                "failed to load remote timeline {} for tenant {}",
-                timeline_id, self.tenant_shard_id
-            )
-        })?;
-        let timelines = self.timelines.lock().unwrap();
-        if let Some(timeline) = timelines.get(&timeline_id) {
-            let mut offloaded_timelines = self.timelines_offloaded.lock().unwrap();
-            if offloaded_timelines.remove(&timeline_id).is_none() {
-                warn!("timeline already removed from offloaded timelines");
-            }
-            info!("timeline unoffloading complete");
-            Ok(Arc::clone(timeline))
-        } else {
-            warn!("timeline not available directly after attach");
-            Err(TimelineArchivalError::Other(anyhow::anyhow!(
-                "timeline not available directly after attach"
-            )))
-        }
-    }
-
    pub(crate) async fn apply_timeline_archival_config(
-        self: &Arc<Self>,
+        &self,
        timeline_id: TimelineId,
-        new_state: TimelineArchivalState,
-        ctx: RequestContext,
+        state: TimelineArchivalState,
    ) -> Result<(), TimelineArchivalError> {
        info!("setting timeline archival config");
-        // First part: figure out what is needed to do, and do validation
-        let timeline_or_unarchive_offloaded = 'outer: {
+        let timeline = {
            let timelines = self.timelines.lock().unwrap();

            let Some(timeline) = timelines.get(&timeline_id) else {
-                let offloaded_timelines = self.timelines_offloaded.lock().unwrap();
-                let Some(offloaded) = offloaded_timelines.get(&timeline_id) else {
-                    return Err(TimelineArchivalError::NotFound);
-                };
-                if new_state == TimelineArchivalState::Archived {
-                    // It's offloaded already, so nothing to do
-                    return Ok(());
-                }
-                if let Some(ancestor_timeline_id) = offloaded.ancestor_timeline_id {
-                    Self::check_ancestor_of_to_be_unarchived_is_not_archived(
-                        ancestor_timeline_id,
-                        &timelines,
-                        &offloaded_timelines,
-                    )?;
-                }
-                break 'outer None;
+                return Err(TimelineArchivalError::NotFound);
            };

-            // Do some validation. We release the timelines lock below, so there is potential
-            // for race conditions: these checks are more present to prevent misunderstandings of
-            // the API's capabilities, instead of serving as the sole way to defend their invariants.
-            match new_state {
-                TimelineArchivalState::Unarchived => {
-                    Self::check_to_be_unarchived_timeline_has_no_archived_parent(timeline)?
-                }
-                TimelineArchivalState::Archived => {
-                    Self::check_to_be_archived_has_no_unarchived_children(timeline_id, &timelines)?
+            if state == TimelineArchivalState::Unarchived {
+                if let Some(ancestor_timeline) = timeline.ancestor_timeline() {
+                    if ancestor_timeline.is_archived() == Some(true) {
+                        return Err(TimelineArchivalError::HasArchivedParent(
+                            ancestor_timeline.timeline_id,
+                        ));
+                    }
                }
            }
-            Some(Arc::clone(timeline))
+
+            // Ensure that there are no non-archived child timelines
+            let children: Vec<TimelineId> = timelines
+                .iter()
+                .filter_map(|(id, entry)| {
+                    if entry.get_ancestor_timeline_id() != Some(timeline_id) {
+                        return None;
+                    }
+                    if entry.is_archived() == Some(true) {
+                        return None;
+                    }
+                    Some(*id)
+                })
+                .collect();
+
+            if !children.is_empty() && state == TimelineArchivalState::Archived {
+                return Err(TimelineArchivalError::HasUnarchivedChildren(children));
+            }
+            Arc::clone(timeline)
        };

-        // Second part: unarchive timeline (if needed)
-        let timeline = if let Some(timeline) = timeline_or_unarchive_offloaded {
-            timeline
-        } else {
-            // Turn offloaded timeline into a non-offloaded one
-            self.unoffload_timeline(timeline_id, ctx).await?
-        };
-
-        // Third part: upload new timeline archival state and block until it is present in S3
        let upload_needed = timeline
            .remote_client
-            .schedule_index_upload_for_timeline_archival_state(new_state)?;
+            .schedule_index_upload_for_timeline_archival_state(state)?;

        if upload_needed {
            info!("Uploading new state");
@@ -1690,21 +1467,6 @@ impl Tenant {
        Ok(())
    }

-    pub fn get_offloaded_timeline(
-        &self,
-        timeline_id: TimelineId,
-    ) -> Result<Arc<OffloadedTimeline>, GetTimelineError> {
-        self.timelines_offloaded
-            .lock()
-            .unwrap()
-            .get(&timeline_id)
-            .map(Arc::clone)
-            .ok_or(GetTimelineError::NotFound {
-                tenant_id: self.tenant_shard_id,
-                timeline_id,
-            })
-    }
-
    pub(crate) fn tenant_shard_id(&self) -> TenantShardId {
        self.tenant_shard_id
    }
@@ -2122,7 +1884,7 @@ impl Tenant {
    ///
    /// Returns whether we have pending compaction task.
    async fn compaction_iteration(
-        self: &Arc<Self>,
+        &self,
        cancel: &CancellationToken,
        ctx: &RequestContext,
    ) -> Result<bool, timeline::CompactionError> {
@@ -2143,28 +1905,21 @@ impl Tenant {
        // while holding the lock. Then drop the lock and actually perform the
        // compactions.  We don't want to block everything else while the
        // compaction runs.
-        let timelines_to_compact_or_offload;
-        {
+        let timelines_to_compact = {
            let timelines = self.timelines.lock().unwrap();
-            timelines_to_compact_or_offload = timelines
+            let timelines_to_compact = timelines
                .iter()
                .filter_map(|(timeline_id, timeline)| {
-                    let (is_active, can_offload) = (timeline.is_active(), timeline.can_offload());
-                    let has_no_unoffloaded_children = {
-                        !timelines
-                            .iter()
-                            .any(|(_id, tl)| tl.get_ancestor_timeline_id() == Some(*timeline_id))
-                    };
-                    let can_offload = can_offload && has_no_unoffloaded_children;
-                    if (is_active, can_offload) == (false, false) {
-                        None
+                    if timeline.is_active() {
+                        Some((*timeline_id, timeline.clone()))
                    } else {
-                        Some((*timeline_id, timeline.clone(), (is_active, can_offload)))
+                        None
                    }
                })
                .collect::<Vec<_>>();
            drop(timelines);
-        }
+            timelines_to_compact
+        };

        // Before doing any I/O work, check our circuit breaker
        if self.compaction_circuit_breaker.lock().unwrap().is_broken() {
@@ -2174,34 +1929,20 @@ impl Tenant {

        let mut has_pending_task = false;

-        for (timeline_id, timeline, (can_compact, can_offload)) in &timelines_to_compact_or_offload
-        {
-            let pending_task_left = if *can_compact {
-                Some(
-                    timeline
-                        .compact(cancel, EnumSet::empty(), ctx)
-                        .instrument(info_span!("compact_timeline", %timeline_id))
-                        .await
-                        .inspect_err(|e| match e {
-                            timeline::CompactionError::ShuttingDown => (),
-                            timeline::CompactionError::Other(e) => {
-                                self.compaction_circuit_breaker
-                                    .lock()
-                                    .unwrap()
-                                    .fail(&CIRCUIT_BREAKERS_BROKEN, e);
-                            }
-                        })?,
-                )
-            } else {
-                None
-            };
-            has_pending_task |= pending_task_left.unwrap_or(false);
-            if pending_task_left == Some(false) && *can_offload {
-                offload_timeline(self, timeline)
-                    .instrument(info_span!("offload_timeline", %timeline_id))
-                    .await
-                    .map_err(timeline::CompactionError::Other)?;
-            }
+        for (timeline_id, timeline) in &timelines_to_compact {
+            has_pending_task |= timeline
+                .compact(cancel, EnumSet::empty(), ctx)
+                .instrument(info_span!("compact_timeline", %timeline_id))
+                .await
+                .inspect_err(|e| match e {
+                    timeline::CompactionError::ShuttingDown => (),
+                    timeline::CompactionError::Other(e) => {
+                        self.compaction_circuit_breaker
+                            .lock()
+                            .unwrap()
+                            .fail(&CIRCUIT_BREAKERS_BROKEN, e);
+                    }
+                })?;
        }

        self.compaction_circuit_breaker
@@ -2240,13 +1981,6 @@ impl Tenant {
        }
    }

-    pub fn timeline_has_no_attached_children(&self, timeline_id: TimelineId) -> bool {
-        let timelines = self.timelines.lock().unwrap();
-        !timelines
-            .iter()
-            .any(|(_id, tl)| tl.get_ancestor_timeline_id() == Some(timeline_id))
-    }
-
    pub fn current_state(&self) -> TenantState {
        self.state.borrow().clone()
    }
@@ -2294,13 +2028,12 @@ impl Tenant {

        if activating {
            let timelines_accessor = self.timelines.lock().unwrap();
-            let timelines_offloaded_accessor = self.timelines_offloaded.lock().unwrap();
            let timelines_to_activate = timelines_accessor
                .values()
                .filter(|timeline| !(timeline.is_broken() || timeline.is_stopping()));

            // Before activation, populate each Timeline's GcInfo with information about its children
-            self.initialize_gc_info(&timelines_accessor, &timelines_offloaded_accessor);
+            self.initialize_gc_info(&timelines_accessor);

            // Spawn gc and compaction loops. The loops will shut themselves
            // down when they notice that the tenant is inactive.
@@ -3119,7 +2852,6 @@ impl Tenant {
            constructed_at: Instant::now(),
            timelines: Mutex::new(HashMap::new()),
            timelines_creating: Mutex::new(HashSet::new()),
-            timelines_offloaded: Mutex::new(HashMap::new()),
            gc_cs: tokio::sync::Mutex::new(()),
            walredo_mgr,
            remote_storage,
@@ -3340,7 +3072,6 @@ impl Tenant {
    fn initialize_gc_info(
        &self,
        timelines: &std::sync::MutexGuard<HashMap<TimelineId, Arc<Timeline>>>,
-        timelines_offloaded: &std::sync::MutexGuard<HashMap<TimelineId, Arc<OffloadedTimeline>>>,
    ) {
        // This function must be called before activation: after activation timeline create/delete operations
        // might happen, and this function is not safe to run concurrently with those.
@@ -3348,37 +3079,20 @@ impl Tenant {

        // Scan all timelines. For each timeline, remember the timeline ID and
        // the branch point where it was created.
-        let mut all_branchpoints: BTreeMap<TimelineId, Vec<(Lsn, TimelineId, MaybeOffloaded)>> =
-            BTreeMap::new();
+        let mut all_branchpoints: BTreeMap<TimelineId, Vec<(Lsn, TimelineId)>> = BTreeMap::new();
        timelines.iter().for_each(|(timeline_id, timeline_entry)| {
            if let Some(ancestor_timeline_id) = &timeline_entry.get_ancestor_timeline_id() {
                let ancestor_children = all_branchpoints.entry(*ancestor_timeline_id).or_default();
-                ancestor_children.push((
-                    timeline_entry.get_ancestor_lsn(),
-                    *timeline_id,
-                    MaybeOffloaded::No,
-                ));
+                ancestor_children.push((timeline_entry.get_ancestor_lsn(), *timeline_id));
            }
        });
-        timelines_offloaded
-            .iter()
-            .for_each(|(timeline_id, timeline_entry)| {
-                let Some(ancestor_timeline_id) = &timeline_entry.ancestor_timeline_id else {
-                    return;
-                };
-                let Some(retain_lsn) = timeline_entry.ancestor_retain_lsn else {
-                    return;
-                };
-                let ancestor_children = all_branchpoints.entry(*ancestor_timeline_id).or_default();
-                ancestor_children.push((retain_lsn, *timeline_id, MaybeOffloaded::Yes));
-            });

        // The number of bytes we always keep, irrespective of PITR: this is a constant across timelines
        let horizon = self.get_gc_horizon();

        // Populate each timeline's GcInfo with information about its child branches
        for timeline in timelines.values() {
-            let mut branchpoints: Vec<(Lsn, TimelineId, MaybeOffloaded)> = all_branchpoints
+            let mut branchpoints: Vec<(Lsn, TimelineId)> = all_branchpoints
                .remove(&timeline.timeline_id)
                .unwrap_or_default();

@@ -4297,8 +4011,7 @@ pub(crate) mod harness {
    use crate::deletion_queue::mock::MockDeletionQueue;
    use crate::l0_flush::L0FlushConfig;
    use crate::walredo::apply_neon;
-    use pageserver_api::key::Key;
-    use pageserver_api::record::NeonWalRecord;
+    use crate::{repository::Key, walrecord::NeonWalRecord};

    use super::*;
    use hex_literal::hex;
@@ -4569,17 +4282,17 @@ mod tests {
    use super::*;
    use crate::keyspace::KeySpaceAccum;
    use crate::pgdatadir_mapping::AuxFilesDirectory;
+    use crate::repository::{Key, Value};
    use crate::tenant::harness::*;
    use crate::tenant::timeline::CompactFlags;
+    use crate::walrecord::NeonWalRecord;
    use crate::DEFAULT_PG_VERSION;
    use bytes::{Bytes, BytesMut};
    use hex_literal::hex;
    use itertools::Itertools;
-    use pageserver_api::key::{Key, AUX_FILES_KEY, AUX_KEY_PREFIX, NON_INHERITED_RANGE};
+    use pageserver_api::key::{AUX_FILES_KEY, AUX_KEY_PREFIX, NON_INHERITED_RANGE};
    use pageserver_api::keyspace::KeySpace;
    use pageserver_api::models::{CompactionAlgorithm, CompactionAlgorithmSettings};
-    use pageserver_api::record::NeonWalRecord;
-    use pageserver_api::value::Value;
    use rand::{thread_rng, Rng};
    use storage_layer::PersistentLayerKey;
    use tests::storage_layer::ValuesReconstructState;
@@ -4939,10 +4652,7 @@ mod tests {
        {
            let branchpoints = &tline.gc_info.read().unwrap().retain_lsns;
            assert_eq!(branchpoints.len(), 1);
-            assert_eq!(
-                branchpoints[0],
-                (Lsn(0x40), NEW_TIMELINE_ID, MaybeOffloaded::No)
-            );
+            assert_eq!(branchpoints[0], (Lsn(0x40), NEW_TIMELINE_ID));
        }

        // You can read the key from the child branch even though the parent is
@@ -8325,8 +8035,8 @@ mod tests {
            let mut guard = tline.gc_info.write().unwrap();
            *guard = GcInfo {
                retain_lsns: vec![
-                    (Lsn(0x10), tline.timeline_id, MaybeOffloaded::No),
-                    (Lsn(0x20), tline.timeline_id, MaybeOffloaded::No),
+                    (Lsn(0x10), tline.timeline_id),
+                    (Lsn(0x20), tline.timeline_id),
                ],
                cutoffs: GcCutoffs {
                    time: Lsn(0x30),
@@ -8553,8 +8263,8 @@ mod tests {
            let mut guard = tline.gc_info.write().unwrap();
            *guard = GcInfo {
                retain_lsns: vec![
-                    (Lsn(0x10), tline.timeline_id, MaybeOffloaded::No),
-                    (Lsn(0x20), tline.timeline_id, MaybeOffloaded::No),
+                    (Lsn(0x10), tline.timeline_id),
+                    (Lsn(0x20), tline.timeline_id),
                ],
                cutoffs: GcCutoffs {
                    time: Lsn(0x30),
@@ -8787,7 +8497,7 @@ mod tests {
            // Update GC info
            let mut guard = parent_tline.gc_info.write().unwrap();
            *guard = GcInfo {
-                retain_lsns: vec![(Lsn(0x18), branch_tline.timeline_id, MaybeOffloaded::No)],
+                retain_lsns: vec![(Lsn(0x18), branch_tline.timeline_id)],
                cutoffs: GcCutoffs {
                    time: Lsn(0x10),
                    space: Lsn(0x10),
@@ -8801,7 +8511,7 @@ mod tests {
            // Update GC info
            let mut guard = branch_tline.gc_info.write().unwrap();
            *guard = GcInfo {
-                retain_lsns: vec![(Lsn(0x40), branch_tline.timeline_id, MaybeOffloaded::No)],
+                retain_lsns: vec![(Lsn(0x40), branch_tline.timeline_id)],
                cutoffs: GcCutoffs {
                    time: Lsn(0x50),
                    space: Lsn(0x50),
--- a/pageserver/src/tenant/ephemeral_file.rs
+++ b/pageserver/src/tenant/ephemeral_file.rs
@@ -84,7 +84,7 @@ impl Drop for EphemeralFile {
    fn drop(&mut self) {
        // unlink the file
        // we are clear to do this, because we have entered a gate
-        let path = self.buffered_writer.as_inner().as_inner().path();
+        let path = &self.buffered_writer.as_inner().as_inner().path;
        let res = std::fs::remove_file(path);
        if let Err(e) = res {
            if e.kind() != std::io::ErrorKind::NotFound {
@@ -356,7 +356,7 @@ mod tests {
        }

        let file_contents =
-            std::fs::read(file.buffered_writer.as_inner().as_inner().path()).unwrap();
+            std::fs::read(&file.buffered_writer.as_inner().as_inner().path).unwrap();
        assert_eq!(file_contents, &content[0..cap]);

        let buffer_contents = file.buffered_writer.inspect_buffer();
@@ -392,7 +392,7 @@ mod tests {
            .buffered_writer
            .as_inner()
            .as_inner()
-            .path()
+            .path
            .metadata()
            .unwrap();
        assert_eq!(
--- a/pageserver/src/tenant/gc_block.rs
+++ b/pageserver/src/tenant/gc_block.rs
@@ -141,14 +141,14 @@ impl GcBlock {
        Ok(())
    }

-    pub(crate) fn before_delete(&self, timeline_id: &super::TimelineId) {
+    pub(crate) fn before_delete(&self, timeline: &super::Timeline) {
        let unblocked = {
            let mut g = self.reasons.lock().unwrap();
            if g.is_empty() {
                return;
            }

-            g.remove(timeline_id);
+            g.remove(&timeline.timeline_id);

            BlockingReasons::clean_and_summarize(g).is_none()
        };
--- a/pageserver/src/tenant/layer_map.rs
+++ b/pageserver/src/tenant/layer_map.rs
@@ -48,9 +48,9 @@ mod layer_coverage;

 use crate::context::RequestContext;
 use crate::keyspace::KeyPartitioning;
+use crate::repository::Key;
 use crate::tenant::storage_layer::InMemoryLayer;
 use anyhow::Result;
-use pageserver_api::key::Key;
 use pageserver_api::keyspace::{KeySpace, KeySpaceAccum};
 use range_set_blaze::{CheckSortedDisjoint, RangeSetBlaze};
 use std::collections::{HashMap, VecDeque};
--- a/pageserver/src/tenant/mgr.rs
+++ b/pageserver/src/tenant/mgr.rs
@@ -2836,7 +2836,7 @@ where
 }

 use {
-    crate::gc_result::GcResult, pageserver_api::models::TimelineGcRequest,
+    crate::repository::GcResult, pageserver_api::models::TimelineGcRequest,
    utils::http::error::ApiError,
 };

--- a/pageserver/src/tenant/remote_timeline_client/download.rs
+++ b/pageserver/src/tenant/remote_timeline_client/download.rs
@@ -27,7 +27,7 @@ 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, DownloadOpts, GenericRemoteStorage, ListingMode, RemotePath};
+use remote_storage::{DownloadError, GenericRemoteStorage, ListingMode, RemotePath};
 use utils::crashsafe::path_with_suffix_extension;
 use utils::id::{TenantId, TimelineId};
 use utils::pausable_failpoint;
@@ -153,9 +153,7 @@ async fn download_object<'a>(
                    .with_context(|| format!("create a destination file for layer '{dst_path}'"))
                    .map_err(DownloadError::Other)?;

-                let download = storage
-                    .download(src_path, &DownloadOpts::default(), cancel)
-                    .await?;
+                let download = storage.download(src_path, cancel).await?;

                pausable_failpoint!("before-downloading-layer-stream-pausable");

@@ -206,9 +204,7 @@ async fn download_object<'a>(
                    .with_context(|| format!("create a destination file for layer '{dst_path}'"))
                    .map_err(DownloadError::Other)?;

-                let mut download = storage
-                    .download(src_path, &DownloadOpts::default(), cancel)
-                    .await?;
+                let mut download = storage.download(src_path, cancel).await?;

                pausable_failpoint!("before-downloading-layer-stream-pausable");

@@ -348,9 +344,7 @@ async fn do_download_index_part(

    let index_part_bytes = download_retry_forever(
        || async {
-            let download = storage
-                .download(&remote_path, &DownloadOpts::default(), cancel)
-                .await?;
+            let download = storage.download(&remote_path, cancel).await?;

            let mut bytes = Vec::new();

@@ -532,15 +526,10 @@ pub(crate) async fn download_initdb_tar_zst(
                .with_context(|| format!("tempfile creation {temp_path}"))
                .map_err(DownloadError::Other)?;

-            let download = match storage
-                .download(&remote_path, &DownloadOpts::default(), cancel)
-                .await
-            {
+            let download = match storage.download(&remote_path, cancel).await {
                Ok(dl) => dl,
                Err(DownloadError::NotFound) => {
-                    storage
-                        .download(&remote_preserved_path, &DownloadOpts::default(), cancel)
-                        .await?
+                    storage.download(&remote_preserved_path, cancel).await?
                }
                Err(other) => Err(other)?,
            };
--- a/pageserver/src/tenant/secondary/downloader.rs
+++ b/pageserver/src/tenant/secondary/downloader.rs
@@ -49,7 +49,7 @@ use futures::Future;
 use metrics::UIntGauge;
 use pageserver_api::models::SecondaryProgress;
 use pageserver_api::shard::TenantShardId;
-use remote_storage::{DownloadError, DownloadOpts, Etag, GenericRemoteStorage};
+use remote_storage::{DownloadError, Etag, GenericRemoteStorage};

 use tokio_util::sync::CancellationToken;
 use tracing::{info_span, instrument, warn, Instrument};
@@ -944,35 +944,36 @@ impl<'a> TenantDownloader<'a> {
    ) -> Result<HeatMapDownload, UpdateError> {
        debug_assert_current_span_has_tenant_id();
        let tenant_shard_id = self.secondary_state.get_tenant_shard_id();
+        // TODO: pull up etag check into the request, to do a conditional GET rather than
+        // issuing a GET and then maybe ignoring the response body
+        // (https://github.com/neondatabase/neon/issues/6199)
        tracing::debug!("Downloading heatmap for secondary tenant",);

        let heatmap_path = remote_heatmap_path(tenant_shard_id);
        let cancel = &self.secondary_state.cancel;
-        let opts = DownloadOpts {
-            etag: prev_etag.cloned(),
-            ..Default::default()
-        };

        backoff::retry(
            || async {
-                let download = match self
+                let download = self
                    .remote_storage
-                    .download(&heatmap_path, &opts, cancel)
+                    .download(&heatmap_path, cancel)
                    .await
-                {
-                    Ok(download) => download,
-                    Err(DownloadError::Unmodified) => return Ok(HeatMapDownload::Unmodified),
-                    Err(err) => return Err(err.into()),
-                };
+                    .map_err(UpdateError::from)?;

-                let mut heatmap_bytes = Vec::new();
-                let mut body = tokio_util::io::StreamReader::new(download.download_stream);
-                let _size = tokio::io::copy_buf(&mut body, &mut heatmap_bytes).await?;
-                Ok(HeatMapDownload::Modified(HeatMapModified {
-                    etag: download.etag,
-                    last_modified: download.last_modified,
-                    bytes: heatmap_bytes,
-                }))
+                SECONDARY_MODE.download_heatmap.inc();
+
+                if Some(&download.etag) == prev_etag {
+                    Ok(HeatMapDownload::Unmodified)
+                } else {
+                    let mut heatmap_bytes = Vec::new();
+                    let mut body = tokio_util::io::StreamReader::new(download.download_stream);
+                    let _size = tokio::io::copy_buf(&mut body, &mut heatmap_bytes).await?;
+                    Ok(HeatMapDownload::Modified(HeatMapModified {
+                        etag: download.etag,
+                        last_modified: download.last_modified,
+                        bytes: heatmap_bytes,
+                    }))
+                }
            },
            |e| matches!(e, UpdateError::NoData | UpdateError::Cancelled),
            FAILED_DOWNLOAD_WARN_THRESHOLD,
@@ -983,7 +984,6 @@ impl<'a> TenantDownloader<'a> {
        .await
        .ok_or_else(|| UpdateError::Cancelled)
        .and_then(|x| x)
-        .inspect(|_| SECONDARY_MODE.download_heatmap.inc())
    }

    /// Download heatmap layers that are not present on local disk, or update their
--- a/pageserver/src/tenant/size.rs
+++ b/pageserver/src/tenant/size.rs
@@ -12,7 +12,7 @@ use crate::context::RequestContext;
 use crate::pgdatadir_mapping::CalculateLogicalSizeError;

 use super::{GcError, LogicalSizeCalculationCause, Tenant};
-use crate::tenant::{MaybeOffloaded, Timeline};
+use crate::tenant::Timeline;
 use utils::id::TimelineId;
 use utils::lsn::Lsn;

@@ -264,12 +264,10 @@ pub(super) async fn gather_inputs(
        let mut lsns: Vec<(Lsn, LsnKind)> = gc_info
            .retain_lsns
            .iter()
-            .filter(|(lsn, _child_id, is_offloaded)| {
-                lsn > &ancestor_lsn && *is_offloaded == MaybeOffloaded::No
-            })
+            .filter(|(lsn, _child_id)| lsn > &ancestor_lsn)
            .copied()
            // this assumes there are no other retain_lsns than the branchpoints
-            .map(|(lsn, _child_id, _is_offloaded)| (lsn, LsnKind::BranchPoint))
+            .map(|(lsn, _child_id)| (lsn, LsnKind::BranchPoint))
            .collect::<Vec<_>>();

        lsns.extend(lease_points.iter().map(|&lsn| (lsn, LsnKind::LeasePoint)));
--- a/pageserver/src/tenant/storage_layer.rs
+++ b/pageserver/src/tenant/storage_layer.rs
@@ -11,11 +11,11 @@ pub mod merge_iterator;
 pub mod split_writer;

 use crate::context::{AccessStatsBehavior, RequestContext};
+use crate::repository::Value;
+use crate::walrecord::NeonWalRecord;
 use bytes::Bytes;
 use pageserver_api::key::Key;
 use pageserver_api::keyspace::{KeySpace, KeySpaceRandomAccum};
-use pageserver_api::record::NeonWalRecord;
-use pageserver_api::value::Value;
 use std::cmp::{Ordering, Reverse};
 use std::collections::hash_map::Entry;
 use std::collections::{BinaryHeap, HashMap};
--- a/pageserver/src/tenant/storage_layer/delta_layer.rs
+++ b/pageserver/src/tenant/storage_layer/delta_layer.rs
@@ -30,6 +30,7 @@
 use crate::config::PageServerConf;
 use crate::context::{PageContentKind, RequestContext, RequestContextBuilder};
 use crate::page_cache::{self, FileId, PAGE_SZ};
+use crate::repository::{Key, Value, KEY_SIZE};
 use crate::tenant::blob_io::BlobWriter;
 use crate::tenant::block_io::{BlockBuf, BlockCursor, BlockLease, BlockReader, FileBlockReader};
 use crate::tenant::disk_btree::{
@@ -44,7 +45,7 @@ use crate::tenant::vectored_blob_io::{
 use crate::tenant::PageReconstructError;
 use crate::virtual_file::owned_buffers_io::io_buf_ext::{FullSlice, IoBufExt};
 use crate::virtual_file::{self, MaybeFatalIo, VirtualFile};
-use crate::TEMP_FILE_SUFFIX;
+use crate::{walrecord, TEMP_FILE_SUFFIX};
 use crate::{DELTA_FILE_MAGIC, STORAGE_FORMAT_VERSION};
 use anyhow::{anyhow, bail, ensure, Context, Result};
 use bytes::BytesMut;
@@ -52,12 +53,9 @@ use camino::{Utf8Path, Utf8PathBuf};
 use futures::StreamExt;
 use itertools::Itertools;
 use pageserver_api::config::MaxVectoredReadBytes;
-use pageserver_api::key::DBDIR_KEY;
-use pageserver_api::key::{Key, KEY_SIZE};
 use pageserver_api::keyspace::KeySpace;
 use pageserver_api::models::ImageCompressionAlgorithm;
 use pageserver_api::shard::TenantShardId;
-use pageserver_api::value::Value;
 use rand::{distributions::Alphanumeric, Rng};
 use serde::{Deserialize, Serialize};
 use std::collections::VecDeque;
@@ -574,7 +572,7 @@ impl DeltaLayerWriterInner {
        ensure!(
            metadata.len() <= S3_UPLOAD_LIMIT,
            "Created delta layer file at {} of size {} above limit {S3_UPLOAD_LIMIT}!",
-            file.path(),
+            file.path,
            metadata.len()
        );

@@ -792,7 +790,7 @@ impl DeltaLayerInner {
        max_vectored_read_bytes: Option<MaxVectoredReadBytes>,
        ctx: &RequestContext,
    ) -> anyhow::Result<Self> {
-        let file = VirtualFile::open_v2(path, ctx)
+        let file = VirtualFile::open(path, ctx)
            .await
            .context("open layer file")?;

@@ -965,25 +963,14 @@ impl DeltaLayerInner {
                .blobs_at
                .as_slice()
                .iter()
-                .filter_map(|(_, blob_meta)| {
-                    if blob_meta.key.is_rel_dir_key() || blob_meta.key == DBDIR_KEY {
-                        // The size of values for these keys is unbounded and can
-                        // grow very large in pathological cases.
-                        None
-                    } else {
-                        Some(format!("{}@{}", blob_meta.key, blob_meta.lsn))
-                    }
-                })
+                .map(|(_, blob_meta)| format!("{}@{}", blob_meta.key, blob_meta.lsn))
                .join(", ");
-
-            if !offenders.is_empty() {
-                tracing::warn!(
-                    "Oversized vectored read ({} > {}) for keys {}",
-                    largest_read_size,
-                    read_size_soft_max,
-                    offenders
-                );
-            }
+            tracing::warn!(
+                "Oversized vectored read ({} > {}) for keys {}",
+                largest_read_size,
+                read_size_soft_max,
+                offenders
+            );
        }

        largest_read_size
@@ -1023,7 +1010,7 @@ impl DeltaLayerInner {
                            blob_meta.key,
                            PageReconstructError::Other(anyhow!(
                                "Failed to read blobs from virtual file {}: {}",
-                                self.file.path(),
+                                self.file.path,
                                kind
                            )),
                        );
@@ -1049,7 +1036,7 @@ impl DeltaLayerInner {
                            meta.meta.key,
                            PageReconstructError::Other(anyhow!(e).context(format!(
                                "Failed to decompress blob from virtual file {}",
-                                self.file.path(),
+                                self.file.path,
                            ))),
                        );

@@ -1067,7 +1054,7 @@ impl DeltaLayerInner {
                            meta.meta.key,
                            PageReconstructError::Other(anyhow!(e).context(format!(
                                "Failed to deserialize blob from virtual file {}",
-                                self.file.path(),
+                                self.file.path,
                            ))),
                        );

@@ -1199,6 +1186,7 @@ impl DeltaLayerInner {
        let mut prev: Option<(Key, Lsn, BlobRef)> = None;

        let mut read_builder: Option<ChunkedVectoredReadBuilder> = None;
+        let align = virtual_file::get_io_buffer_alignment();

        let max_read_size = self
            .max_vectored_read_bytes
@@ -1247,6 +1235,7 @@ impl DeltaLayerInner {
                        offsets.end.pos(),
                        meta,
                        max_read_size,
+                        align,
                    ))
                }
            } else {
@@ -1295,7 +1284,7 @@ impl DeltaLayerInner {
                    // is it an image or will_init walrecord?
                    // FIXME: this could be handled by threading the BlobRef to the
                    // VectoredReadBuilder
-                    let will_init = pageserver_api::value::ValueBytes::will_init(&data)
+                    let will_init = crate::repository::ValueBytes::will_init(&data)
                        .inspect_err(|_e| {
                            #[cfg(feature = "testing")]
                            tracing::error!(data=?utils::Hex(&data), err=?_e, %key, %lsn, "failed to parse will_init out of serialized value");
@@ -1358,7 +1347,7 @@ impl DeltaLayerInner {
                    format!(" img {} bytes", img.len())
                }
                Value::WalRecord(rec) => {
-                    let wal_desc = pageserver_api::record::describe_wal_record(&rec)?;
+                    let wal_desc = walrecord::describe_wal_record(&rec)?;
                    format!(
                        " rec {} bytes will_init: {} {}",
                        buf.len(),
@@ -1603,6 +1592,7 @@ pub(crate) mod test {
    use rand::RngCore;

    use super::*;
+    use crate::repository::Value;
    use crate::tenant::harness::TIMELINE_ID;
    use crate::tenant::storage_layer::{Layer, ResidentLayer};
    use crate::tenant::vectored_blob_io::StreamingVectoredReadPlanner;
@@ -1614,7 +1604,6 @@ pub(crate) mod test {
        DEFAULT_PG_VERSION,
    };
    use bytes::Bytes;
-    use pageserver_api::value::Value;

    /// Construct an index for a fictional delta layer and and then
    /// traverse in order to plan vectored reads for a query. Finally,
@@ -1967,8 +1956,8 @@ pub(crate) mod test {

    #[tokio::test]
    async fn copy_delta_prefix_smoke() {
+        use crate::walrecord::NeonWalRecord;
        use bytes::Bytes;
-        use pageserver_api::record::NeonWalRecord;

        let h = crate::tenant::harness::TenantHarness::create("truncate_delta_smoke")
            .await
--- a/pageserver/src/tenant/storage_layer/filter_iterator.rs
+++ b/pageserver/src/tenant/storage_layer/filter_iterator.rs
@@ -7,7 +7,7 @@ use pageserver_api::{
 };
 use utils::lsn::Lsn;

-use pageserver_api::value::Value;
+use crate::repository::Value;

 use super::merge_iterator::MergeIterator;

@@ -121,8 +121,8 @@ mod tests {

    #[tokio::test]
    async fn filter_keyspace_iterator() {
+        use crate::repository::Value;
        use bytes::Bytes;
-        use pageserver_api::value::Value;

        let harness = TenantHarness::create("filter_iterator_filter_keyspace_iterator")
            .await
--- a/pageserver/src/tenant/storage_layer/image_layer.rs
+++ b/pageserver/src/tenant/storage_layer/image_layer.rs
@@ -28,6 +28,7 @@
 use crate::config::PageServerConf;
 use crate::context::{PageContentKind, RequestContext, RequestContextBuilder};
 use crate::page_cache::{self, FileId, PAGE_SZ};
+use crate::repository::{Key, Value, KEY_SIZE};
 use crate::tenant::blob_io::BlobWriter;
 use crate::tenant::block_io::{BlockBuf, FileBlockReader};
 use crate::tenant::disk_btree::{
@@ -48,11 +49,8 @@ use camino::{Utf8Path, Utf8PathBuf};
 use hex;
 use itertools::Itertools;
 use pageserver_api::config::MaxVectoredReadBytes;
-use pageserver_api::key::DBDIR_KEY;
-use pageserver_api::key::{Key, KEY_SIZE};
 use pageserver_api::keyspace::KeySpace;
 use pageserver_api::shard::{ShardIdentity, TenantShardId};
-use pageserver_api::value::Value;
 use rand::{distributions::Alphanumeric, Rng};
 use serde::{Deserialize, Serialize};
 use std::collections::VecDeque;
@@ -390,7 +388,7 @@ impl ImageLayerInner {
        max_vectored_read_bytes: Option<MaxVectoredReadBytes>,
        ctx: &RequestContext,
    ) -> anyhow::Result<Self> {
-        let file = VirtualFile::open_v2(path, ctx)
+        let file = VirtualFile::open(path, ctx)
            .await
            .context("open layer file")?;
        let file_id = page_cache::next_file_id();
@@ -589,25 +587,14 @@ impl ImageLayerInner {
                    .blobs_at
                    .as_slice()
                    .iter()
-                    .filter_map(|(_, blob_meta)| {
-                        if blob_meta.key.is_rel_dir_key() || blob_meta.key == DBDIR_KEY {
-                            // The size of values for these keys is unbounded and can
-                            // grow very large in pathological cases.
-                            None
-                        } else {
-                            Some(format!("{}@{}", blob_meta.key, blob_meta.lsn))
-                        }
-                    })
+                    .map(|(_, blob_meta)| format!("{}@{}", blob_meta.key, blob_meta.lsn))
                    .join(", ");
-
-                if !offenders.is_empty() {
-                    tracing::warn!(
-                        "Oversized vectored read ({} > {}) for keys {}",
-                        buf_size,
-                        max_vectored_read_bytes,
-                        offenders
-                    );
-                }
+                tracing::warn!(
+                    "Oversized vectored read ({} > {}) for keys {}",
+                    buf_size,
+                    max_vectored_read_bytes,
+                    offenders
+                );
            }

            let buf = BytesMut::with_capacity(buf_size);
@@ -627,7 +614,7 @@ impl ImageLayerInner {
                                    meta.meta.key,
                                    PageReconstructError::Other(anyhow!(e).context(format!(
                                        "Failed to decompress blob from virtual file {}",
-                                        self.file.path(),
+                                        self.file.path,
                                    ))),
                                );

@@ -648,7 +635,7 @@ impl ImageLayerInner {
                            blob_meta.key,
                            PageReconstructError::from(anyhow!(
                                "Failed to read blobs from virtual file {}: {}",
-                                self.file.path(),
+                                self.file.path,
                                kind
                            )),
                        );
@@ -1094,7 +1081,6 @@ mod test {
    use pageserver_api::{
        key::Key,
        shard::{ShardCount, ShardIdentity, ShardNumber, ShardStripeSize},
-        value::Value,
    };
    use utils::{
        generation::Generation,
@@ -1104,6 +1090,7 @@ mod test {

    use crate::{
        context::RequestContext,
+        repository::Value,
        tenant::{
            config::TenantConf,
            harness::{TenantHarness, TIMELINE_ID},
--- a/pageserver/src/tenant/storage_layer/inmemory_layer.rs
+++ b/pageserver/src/tenant/storage_layer/inmemory_layer.rs
@@ -7,6 +7,7 @@
 use crate::assert_u64_eq_usize::{u64_to_usize, U64IsUsize, UsizeIsU64};
 use crate::config::PageServerConf;
 use crate::context::{PageContentKind, RequestContext, RequestContextBuilder};
+use crate::repository::{Key, Value};
 use crate::tenant::ephemeral_file::EphemeralFile;
 use crate::tenant::timeline::GetVectoredError;
 use crate::tenant::PageReconstructError;
@@ -16,11 +17,9 @@ use anyhow::{anyhow, Context, Result};
 use bytes::Bytes;
 use camino::Utf8PathBuf;
 use pageserver_api::key::CompactKey;
-use pageserver_api::key::Key;
 use pageserver_api::keyspace::KeySpace;
 use pageserver_api::models::InMemoryLayerInfo;
 use pageserver_api::shard::TenantShardId;
-use pageserver_api::value::Value;
 use std::collections::{BTreeMap, HashMap};
 use std::sync::{Arc, OnceLock};
 use std::time::Instant;
--- a/pageserver/src/tenant/storage_layer/layer/tests.rs
+++ b/pageserver/src/tenant/storage_layer/layer/tests.rs
@@ -760,8 +760,8 @@ async fn evict_and_wait_does_not_wait_for_download() {
 /// Also checks that the same does not happen on a non-evicted layer (regression test).
 #[tokio::test(start_paused = true)]
 async fn eviction_cancellation_on_drop() {
+    use crate::repository::Value;
    use bytes::Bytes;
-    use pageserver_api::value::Value;

    // this is the runtime on which Layer spawns the blocking tasks on
    let handle = tokio::runtime::Handle::current();
@@ -782,7 +782,7 @@ async fn eviction_cancellation_on_drop() {
        let mut writer = timeline.writer().await;
        writer
            .put(
-                pageserver_api::key::Key::from_i128(5),
+                crate::repository::Key::from_i128(5),
                Lsn(0x20),
                &Value::Image(Bytes::from_static(b"this does not matter either")),
                &ctx,
--- a/pageserver/src/tenant/storage_layer/layer_desc.rs
+++ b/pageserver/src/tenant/storage_layer/layer_desc.rs
@@ -3,7 +3,7 @@ use pageserver_api::shard::TenantShardId;
 use std::ops::Range;
 use utils::{id::TimelineId, lsn::Lsn};

-use pageserver_api::key::Key;
+use crate::repository::Key;

 use super::{DeltaLayerName, ImageLayerName, LayerName};

--- a/Show More
+++ b/Show More
Author	SHA1	Message	Date
John Spray	62aa8f2ba2	hacky way to make tests pass	2024-10-04 09:56:56 +00:00
John Spray	efc3f1cfe5	DNM: test-log	2024-10-03 11:29:06 +00:00
John Spray	b66bf890fa	storage controller: fix a disagreement between schedule + optimize Closes: https://github.com/neondatabase/neon/issues/8969	2024-10-03 11:14:34 +00:00