Reduce noise from moto GET/PUT operations

Moto prints messages like this: 127.0.0.1 - - [07/Oct/2024 12:35:16] "PUT /bucket-name/path?x-id=PutObject HTTP/1.1" 200 - After the root logger adds its context information, this is what actually gets printed to the log: 2024-10-07 22:35:16.371 INFO [_internal.py:97] 127.0.0.1 - - [07/Oct/2024 22:35:16] "PUT /bucket-name/path?x-id=PutObject HTTP/1.1" 200 - That's very verbose. Remove the hostname and the extra timestamp, to make it a little less verbose. With this PR, the final output looks like this: 2024-10-07 22:35:16.371 INFO [_internal.py:97] "PUT /bucket-name/path?x-id=PutObject HTTP/1.1" 200 -
2026-05-21 07:00:38 +00:00 · 2024-10-09 18:53:08 +03:00
137 changed files with 3025 additions and 5983 deletions
--- 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
@@ -33,7 +33,7 @@ jobs:
      actions: read
    steps:
    - name: Export GH Workflow Stats
-      uses: neondatabase/gh-workflow-stats-action@v0.1.4
+      uses: fedordikarev/gh-workflow-stats-action@v0.1.2
      with:
        DB_URI: ${{ secrets.GH_REPORT_STATS_DB_RW_CONNSTR }}
        DB_TABLE: "gh_workflow_stats_neon"
--- 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
@@ -1820,7 +1820,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 +1829,6 @@ dependencies = [
 "pkcs8 0.10.2",
 "rand_core 0.6.4",
 "sec1 0.7.3",
- "serde_json",
- "serdect",
 "subtle",
 "zeroize",
 ]
@@ -2695,7 +2692,6 @@ checksum = "ad227c3af19d4914570ad36d30409928b75967c298feb9ea1969db3a610bb14e"
 dependencies = [
 "equivalent",
 "hashbrown 0.14.5",
- "serde",
 ]

 [[package]]
@@ -2795,9 +2791,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"
@@ -3656,7 +3652,6 @@ dependencies = [
 "tracing",
 "url",
 "utils",
- "wal_decoder",
 "walkdir",
 "workspace_hack",
 ]
@@ -4042,8 +4037,6 @@ dependencies = [
 "bytes",
 "fallible-iterator",
 "postgres-protocol",
- "serde",
- "serde_json",
 ]

 [[package]]
@@ -4094,7 +4087,6 @@ dependencies = [
 "regex",
 "serde",
 "thiserror",
- "tracing",
 "utils",
 ]

@@ -4299,7 +4291,6 @@ dependencies = [
 "indexmap 2.0.1",
 "ipnet",
 "itertools 0.10.5",
- "itoa",
 "jose-jwa",
 "jose-jwk",
 "lasso",
@@ -5265,7 +5256,6 @@ dependencies = [
 "der 0.7.8",
 "generic-array",
 "pkcs8 0.10.2",
- "serdect",
 "subtle",
 "zeroize",
 ]
@@ -5520,16 +5510,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"
@@ -6862,19 +6842,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"
@@ -7324,7 +7291,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 +7302,6 @@ dependencies = [
 "num-traits",
 "once_cell",
 "parquet",
- "postgres-types",
 "prettyplease",
 "proc-macro2",
 "prost",
@@ -7361,7 +7326,6 @@ dependencies = [
 "time",
 "time-macros",
 "tokio",
- "tokio-postgres",
 "tokio-stream",
 "tokio-util",
 "toml_edit",
--- 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]
@@ -108,7 +107,6 @@ 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/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,8 +875,8 @@ 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
@@ -1095,6 +1041,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 +1052,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 \
@@ -1259,7 +1207,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 && \
@@ -1307,22 +1255,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
--- 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));
        }

--- 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/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/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/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
@@ -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},
--- 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/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};

--- 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};
--- 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/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
@@ -77,7 +77,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 +325,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())
            }
@@ -717,8 +715,6 @@ 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)
            .await?;
@@ -1787,49 +1783,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 +2003,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)
        }
    }

@@ -3053,10 +3006,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),
--- 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
@@ -20,11 +20,11 @@ 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 +32,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
@@ -79,7 +79,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 +87,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;
@@ -461,10 +461,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 {
@@ -493,8 +493,6 @@ 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>,
@@ -506,14 +504,10 @@ pub struct OffloadedTimeline {

 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(),
@@ -521,12 +515,6 @@ impl OffloadedTimeline {
    }
 }

-#[derive(Copy, Clone, PartialEq, Eq, Hash, Debug)]
-pub enum MaybeOffloaded {
-    Yes,
-    No,
-}
-
 #[derive(Clone)]
 pub enum TimelineOrOffloaded {
    Timeline(Arc<Timeline>),
@@ -619,9 +607,6 @@ pub enum TimelineArchivalError {
    #[error("Timeout")]
    Timeout,

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

@@ -640,7 +625,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()
@@ -1556,7 +1540,6 @@ impl Tenant {
        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)
@@ -1571,7 +1554,6 @@ impl Tenant {
                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})");
@@ -1609,7 +1591,6 @@ impl Tenant {
            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");
@@ -1690,21 +1671,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
    }
@@ -2240,13 +2206,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 +2253,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.
@@ -3340,7 +3298,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 +3305,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 +4237,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 +4508,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 +4878,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 +8261,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 +8489,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 +8723,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 +8737,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/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/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;
@@ -53,11 +54,9 @@ 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;
@@ -1295,7 +1294,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 +1357,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 +1602,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 +1614,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 +1966,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::{
@@ -49,10 +50,8 @@ 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;
@@ -1094,7 +1093,6 @@ mod test {
    use pageserver_api::{
        key::Key,
        shard::{ShardCount, ShardIdentity, ShardNumber, ShardStripeSize},
-        value::Value,
    };
    use utils::{
        generation::Generation,
@@ -1104,6 +1102,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/pageserver/src/tenant/storage_layer/layer_name.rs
+++ b/pageserver/src/tenant/storage_layer/layer_name.rs
@@ -1,7 +1,7 @@
 //!
 //! Helper functions for dealing with filenames of the image and delta layer files.
 //!
-use pageserver_api::key::Key;
+use crate::repository::Key;
 use std::borrow::Cow;
 use std::cmp::Ordering;
 use std::fmt;
--- a/pageserver/src/tenant/storage_layer/merge_iterator.rs
+++ b/pageserver/src/tenant/storage_layer/merge_iterator.rs
@@ -7,8 +7,7 @@ use anyhow::bail;
 use pageserver_api::key::Key;
 use utils::lsn::Lsn;

-use crate::context::RequestContext;
-use pageserver_api::value::Value;
+use crate::{context::RequestContext, repository::Value};

 use super::{
    delta_layer::{DeltaLayerInner, DeltaLayerIterator},
@@ -294,9 +293,9 @@ mod tests {
            harness::{TenantHarness, TIMELINE_ID},
            storage_layer::delta_layer::test::{produce_delta_layer, sort_delta, sort_delta_value},
        },
+        walrecord::NeonWalRecord,
        DEFAULT_PG_VERSION,
    };
-    use pageserver_api::record::NeonWalRecord;

    async fn assert_merge_iter_equal(
        merge_iter: &mut MergeIterator<'_>,
@@ -320,8 +319,8 @@ mod tests {

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

        let harness = TenantHarness::create("merge_iterator_merge_in_between")
            .await
@@ -385,8 +384,8 @@ mod tests {

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

        let harness = TenantHarness::create("merge_iterator_delta_merge")
            .await
@@ -461,8 +460,8 @@ mod tests {

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

        let harness = TenantHarness::create("merge_iterator_delta_image_mixed_merge")
            .await
--- a/pageserver/src/tenant/storage_layer/split_writer.rs
+++ b/pageserver/src/tenant/storage_layer/split_writer.rs
@@ -5,8 +5,7 @@ use pageserver_api::key::{Key, KEY_SIZE};
 use utils::{id::TimelineId, lsn::Lsn, shard::TenantShardId};

 use crate::tenant::storage_layer::Layer;
-use crate::{config::PageServerConf, context::RequestContext, tenant::Timeline};
-use pageserver_api::value::Value;
+use crate::{config::PageServerConf, context::RequestContext, repository::Value, tenant::Timeline};

 use super::layer::S3_UPLOAD_LIMIT;
 use super::{
--- a/pageserver/src/tenant/timeline.rs
+++ b/pageserver/src/tenant/timeline.rs
@@ -125,12 +125,11 @@ use utils::{
    simple_rcu::{Rcu, RcuReadGuard},
 };

-use crate::gc_result::GcResult;
+use crate::repository::GcResult;
+use crate::repository::{Key, Value};
 use crate::task_mgr;
 use crate::task_mgr::TaskKind;
 use crate::ZERO_PAGE;
-use pageserver_api::key::Key;
-use pageserver_api::value::Value;

 use self::delete::DeleteTimelineFlow;
 pub(super) use self::eviction_task::EvictionTaskTenantState;
@@ -140,10 +139,8 @@ use self::logical_size::LogicalSize;
 use self::walreceiver::{WalReceiver, WalReceiverConf};

 use super::{
-    config::TenantConf,
-    storage_layer::{inmemory_layer, LayerVisibilityHint},
+    config::TenantConf, storage_layer::inmemory_layer, storage_layer::LayerVisibilityHint,
    upload_queue::NotInitialized,
-    MaybeOffloaded,
 };
 use super::{debug_assert_current_span_has_tenant_and_timeline_id, AttachedTenantConf};
 use super::{remote_timeline_client::index::IndexPart, storage_layer::LayerFringe};
@@ -453,7 +450,7 @@ pub(crate) struct GcInfo {
    /// Currently, this includes all points where child branches have
    /// been forked off from. In the future, could also include
    /// explicit user-defined snapshot points.
-    pub(crate) retain_lsns: Vec<(Lsn, TimelineId, MaybeOffloaded)>,
+    pub(crate) retain_lsns: Vec<(Lsn, TimelineId)>,

    /// The cutoff coordinates, which are combined by selecting the minimum.
    pub(crate) cutoffs: GcCutoffs,
@@ -470,13 +467,8 @@ impl GcInfo {
        self.cutoffs.select_min()
    }

-    pub(super) fn insert_child(
-        &mut self,
-        child_id: TimelineId,
-        child_lsn: Lsn,
-        is_offloaded: MaybeOffloaded,
-    ) {
-        self.retain_lsns.push((child_lsn, child_id, is_offloaded));
+    pub(super) fn insert_child(&mut self, child_id: TimelineId, child_lsn: Lsn) {
+        self.retain_lsns.push((child_lsn, child_id));
        self.retain_lsns.sort_by_key(|i| i.0);
    }

@@ -2172,9 +2164,7 @@ impl Timeline {

        if let Some(ancestor) = &ancestor {
            let mut ancestor_gc_info = ancestor.gc_info.write().unwrap();
-            // If we construct an explicit timeline object, it's obviously not offloaded
-            let is_offloaded = MaybeOffloaded::No;
-            ancestor_gc_info.insert_child(timeline_id, metadata.ancestor_lsn(), is_offloaded);
+            ancestor_gc_info.insert_child(timeline_id, metadata.ancestor_lsn());
        }

        Arc::new_cyclic(|myself| {
@@ -4885,7 +4875,7 @@ impl Timeline {
            let retain_lsns = gc_info
                .retain_lsns
                .iter()
-                .map(|(lsn, _child_id, _is_offloaded)| *lsn)
+                .map(|(lsn, _child_id)| *lsn)
                .collect();

            // Gets the maximum LSN that holds the valid lease.
@@ -5855,15 +5845,17 @@ fn is_send() {
 #[cfg(test)]
 mod tests {
    use pageserver_api::key::Key;
-    use pageserver_api::value::Value;
    use utils::{id::TimelineId, lsn::Lsn};

-    use crate::tenant::{
-        harness::{test_img, TenantHarness},
-        layer_map::LayerMap,
-        storage_layer::{Layer, LayerName},
-        timeline::{DeltaLayerTestDesc, EvictionError},
-        Timeline,
+    use crate::{
+        repository::Value,
+        tenant::{
+            harness::{test_img, TenantHarness},
+            layer_map::LayerMap,
+            storage_layer::{Layer, LayerName},
+            timeline::{DeltaLayerTestDesc, EvictionError},
+            Timeline,
+        },
    };

    #[tokio::test]
--- a/pageserver/src/tenant/timeline/compaction.rs
+++ b/pageserver/src/tenant/timeline/compaction.rs
@@ -42,16 +42,15 @@ use crate::tenant::storage_layer::{
 use crate::tenant::timeline::ImageLayerCreationOutcome;
 use crate::tenant::timeline::{drop_rlock, DeltaLayerWriter, ImageLayerWriter};
 use crate::tenant::timeline::{Layer, ResidentLayer};
-use crate::tenant::{DeltaLayer, MaybeOffloaded};
+use crate::tenant::DeltaLayer;
 use crate::virtual_file::{MaybeFatalIo, VirtualFile};
 use pageserver_api::config::tenant_conf_defaults::{
    DEFAULT_CHECKPOINT_DISTANCE, DEFAULT_COMPACTION_THRESHOLD,
 };

-use pageserver_api::key::Key;
-use pageserver_api::keyspace::KeySpace;
-use pageserver_api::record::NeonWalRecord;
-use pageserver_api::value::Value;
+use crate::keyspace::KeySpace;
+use crate::repository::{Key, Value};
+use crate::walrecord::NeonWalRecord;

 use utils::lsn::Lsn;

@@ -640,10 +639,7 @@ impl Timeline {
            let children = self.gc_info.read().unwrap().retain_lsns.clone();

            let mut readable_points = Vec::with_capacity(children.len() + 1);
-            for (child_lsn, _child_timeline_id, is_offloaded) in &children {
-                if *is_offloaded == MaybeOffloaded::Yes {
-                    continue;
-                }
+            for (child_lsn, _child_timeline_id) in &children {
                readable_points.push(*child_lsn);
            }
            readable_points.push(head_lsn);
@@ -1745,7 +1741,7 @@ impl Timeline {
            let gc_info = self.gc_info.read().unwrap();
            let mut retain_lsns_below_horizon = Vec::new();
            let gc_cutoff = gc_info.cutoffs.select_min();
-            for (lsn, _timeline_id, _is_offloaded) in &gc_info.retain_lsns {
+            for (lsn, _timeline_id) in &gc_info.retain_lsns {
                if lsn < &gc_cutoff {
                    retain_lsns_below_horizon.push(*lsn);
                }
@@ -2129,7 +2125,7 @@ struct ResidentDeltaLayer(ResidentLayer);
 struct ResidentImageLayer(ResidentLayer);

 impl CompactionJobExecutor for TimelineAdaptor {
-    type Key = pageserver_api::key::Key;
+    type Key = crate::repository::Key;

    type Layer = OwnArc<PersistentLayerDesc>;
    type DeltaLayer = ResidentDeltaLayer;
--- a/pageserver/src/tenant/timeline/offload.rs
+++ b/pageserver/src/tenant/timeline/offload.rs
@@ -19,9 +19,6 @@ pub(crate) async fn offload_timeline(
        return Ok(());
    };

-    // Now that the Timeline is in Stopping state, request all the related tasks to shut down.
-    timeline.shutdown(super::ShutdownMode::Hard).await;
-
    // TODO extend guard mechanism above with method
    // to make deletions possible while offloading is in progress

--- a/pageserver/src/tenant/timeline/walreceiver/walreceiver_connection.rs
+++ b/pageserver/src/tenant/timeline/walreceiver/walreceiver_connection.rs
@@ -22,7 +22,6 @@ use tokio::{select, sync::watch, time};
 use tokio_postgres::{replication::ReplicationStream, Client};
 use tokio_util::sync::CancellationToken;
 use tracing::{debug, error, info, trace, warn, Instrument};
-use wal_decoder::models::{FlushUncommittedRecords, InterpretedWalRecord};

 use super::TaskStateUpdate;
 use crate::{
@@ -32,6 +31,7 @@ use crate::{
    task_mgr::{TaskKind, WALRECEIVER_RUNTIME},
    tenant::{debug_assert_current_span_has_tenant_and_timeline_id, Timeline, WalReceiverInfo},
    walingest::WalIngest,
+    walrecord::{decode_wal_record, DecodedWALRecord},
 };
 use postgres_backend::is_expected_io_error;
 use postgres_connection::PgConnectionConfig;
@@ -339,15 +339,11 @@ pub(super) async fn handle_walreceiver_connection(
                            return Err(WalReceiverError::Other(anyhow!("LSN not aligned")));
                        }

-                        // Deserialize and interpret WAL record
-                        let interpreted = InterpretedWalRecord::from_bytes(
-                            recdata,
-                            modification.tline.get_shard_identity(),
-                            lsn,
-                            modification.tline.pg_version,
-                        )?;
+                        // Deserialize WAL record
+                        let mut decoded = DecodedWALRecord::default();
+                        decode_wal_record(recdata, &mut decoded, modification.tline.pg_version)?;

-                        if matches!(interpreted.flush_uncommitted, FlushUncommittedRecords::Yes)
+                        if decoded.is_dbase_create_copy(timeline.pg_version)
                            && uncommitted_records > 0
                        {
                            // Special case: legacy PG database creations operate by reading pages from a 'template' database:
@@ -364,7 +360,7 @@ pub(super) async fn handle_walreceiver_connection(

                        // Ingest the records without immediately committing them.
                        let ingested = walingest
-                            .ingest_record(interpreted, &mut modification, &ctx)
+                            .ingest_record(decoded, lsn, &mut modification, &ctx)
                            .await
                            .with_context(|| format!("could not ingest record at {lsn}"))?;
                        if !ingested {
--- a/pageserver/src/walingest.rs
+++ b/pageserver/src/walingest.rs
--- a/libs/postgres_ffi/src/record.rs
+++ b/libs/postgres_ffi/src/record.rs
--- a/pageserver/src/walredo.rs
+++ b/pageserver/src/walredo.rs
@@ -29,11 +29,11 @@ use crate::metrics::{
    WAL_REDO_BYTES_HISTOGRAM, WAL_REDO_PROCESS_LAUNCH_DURATION_HISTOGRAM,
    WAL_REDO_RECORDS_HISTOGRAM, WAL_REDO_TIME,
 };
+use crate::repository::Key;
+use crate::walrecord::NeonWalRecord;
 use anyhow::Context;
 use bytes::{Bytes, BytesMut};
-use pageserver_api::key::Key;
 use pageserver_api::models::{WalRedoManagerProcessStatus, WalRedoManagerStatus};
-use pageserver_api::record::NeonWalRecord;
 use pageserver_api::shard::TenantShardId;
 use std::future::Future;
 use std::sync::Arc;
@@ -548,10 +548,9 @@ impl PostgresRedoManager {
 #[cfg(test)]
 mod tests {
    use super::PostgresRedoManager;
-    use crate::config::PageServerConf;
+    use crate::repository::Key;
+    use crate::{config::PageServerConf, walrecord::NeonWalRecord};
    use bytes::Bytes;
-    use pageserver_api::key::Key;
-    use pageserver_api::record::NeonWalRecord;
    use pageserver_api::shard::TenantShardId;
    use std::str::FromStr;
    use tracing::Instrument;
--- a/pageserver/src/walredo/apply_neon.rs
+++ b/pageserver/src/walredo/apply_neon.rs
@@ -1,9 +1,9 @@
 use crate::pgdatadir_mapping::AuxFilesDirectory;
+use crate::walrecord::NeonWalRecord;
 use anyhow::Context;
 use byteorder::{ByteOrder, LittleEndian};
 use bytes::{BufMut, BytesMut};
 use pageserver_api::key::Key;
-use pageserver_api::record::NeonWalRecord;
 use pageserver_api::reltag::SlruKind;
 use postgres_ffi::pg_constants;
 use postgres_ffi::relfile_utils::VISIBILITYMAP_FORKNUM;
@@ -244,7 +244,7 @@ pub(crate) fn apply_in_neon(
            let mut writer = page.writer();
            dir.ser_into(&mut writer)?;
        }
-        #[cfg(feature = "testing")]
+        #[cfg(test)]
        NeonWalRecord::Test {
            append,
            clear,
--- a/pageserver/src/walredo/process.rs
+++ b/pageserver/src/walredo/process.rs
@@ -8,10 +8,10 @@ use crate::{
    metrics::{WalRedoKillCause, WAL_REDO_PROCESS_COUNTERS, WAL_REDO_RECORD_COUNTER},
    page_cache::PAGE_SZ,
    span::debug_assert_current_span_has_tenant_id,
+    walrecord::NeonWalRecord,
 };
 use anyhow::Context;
 use bytes::Bytes;
-use pageserver_api::record::NeonWalRecord;
 use pageserver_api::{reltag::RelTag, shard::TenantShardId};
 use postgres_ffi::BLCKSZ;
 #[cfg(feature = "testing")]
--- a/pgxn/neon/file_cache.c
+++ b/pgxn/neon/file_cache.c
@@ -43,7 +43,6 @@
 #include "hll.h"
 #include "bitmap.h"
 #include "neon.h"
-#include "neon_perf_counters.h"

 #define CriticalAssert(cond) do if (!(cond)) elog(PANIC, "Assertion %s failed at %s:%d: ", #cond, __FILE__, __LINE__); while (0)

@@ -115,9 +114,7 @@ typedef struct FileCacheControl
 	uint32		limit;			/* shared copy of lfc_size_limit */
 	uint64		hits;
 	uint64		misses;
-	uint64		writes;			/* number of writes issued */
-	uint64		time_read;		/* time spent reading (us) */
-	uint64		time_write;		/* time spent writing (us) */
+	uint64		writes;
 	dlist_head	lru;			/* double linked list for LRU replacement
 								 * algorithm */
 	dlist_head  holes;          /* double linked list of punched holes */
@@ -273,8 +270,6 @@ lfc_shmem_startup(void)
 		lfc_ctl->hits = 0;
 		lfc_ctl->misses = 0;
 		lfc_ctl->writes = 0;
-		lfc_ctl->time_read = 0;
-		lfc_ctl->time_write = 0;
 		dlist_init(&lfc_ctl->lru);
 		dlist_init(&lfc_ctl->holes);

@@ -706,7 +701,6 @@ lfc_readv_select(NRelFileInfo rinfo, ForkNumber forkNum, BlockNumber blkno,
 		int		blocks_in_chunk = Min(nblocks, BLOCKS_PER_CHUNK - (blkno % BLOCKS_PER_CHUNK));
 		int		iteration_hits = 0;
 		int		iteration_misses = 0;
-		uint64	io_time_us = 0;
 		Assert(blocks_in_chunk > 0);

 		for (int i = 0; i < blocks_in_chunk; i++)
@@ -801,13 +795,6 @@ lfc_readv_select(NRelFileInfo rinfo, ForkNumber forkNum, BlockNumber blkno,
 			lfc_ctl->misses += iteration_misses;
 			pgBufferUsage.file_cache.hits += iteration_hits;
 			pgBufferUsage.file_cache.misses += iteration_misses;
-
-			if (iteration_hits)
-			{
-				lfc_ctl->time_read += io_time_us;
-				inc_page_cache_read_wait(io_time_us);
-			}
-
 			CriticalAssert(entry->access_count > 0);
 			if (--entry->access_count == 0)
 				dlist_push_tail(&lfc_ctl->lru, &entry->list_node);
@@ -872,7 +859,6 @@ lfc_writev(NRelFileInfo rinfo, ForkNumber forkNum, BlockNumber blkno,
 		struct iovec iov[PG_IOV_MAX];
 		int		chunk_offs = blkno & (BLOCKS_PER_CHUNK - 1);
 		int		blocks_in_chunk = Min(nblocks, BLOCKS_PER_CHUNK - (blkno % BLOCKS_PER_CHUNK));
-		instr_time io_start, io_end;
 		Assert(blocks_in_chunk > 0);

 		for (int i = 0; i < blocks_in_chunk; i++)
@@ -961,13 +947,12 @@ lfc_writev(NRelFileInfo rinfo, ForkNumber forkNum, BlockNumber blkno,

 		generation = lfc_ctl->generation;
 		entry_offset = entry->offset;
+		lfc_ctl->writes += blocks_in_chunk;
 		LWLockRelease(lfc_lock);

 		pgstat_report_wait_start(WAIT_EVENT_NEON_LFC_WRITE);
-		INSTR_TIME_SET_CURRENT(io_start);
 		rc = pwritev(lfc_desc, iov, blocks_in_chunk,
 					 ((off_t) entry_offset * BLOCKS_PER_CHUNK + chunk_offs) * BLCKSZ);
-		INSTR_TIME_SET_CURRENT(io_end);
 		pgstat_report_wait_end();

 		if (rc != BLCKSZ * blocks_in_chunk)
@@ -980,17 +965,9 @@ lfc_writev(NRelFileInfo rinfo, ForkNumber forkNum, BlockNumber blkno,

 			if (lfc_ctl->generation == generation)
 			{
-				uint64	time_spent_us;
 				CriticalAssert(LFC_ENABLED());
 				/* Place entry to the head of LRU list */
 				CriticalAssert(entry->access_count > 0);
-
-				lfc_ctl->writes += blocks_in_chunk;
-				INSTR_TIME_SUBTRACT(io_start, io_end);
-				time_spent_us = INSTR_TIME_GET_MICROSEC(io_start);
-				lfc_ctl->time_write += time_spent_us;
-				inc_page_cache_write_wait(time_spent_us);
-
 				if (--entry->access_count == 0)
 					dlist_push_tail(&lfc_ctl->lru, &entry->list_node);

--- a/pgxn/neon/neon_perf_counters.c
+++ b/pgxn/neon/neon_perf_counters.c
@@ -50,52 +50,28 @@ NeonPerfCountersShmemInit(void)
 	}
 }

-static inline void
-inc_iohist(IOHistogram hist, uint64 latency_us)
+/*
+ * Count a GetPage wait operation.
+ */
+void
+inc_getpage_wait(uint64 latency_us)
 {
 	int			lo = 0;
-	int			hi = NUM_IO_WAIT_BUCKETS - 1;
+	int			hi = NUM_GETPAGE_WAIT_BUCKETS - 1;

 	/* Find the right bucket with binary search */
 	while (lo < hi)
 	{
 		int			mid = (lo + hi) / 2;

-		if (latency_us < io_wait_bucket_thresholds[mid])
+		if (latency_us < getpage_wait_bucket_thresholds[mid])
 			hi = mid;
 		else
 			lo = mid + 1;
 	}
-	hist->wait_us_bucket[lo]++;
-	hist->wait_us_sum += latency_us;
-	hist->wait_us_count++;
-}
-
-/*
- * Count a GetPage wait operation.
- */
-void
-inc_getpage_wait(uint64 latency)
-{
-	inc_iohist(&MyNeonCounters->getpage_hist, latency);
-}
-
-/*
- * Count an LFC read wait operation.
- */
-void
-inc_page_cache_read_wait(uint64 latency)
-{
-	inc_iohist(&MyNeonCounters->file_cache_read_hist, latency);
-}
-
-/*
- * Count an LFC write wait operation.
- */
-void
-inc_page_cache_write_wait(uint64 latency)
-{
-	inc_iohist(&MyNeonCounters->file_cache_write_hist, latency);
+	MyNeonCounters->getpage_wait_us_bucket[lo]++;
+	MyNeonCounters->getpage_wait_us_sum += latency_us;
+	MyNeonCounters->getpage_wait_us_count++;
 }

 /*
@@ -105,91 +81,77 @@ inc_page_cache_write_wait(uint64 latency)

 typedef struct
 {
-	const char *name;
+	char	   *name;
 	bool		is_bucket;
 	double		bucket_le;
 	double		value;
 } metric_t;

-static int
-histogram_to_metrics(IOHistogram histogram,
-					 metric_t *metrics,
-					 const char *count,
-					 const char *sum,
-					 const char *bucket)
+static metric_t *
+neon_perf_counters_to_metrics(neon_per_backend_counters *counters)
 {
-	int		i = 0;
-	uint64	bucket_accum = 0;
+#define NUM_METRICS (2 + NUM_GETPAGE_WAIT_BUCKETS + 8)
+	metric_t   *metrics = palloc((NUM_METRICS + 1) * sizeof(metric_t));
+	uint64		bucket_accum;
+	int			i = 0;

-	metrics[i].name = count;
+	metrics[i].name = "getpage_wait_seconds_count";
 	metrics[i].is_bucket = false;
-	metrics[i].value = (double) histogram->wait_us_count;
+	metrics[i].value = (double) counters->getpage_wait_us_count;
 	i++;
-	metrics[i].name = sum;
+	metrics[i].name = "getpage_wait_seconds_sum";
 	metrics[i].is_bucket = false;
-	metrics[i].value = (double) histogram->wait_us_sum / 1000000.0;
+	metrics[i].value = ((double) counters->getpage_wait_us_sum) / 1000000.0;
 	i++;
-	for (int bucketno = 0; bucketno < NUM_IO_WAIT_BUCKETS; bucketno++)
+
+	bucket_accum = 0;
+	for (int bucketno = 0; bucketno < NUM_GETPAGE_WAIT_BUCKETS; bucketno++)
 	{
-		uint64		threshold = io_wait_bucket_thresholds[bucketno];
+		uint64		threshold = getpage_wait_bucket_thresholds[bucketno];

-		bucket_accum += histogram->wait_us_bucket[bucketno];
+		bucket_accum += counters->getpage_wait_us_bucket[bucketno];

-		metrics[i].name = bucket;
+		metrics[i].name = "getpage_wait_seconds_bucket";
 		metrics[i].is_bucket = true;
 		metrics[i].bucket_le = (threshold == UINT64_MAX) ? INFINITY : ((double) threshold) / 1000000.0;
 		metrics[i].value = (double) bucket_accum;
 		i++;
 	}
-
-	return i;
-}
-
-static metric_t *
-neon_perf_counters_to_metrics(neon_per_backend_counters *counters)
-{
-#define NUM_METRICS ((2 + NUM_IO_WAIT_BUCKETS) * 3 + 10)
-	metric_t   *metrics = palloc((NUM_METRICS + 1) * sizeof(metric_t));
-	int			i = 0;
-
-#define APPEND_METRIC(_name) do { \
-		metrics[i].name = #_name; \
-		metrics[i].is_bucket = false; \
-		metrics[i].value = (double) counters->_name; \
-		i++; \
-	} while (false)
-
-	i += histogram_to_metrics(&counters->getpage_hist, &metrics[i],
-							  "getpage_wait_seconds_count",
-							  "getpage_wait_seconds_sum",
-							  "getpage_wait_seconds_bucket");
-
-	APPEND_METRIC(getpage_prefetch_requests_total);
-	APPEND_METRIC(getpage_sync_requests_total);
-	APPEND_METRIC(getpage_prefetch_misses_total);
-	APPEND_METRIC(getpage_prefetch_discards_total);
-	APPEND_METRIC(pageserver_requests_sent_total);
-	APPEND_METRIC(pageserver_disconnects_total);
-	APPEND_METRIC(pageserver_send_flushes_total);
-	APPEND_METRIC(pageserver_open_requests);
-	APPEND_METRIC(getpage_prefetches_buffered);
-
-	APPEND_METRIC(file_cache_hits_total);
-
-	i += histogram_to_metrics(&counters->file_cache_read_hist, &metrics[i],
-							  "file_cache_read_wait_seconds_count",
-							  "file_cache_read_wait_seconds_sum",
-							  "file_cache_read_wait_seconds_bucket");
-	i += histogram_to_metrics(&counters->file_cache_write_hist, &metrics[i],
-							  "file_cache_write_wait_seconds_count",
-							  "file_cache_write_wait_seconds_sum",
-							  "file_cache_write_wait_seconds_bucket");
+	metrics[i].name = "getpage_prefetch_requests_total";
+	metrics[i].is_bucket = false;
+	metrics[i].value = (double) counters->getpage_prefetch_requests_total;
+	i++;
+	metrics[i].name = "getpage_sync_requests_total";
+	metrics[i].is_bucket = false;
+	metrics[i].value = (double) counters->getpage_sync_requests_total;
+	i++;
+	metrics[i].name = "getpage_prefetch_misses_total";
+	metrics[i].is_bucket = false;
+	metrics[i].value = (double) counters->getpage_prefetch_misses_total;
+	i++;
+	metrics[i].name = "getpage_prefetch_discards_total";
+	metrics[i].is_bucket = false;
+	metrics[i].value = (double) counters->getpage_prefetch_discards_total;
+	i++;
+	metrics[i].name = "pageserver_requests_sent_total";
+	metrics[i].is_bucket = false;
+	metrics[i].value = (double) counters->pageserver_requests_sent_total;
+	i++;
+	metrics[i].name = "pageserver_disconnects_total";
+	metrics[i].is_bucket = false;
+	metrics[i].value = (double) counters->pageserver_disconnects_total;
+	i++;
+	metrics[i].name = "pageserver_send_flushes_total";
+	metrics[i].is_bucket = false;
+	metrics[i].value = (double) counters->pageserver_send_flushes_total;
+	i++;
+	metrics[i].name = "file_cache_hits_total";
+	metrics[i].is_bucket = false;
+	metrics[i].value = (double) counters->file_cache_hits_total;
+	i++;

 	Assert(i == NUM_METRICS);

-#undef APPEND_METRIC
-#undef NUM_METRICS
-
 	/* NULL entry marks end of array */
 	metrics[i].name = NULL;
 	metrics[i].value = 0;
@@ -254,15 +216,6 @@ neon_get_backend_perf_counters(PG_FUNCTION_ARGS)
 	return (Datum) 0;
 }

-static inline void
-histogram_merge_into(IOHistogram into, IOHistogram from)
-{
-	into->wait_us_count += from->wait_us_count;
-	into->wait_us_sum += from->wait_us_sum;
-	for (int bucketno = 0; bucketno < NUM_IO_WAIT_BUCKETS; bucketno++)
-		into->wait_us_bucket[bucketno] += from->wait_us_bucket[bucketno];
-}
-
 PG_FUNCTION_INFO_V1(neon_get_perf_counters);
 Datum
 neon_get_perf_counters(PG_FUNCTION_ARGS)
@@ -281,7 +234,10 @@ neon_get_perf_counters(PG_FUNCTION_ARGS)
 	{
 		neon_per_backend_counters *counters = &neon_per_backend_counters_shared[procno];

-		histogram_merge_into(&totals.getpage_hist, &counters->getpage_hist);
+		totals.getpage_wait_us_count += counters->getpage_wait_us_count;
+		totals.getpage_wait_us_sum += counters->getpage_wait_us_sum;
+		for (int bucketno = 0; bucketno < NUM_GETPAGE_WAIT_BUCKETS; bucketno++)
+			totals.getpage_wait_us_bucket[bucketno] += counters->getpage_wait_us_bucket[bucketno];
 		totals.getpage_prefetch_requests_total += counters->getpage_prefetch_requests_total;
 		totals.getpage_sync_requests_total += counters->getpage_sync_requests_total;
 		totals.getpage_prefetch_misses_total += counters->getpage_prefetch_misses_total;
@@ -289,11 +245,7 @@ neon_get_perf_counters(PG_FUNCTION_ARGS)
 		totals.pageserver_requests_sent_total += counters->pageserver_requests_sent_total;
 		totals.pageserver_disconnects_total += counters->pageserver_disconnects_total;
 		totals.pageserver_send_flushes_total += counters->pageserver_send_flushes_total;
-		totals.pageserver_open_requests += counters->pageserver_open_requests;
-		totals.getpage_prefetches_buffered += counters->getpage_prefetches_buffered;
 		totals.file_cache_hits_total += counters->file_cache_hits_total;
-		histogram_merge_into(&totals.file_cache_read_hist, &counters->file_cache_read_hist);
-		histogram_merge_into(&totals.file_cache_write_hist, &counters->file_cache_write_hist);
 	}

 	metrics = neon_perf_counters_to_metrics(&totals);
--- a/pgxn/neon/neon_perf_counters.h
+++ b/pgxn/neon/neon_perf_counters.h
@@ -15,26 +15,17 @@
 #include "storage/proc.h"
 #endif

-static const uint64 io_wait_bucket_thresholds[] = {
-	       2,        3,        6,        10,  /* 0 us   - 10 us */
-	      20,       30,       60,       100,  /* 10 us  - 100 us */
+static const uint64 getpage_wait_bucket_thresholds[] = {
+	      20,       30,       60,       100,  /* 0      -  100 us */
 	     200,      300,      600,	   1000,  /* 100 us - 1 ms */
 	    2000,     3000,     6000,     10000,  /* 1 ms   - 10 ms */
 	   20000,    30000,    60000,    100000,  /* 10 ms  - 100 ms */
 	  200000,   300000,   600000,   1000000,  /* 100 ms - 1 s */
 	 2000000,  3000000,  6000000,  10000000,  /* 1 s - 10 s */
+    20000000, 30000000, 60000000, 100000000,  /* 10 s - 100 s */
 	UINT64_MAX,
 };
-#define NUM_IO_WAIT_BUCKETS (lengthof(io_wait_bucket_thresholds))
-
-typedef struct IOHistogramData
-{
-	uint64		wait_us_count;
-	uint64		wait_us_sum;
-	uint64		wait_us_bucket[NUM_IO_WAIT_BUCKETS];
-} IOHistogramData;
-
-typedef IOHistogramData *IOHistogram;
+#define NUM_GETPAGE_WAIT_BUCKETS (lengthof(getpage_wait_bucket_thresholds))

 typedef struct
 {
@@ -48,7 +39,9 @@ typedef struct
 	 * the backend, but the 'neon_backend_perf_counters' view will convert
 	 * them to seconds, to make them more idiomatic as prometheus metrics.
 	 */
-	IOHistogramData getpage_hist;
+	uint64		getpage_wait_us_count;
+	uint64		getpage_wait_us_sum;
+	uint64		getpage_wait_us_bucket[NUM_GETPAGE_WAIT_BUCKETS];

 	/*
 	 * Total number of speculative prefetch Getpage requests and synchronous
@@ -57,11 +50,7 @@ typedef struct
 	uint64		getpage_prefetch_requests_total;
 	uint64		getpage_sync_requests_total;

-	/*
-	 * Total number of readahead misses; consisting of either prefetches that
-	 * don't satisfy the LSN bounds, or cases where no readahead was issued
-	 * for the read.
-	 */
+	/* XXX: It's not clear to me when these misses happen. */
 	uint64		getpage_prefetch_misses_total;

 	/*
@@ -91,16 +80,6 @@ typedef struct
 	 * this can be smaller than pageserver_requests_sent_total.
 	 */
 	uint64		pageserver_send_flushes_total;
-	
-	/*
-	 * Number of open requests to PageServer.
-	 */
-	uint64		pageserver_open_requests;
-
-	/*
-	 * Number of unused prefetches currently cached in this backend.
-	 */
-	uint64		getpage_prefetches_buffered;

 	/*
 	 * Number of requests satisfied from the LFC.
@@ -112,9 +91,6 @@ typedef struct
 	 */
 	uint64		file_cache_hits_total;

-	/* LFC I/O time buckets */
-	IOHistogramData file_cache_read_hist;
-	IOHistogramData file_cache_write_hist;
 } neon_per_backend_counters;

 /* Pointer to the shared memory array of neon_per_backend_counters structs */
@@ -135,8 +111,6 @@ extern neon_per_backend_counters *neon_per_backend_counters_shared;
 #endif

 extern void inc_getpage_wait(uint64 latency);
-extern void inc_page_cache_read_wait(uint64 latency);
-extern void inc_page_cache_write_wait(uint64 latency);

 extern Size NeonPerfCountersShmemSize(void);
 extern void NeonPerfCountersShmemInit(void);
--- a/pgxn/neon/pagestore_smgr.c
+++ b/pgxn/neon/pagestore_smgr.c
@@ -488,11 +488,6 @@ readahead_buffer_resize(int newsize, void *extra)
 		newPState->n_unused -= 1;
 	}

-	MyNeonCounters->getpage_prefetches_buffered =
-		MyPState->n_responses_buffered;
-	MyNeonCounters->pageserver_open_requests =
-		MyPState->n_requests_inflight;
-
 	for (; end >= MyPState->ring_last && end != UINT64_MAX; end -= 1)
 	{
 		prefetch_set_unused(end);
@@ -626,8 +621,6 @@ prefetch_read(PrefetchRequest *slot)
 		MyPState->n_responses_buffered += 1;
 		MyPState->n_requests_inflight -= 1;
 		MyPState->ring_receive += 1;
-		MyNeonCounters->getpage_prefetches_buffered =
-			MyPState->n_responses_buffered;

 		/* update slot state */
 		slot->status = PRFS_RECEIVED;
@@ -681,15 +674,6 @@ prefetch_on_ps_disconnect(void)

 		prefetch_set_unused(ring_index);
 	}
-
-	/*
-	 * We can have gone into retry due to network error, so update stats with
-	 * the latest available 
-	 */
-	MyNeonCounters->pageserver_open_requests =
-		MyPState->n_requests_inflight;
-	MyNeonCounters->getpage_prefetches_buffered =
-		MyPState->n_responses_buffered;
 }

 /*
@@ -722,9 +706,6 @@ prefetch_set_unused(uint64 ring_index)

 		MyPState->n_responses_buffered -= 1;
 		MyPState->n_unused += 1;
-
-		MyNeonCounters->getpage_prefetches_buffered =
-			MyPState->n_responses_buffered;
 	}
 	else
 	{
@@ -839,15 +820,6 @@ prefetch_register_bufferv(BufferTag tag, neon_request_lsns *frlsns,
 	hashkey.buftag = tag;

 Retry:
-	/*
-	 * We can have gone into retry due to network error, so update stats with
-	 * the latest available 
-	 */
-	MyNeonCounters->pageserver_open_requests =
-		MyPState->ring_unused - MyPState->ring_receive;
-	MyNeonCounters->getpage_prefetches_buffered =
-		MyPState->n_responses_buffered;
-
 	min_ring_index = UINT64_MAX;
 	for (int i = 0; i < nblocks; i++)
 	{
@@ -1029,9 +1001,6 @@ Retry:
 		prefetch_do_request(slot, lsns);
 	}

-	MyNeonCounters->pageserver_open_requests =
-		MyPState->ring_unused - MyPState->ring_receive;
-
 	Assert(any_hits);

 	Assert(GetPrfSlot(min_ring_index)->status == PRFS_REQUESTED ||
@@ -1107,10 +1076,8 @@ page_server_request(void const *req)
 			{
 				/* do nothing */
 			}
-			MyNeonCounters->pageserver_open_requests++;
 			consume_prefetch_responses();
 			resp = page_server->receive(shard_no);
-			MyNeonCounters->pageserver_open_requests--;
 		}
 		PG_CATCH();
 		{
@@ -1119,8 +1086,6 @@ page_server_request(void const *req)
 			 * point, but this currently seems fine for now.
 			 */
 			page_server->disconnect(shard_no);
-			MyNeonCounters->pageserver_open_requests = 0;
-
 			PG_RE_THROW();
 		}
 		PG_END_TRY();
--- a/proxy/Cargo.toml
+++ b/proxy/Cargo.toml
@@ -42,10 +42,9 @@ hyper0.workspace = true
 hyper = { workspace = true, features = ["server", "http1", "http2"] }
 hyper-util = { version = "0.1", features = ["server", "http1", "http2", "tokio"] }
 http-body-util = { version = "0.1" }
-indexmap = { workspace = true, features = ["serde"] }
+indexmap.workspace = true
 ipnet.workspace = true
 itertools.workspace = true
-itoa.workspace = true
 lasso = { workspace = true, features = ["multi-threaded"] }
 measured = { workspace = true, features = ["lasso"] }
 metrics.workspace = true
@@ -78,7 +77,7 @@ subtle.workspace = true
 thiserror.workspace = true
 tikv-jemallocator.workspace = true
 tikv-jemalloc-ctl = { workspace = true, features = ["use_std"] }
-tokio-postgres = { workspace = true, features = ["with-serde_json-1"] }
+tokio-postgres.workspace = true
 tokio-postgres-rustls.workspace = true
 tokio-rustls.workspace = true
 tokio-util.workspace = true
@@ -102,7 +101,7 @@ jose-jwa = "0.1.2"
 jose-jwk = { version = "0.1.2", features = ["p256", "p384", "rsa"] }
 signature = "2"
 ecdsa = "0.16"
-p256 = { version = "0.13", features = ["jwk"] }
+p256 = "0.13"
 rsa = "0.9"

 workspace_hack.workspace = true
--- a/proxy/src/auth/backend/console_redirect.rs
+++ b/proxy/src/auth/backend/console_redirect.rs
@@ -1,24 +1,18 @@
 use crate::{
-    auth,
-    cache::Cached,
-    compute,
+    auth, compute,
    config::AuthenticationConfig,
    context::RequestMonitoring,
-    control_plane::{self, provider::NodeInfo, CachedNodeInfo},
+    control_plane::{self, provider::NodeInfo},
    error::{ReportableError, UserFacingError},
-    proxy::connect_compute::ComputeConnectBackend,
    stream::PqStream,
    waiters,
 };
-use async_trait::async_trait;
 use pq_proto::BeMessage as Be;
 use thiserror::Error;
 use tokio::io::{AsyncRead, AsyncWrite};
 use tokio_postgres::config::SslMode;
 use tracing::{info, info_span};

-use super::ComputeCredentialKeys;
-
 #[derive(Debug, Error)]
 pub(crate) enum WebAuthError {
    #[error(transparent)]
@@ -31,11 +25,6 @@ pub(crate) enum WebAuthError {
    Io(#[from] std::io::Error),
 }

-#[derive(Debug)]
-pub struct ConsoleRedirectBackend {
-    console_uri: reqwest::Url,
-}
-
 impl UserFacingError for WebAuthError {
    fn to_string_client(&self) -> String {
        "Internal error".to_string()
@@ -68,40 +57,7 @@ pub(crate) fn new_psql_session_id() -> String {
    hex::encode(rand::random::<[u8; 8]>())
 }

-impl ConsoleRedirectBackend {
-    pub fn new(console_uri: reqwest::Url) -> Self {
-        Self { console_uri }
-    }
-
-    pub(crate) async fn authenticate(
-        &self,
-        ctx: &RequestMonitoring,
-        auth_config: &'static AuthenticationConfig,
-        client: &mut PqStream<impl AsyncRead + AsyncWrite + Unpin>,
-    ) -> auth::Result<ConsoleRedirectNodeInfo> {
-        authenticate(ctx, auth_config, &self.console_uri, client)
-            .await
-            .map(ConsoleRedirectNodeInfo)
-    }
-}
-
-pub struct ConsoleRedirectNodeInfo(pub(super) NodeInfo);
-
-#[async_trait]
-impl ComputeConnectBackend for ConsoleRedirectNodeInfo {
-    async fn wake_compute(
-        &self,
-        _ctx: &RequestMonitoring,
-    ) -> Result<CachedNodeInfo, control_plane::errors::WakeComputeError> {
-        Ok(Cached::new_uncached(self.0.clone()))
-    }
-
-    fn get_keys(&self) -> &ComputeCredentialKeys {
-        &ComputeCredentialKeys::None
-    }
-}
-
-async fn authenticate(
+pub(super) async fn authenticate(
    ctx: &RequestMonitoring,
    auth_config: &'static AuthenticationConfig,
    link_uri: &reqwest::Url,
--- a/proxy/src/auth/backend/jwt.rs
+++ b/proxy/src/auth/backend/jwt.rs
@@ -4,18 +4,17 @@ use std::{
    time::{Duration, SystemTime},
 };

+use anyhow::{bail, ensure, Context};
 use arc_swap::ArcSwapOption;
 use dashmap::DashMap;
 use jose_jwk::crypto::KeyInfo;
 use serde::{de::Visitor, Deserialize, Deserializer};
 use signature::Verifier;
-use thiserror::Error;
 use tokio::time::Instant;

 use crate::{
-    auth::backend::ComputeCredentialKeys, context::RequestMonitoring,
-    control_plane::errors::GetEndpointJwksError, http::parse_json_body_with_limit,
-    intern::RoleNameInt, EndpointId, RoleName,
+    context::RequestMonitoring, http::parse_json_body_with_limit, intern::RoleNameInt, EndpointId,
+    RoleName,
 };

 // TODO(conrad): make these configurable.
@@ -31,16 +30,7 @@ pub(crate) trait FetchAuthRules: Clone + Send + Sync + 'static {
        &self,
        ctx: &RequestMonitoring,
        endpoint: EndpointId,
-    ) -> impl Future<Output = Result<Vec<AuthRule>, FetchAuthRulesError>> + Send;
-}
-
-#[derive(Error, Debug)]
-pub(crate) enum FetchAuthRulesError {
-    #[error(transparent)]
-    GetEndpointJwks(#[from] GetEndpointJwksError),
-
-    #[error("JWKs settings for this role were not configured")]
-    RoleJwksNotConfigured,
+    ) -> impl Future<Output = anyhow::Result<Vec<AuthRule>>> + Send;
 }

 pub(crate) struct AuthRule {
@@ -130,7 +120,7 @@ impl JwkCacheEntryLock {
        client: &reqwest::Client,
        endpoint: EndpointId,
        auth_rules: &F,
-    ) -> Result<Arc<JwkCacheEntry>, JwtError> {
+    ) -> anyhow::Result<Arc<JwkCacheEntry>> {
        // double check that no one beat us to updating the cache.
        let now = Instant::now();
        let guard = self.cached.load_full();
@@ -196,7 +186,7 @@ impl JwkCacheEntryLock {
        client: &reqwest::Client,
        endpoint: EndpointId,
        fetch: &F,
-    ) -> Result<Arc<JwkCacheEntry>, JwtError> {
+    ) -> Result<Arc<JwkCacheEntry>, anyhow::Error> {
        let now = Instant::now();
        let guard = self.cached.load_full();

@@ -251,24 +241,27 @@ impl JwkCacheEntryLock {
        endpoint: EndpointId,
        role_name: &RoleName,
        fetch: &F,
-    ) -> Result<ComputeCredentialKeys, JwtError> {
+    ) -> Result<(), anyhow::Error> {
        // JWT compact form is defined to be
        // <B64(Header)> || . || <B64(Payload)> || . || <B64(Signature)>
        // where Signature = alg(<B64(Header)> || . || <B64(Payload)>);

        let (header_payload, signature) = jwt
            .rsplit_once('.')
-            .ok_or(JwtEncodingError::InvalidCompactForm)?;
+            .context("Provided authentication token is not a valid JWT encoding")?;
        let (header, payload) = header_payload
            .split_once('.')
-            .ok_or(JwtEncodingError::InvalidCompactForm)?;
+            .context("Provided authentication token is not a valid JWT encoding")?;

-        let header = base64::decode_config(header, base64::URL_SAFE_NO_PAD)?;
-        let header = serde_json::from_slice::<JwtHeader<'_>>(&header)?;
+        let header = base64::decode_config(header, base64::URL_SAFE_NO_PAD)
+            .context("Provided authentication token is not a valid JWT encoding")?;
+        let header = serde_json::from_slice::<JwtHeader<'_>>(&header)
+            .context("Provided authentication token is not a valid JWT encoding")?;

-        let sig = base64::decode_config(signature, base64::URL_SAFE_NO_PAD)?;
+        let sig = base64::decode_config(signature, base64::URL_SAFE_NO_PAD)
+            .context("Provided authentication token is not a valid JWT encoding")?;

-        let kid = header.key_id.ok_or(JwtError::MissingKeyId)?;
+        let kid = header.key_id.context("missing key id")?;

        let mut guard = self
            .get_or_update_jwk_cache(ctx, client, endpoint.clone(), fetch)
@@ -286,13 +279,16 @@ impl JwkCacheEntryLock {
                        .renew_jwks(permit, ctx, client, endpoint.clone(), fetch)
                        .await?;
                }
-                _ => return Err(JwtError::JwkNotFound),
+                _ => {
+                    bail!("jwk not found");
+                }
            }
        };

-        if !jwk.is_supported(&header.algorithm) {
-            return Err(JwtError::SignatureAlgorithmNotSupported);
-        }
+        ensure!(
+            jwk.is_supported(&header.algorithm),
+            "signature algorithm not supported"
+        );

        match &jwk.key {
            jose_jwk::Key::Ec(key) => {
@@ -301,35 +297,37 @@ impl JwkCacheEntryLock {
            jose_jwk::Key::Rsa(key) => {
                verify_rsa_signature(header_payload.as_bytes(), &sig, key, &header.algorithm)?;
            }
-            key => return Err(JwtError::UnsupportedKeyType(key.into())),
+            key => bail!("unsupported key type {key:?}"),
        };

-        let payloadb = base64::decode_config(payload, base64::URL_SAFE_NO_PAD)?;
-        let payload = serde_json::from_slice::<JwtPayload<'_>>(&payloadb)?;
+        let payload = base64::decode_config(payload, base64::URL_SAFE_NO_PAD)
+            .context("Provided authentication token is not a valid JWT encoding")?;
+        let payload = serde_json::from_slice::<JwtPayload<'_>>(&payload)
+            .context("Provided authentication token is not a valid JWT encoding")?;

        tracing::debug!(?payload, "JWT signature valid with claims");

        if let Some(aud) = expected_audience {
-            if payload.audience.0.iter().all(|s| s != aud) {
-                return Err(JwtError::InvalidJwtTokenAudience);
-            }
+            ensure!(
+                payload.audience.0.iter().any(|s| s == aud),
+                "invalid JWT token audience"
+            );
        }

        let now = SystemTime::now();

        if let Some(exp) = payload.expiration {
-            if now >= exp + CLOCK_SKEW_LEEWAY {
-                return Err(JwtError::JwtTokenHasExpired);
-            }
+            ensure!(now < exp + CLOCK_SKEW_LEEWAY, "JWT token has expired");
        }

        if let Some(nbf) = payload.not_before {
-            if nbf >= now + CLOCK_SKEW_LEEWAY {
-                return Err(JwtError::JwtTokenNotYetReadyToUse);
-            }
+            ensure!(
+                nbf < now + CLOCK_SKEW_LEEWAY,
+                "JWT token is not yet ready to use"
+            );
        }

-        Ok(ComputeCredentialKeys::JwtPayload(payloadb))
+        Ok(())
    }
 }

@@ -341,7 +339,7 @@ impl JwkCache {
        role_name: &RoleName,
        fetch: &F,
        jwt: &str,
-    ) -> Result<ComputeCredentialKeys, JwtError> {
+    ) -> Result<(), anyhow::Error> {
        // try with just a read lock first
        let key = (endpoint.clone(), role_name.clone());
        let entry = self.map.get(&key).as_deref().map(Arc::clone);
@@ -357,18 +355,19 @@ impl JwkCache {
    }
 }

-fn verify_ec_signature(data: &[u8], sig: &[u8], key: &jose_jwk::Ec) -> Result<(), JwtError> {
+fn verify_ec_signature(data: &[u8], sig: &[u8], key: &jose_jwk::Ec) -> anyhow::Result<()> {
    use ecdsa::Signature;
    use signature::Verifier;

    match key.crv {
        jose_jwk::EcCurves::P256 => {
-            let pk = p256::PublicKey::try_from(key).map_err(JwtError::InvalidP256Key)?;
+            let pk =
+                p256::PublicKey::try_from(key).map_err(|_| anyhow::anyhow!("invalid P256 key"))?;
            let key = p256::ecdsa::VerifyingKey::from(&pk);
            let sig = Signature::from_slice(sig)?;
            key.verify(data, &sig)?;
        }
-        key => return Err(JwtError::UnsupportedEcKeyType(key)),
+        key => bail!("unsupported ec key type {key:?}"),
    }

    Ok(())
@@ -379,14 +378,14 @@ fn verify_rsa_signature(
    sig: &[u8],
    key: &jose_jwk::Rsa,
    alg: &jose_jwa::Algorithm,
-) -> Result<(), JwtError> {
+) -> anyhow::Result<()> {
    use jose_jwa::{Algorithm, Signing};
    use rsa::{
        pkcs1v15::{Signature, VerifyingKey},
        RsaPublicKey,
    };

-    let key = RsaPublicKey::try_from(key).map_err(JwtError::InvalidRsaKey)?;
+    let key = RsaPublicKey::try_from(key).map_err(|_| anyhow::anyhow!("invalid RSA key"))?;

    match alg {
        Algorithm::Signing(Signing::Rs256) => {
@@ -394,7 +393,7 @@ fn verify_rsa_signature(
            let sig = Signature::try_from(sig)?;
            key.verify(data, &sig)?;
        }
-        _ => return Err(JwtError::InvalidRsaSigningAlgorithm),
+        _ => bail!("invalid RSA signing algorithm"),
    };

    Ok(())
@@ -560,99 +559,6 @@ impl Drop for JwkRenewalPermit<'_> {
    }
 }

-#[derive(Error, Debug)]
-#[non_exhaustive]
-pub(crate) enum JwtError {
-    #[error("jwk not found")]
-    JwkNotFound,
-
-    #[error("missing key id")]
-    MissingKeyId,
-
-    #[error("Provided authentication token is not a valid JWT encoding")]
-    JwtEncoding(#[from] JwtEncodingError),
-
-    #[error("invalid JWT token audience")]
-    InvalidJwtTokenAudience,
-
-    #[error("JWT token has expired")]
-    JwtTokenHasExpired,
-
-    #[error("JWT token is not yet ready to use")]
-    JwtTokenNotYetReadyToUse,
-
-    #[error("invalid P256 key")]
-    InvalidP256Key(jose_jwk::crypto::Error),
-
-    #[error("invalid RSA key")]
-    InvalidRsaKey(jose_jwk::crypto::Error),
-
-    #[error("invalid RSA signing algorithm")]
-    InvalidRsaSigningAlgorithm,
-
-    #[error("unsupported EC key type {0:?}")]
-    UnsupportedEcKeyType(jose_jwk::EcCurves),
-
-    #[error("unsupported key type {0:?}")]
-    UnsupportedKeyType(KeyType),
-
-    #[error("signature algorithm not supported")]
-    SignatureAlgorithmNotSupported,
-
-    #[error("signature error: {0}")]
-    Signature(#[from] signature::Error),
-
-    #[error("failed to fetch auth rules: {0}")]
-    FetchAuthRules(#[from] FetchAuthRulesError),
-}
-
-impl From<base64::DecodeError> for JwtError {
-    fn from(err: base64::DecodeError) -> Self {
-        JwtEncodingError::Base64Decode(err).into()
-    }
-}
-
-impl From<serde_json::Error> for JwtError {
-    fn from(err: serde_json::Error) -> Self {
-        JwtEncodingError::SerdeJson(err).into()
-    }
-}
-
-#[derive(Error, Debug)]
-#[non_exhaustive]
-pub enum JwtEncodingError {
-    #[error(transparent)]
-    Base64Decode(#[from] base64::DecodeError),
-
-    #[error(transparent)]
-    SerdeJson(#[from] serde_json::Error),
-
-    #[error("invalid compact form")]
-    InvalidCompactForm,
-}
-
-#[allow(dead_code, reason = "Debug use only")]
-#[derive(Debug)]
-pub(crate) enum KeyType {
-    Ec(jose_jwk::EcCurves),
-    Rsa,
-    Oct,
-    Okp(jose_jwk::OkpCurves),
-    Unknown,
-}
-
-impl From<&jose_jwk::Key> for KeyType {
-    fn from(key: &jose_jwk::Key) -> Self {
-        match key {
-            jose_jwk::Key::Ec(ec) => Self::Ec(ec.crv),
-            jose_jwk::Key::Rsa(_rsa) => Self::Rsa,
-            jose_jwk::Key::Oct(_oct) => Self::Oct,
-            jose_jwk::Key::Okp(okp) => Self::Okp(okp.crv),
-            _ => Self::Unknown,
-        }
-    }
-}
-
 #[cfg(test)]
 mod tests {
    use crate::RoleName;
@@ -850,7 +756,7 @@ X0n5X2/pBLJzxZc62ccvZYVnctBiFs6HbSnxpuMQCfkt/BcR/ttIepBQQIW86wHL
                &self,
                _ctx: &RequestMonitoring,
                _endpoint: EndpointId,
-            ) -> Result<Vec<AuthRule>, FetchAuthRulesError> {
+            ) -> anyhow::Result<Vec<AuthRule>> {
                Ok(vec![
                    AuthRule {
                        id: "foo".to_owned(),
--- a/proxy/src/auth/backend/local.rs
+++ b/proxy/src/auth/backend/local.rs
@@ -1,9 +1,9 @@
 use std::net::SocketAddr;

+use anyhow::Context;
 use arc_swap::ArcSwapOption;

 use crate::{
-    auth::backend::jwt::FetchAuthRulesError,
    compute::ConnCfg,
    context::RequestMonitoring,
    control_plane::{
@@ -53,11 +53,11 @@ impl FetchAuthRules for StaticAuthRules {
        &self,
        _ctx: &RequestMonitoring,
        _endpoint: EndpointId,
-    ) -> Result<Vec<AuthRule>, FetchAuthRulesError> {
+    ) -> anyhow::Result<Vec<AuthRule>> {
        let mappings = JWKS_ROLE_MAP.load();
        let role_mappings = mappings
            .as_deref()
-            .ok_or(FetchAuthRulesError::RoleJwksNotConfigured)?;
+            .context("JWKs settings for this role were not configured")?;
        let mut rules = vec![];
        for setting in &role_mappings.jwks {
            rules.push(AuthRule {
--- a/proxy/src/auth/backend/mod.rs
+++ b/proxy/src/auth/backend/mod.rs
@@ -8,7 +8,6 @@ use std::net::IpAddr;
 use std::sync::Arc;
 use std::time::Duration;

-pub use console_redirect::ConsoleRedirectBackend;
 pub(crate) use console_redirect::WebAuthError;
 use ipnet::{Ipv4Net, Ipv6Net};
 use local::LocalBackend;
@@ -22,7 +21,7 @@ use crate::cache::Cached;
 use crate::context::RequestMonitoring;
 use crate::control_plane::errors::GetAuthInfoError;
 use crate::control_plane::provider::{CachedRoleSecret, ControlPlaneBackend};
-use crate::control_plane::AuthSecret;
+use crate::control_plane::{AuthSecret, NodeInfo};
 use crate::intern::EndpointIdInt;
 use crate::metrics::Metrics;
 use crate::proxy::connect_compute::ComputeConnectBackend;
@@ -37,7 +36,7 @@ use crate::{
        provider::{CachedAllowedIps, CachedNodeInfo},
        Api,
    },
-    stream,
+    stream, url,
 };
 use crate::{scram, EndpointCacheKey, EndpointId, RoleName};

@@ -66,9 +65,11 @@ impl<T> std::ops::Deref for MaybeOwned<'_, T> {
 /// * However, when we substitute `T` with [`ComputeUserInfoMaybeEndpoint`],
 ///   this helps us provide the credentials only to those auth
 ///   backends which require them for the authentication process.
-pub enum Backend<'a, T> {
+pub enum Backend<'a, T, D> {
    /// Cloud API (V2).
    ControlPlane(MaybeOwned<'a, ControlPlaneBackend>, T),
+    /// Authentication via a web browser.
+    ConsoleRedirect(MaybeOwned<'a, url::ApiUrl>, D),
    /// Local proxy uses configured auth credentials and does not wake compute
    Local(MaybeOwned<'a, LocalBackend>),
 }
@@ -89,7 +90,7 @@ impl Clone for Box<dyn TestBackend> {
    }
 }

-impl std::fmt::Display for Backend<'_, ()> {
+impl std::fmt::Display for Backend<'_, (), ()> {
    fn fmt(&self, fmt: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        match self {
            Self::ControlPlane(api, ()) => match &**api {
@@ -105,39 +106,46 @@ impl std::fmt::Display for Backend<'_, ()> {
                #[cfg(test)]
                ControlPlaneBackend::Test(_) => fmt.debug_tuple("ControlPlane::Test").finish(),
            },
+            Self::ConsoleRedirect(url, ()) => fmt
+                .debug_tuple("ConsoleRedirect")
+                .field(&url.as_str())
+                .finish(),
            Self::Local(_) => fmt.debug_tuple("Local").finish(),
        }
    }
 }

-impl<T> Backend<'_, T> {
+impl<T, D> Backend<'_, T, D> {
    /// Very similar to [`std::option::Option::as_ref`].
    /// This helps us pass structured config to async tasks.
-    pub(crate) fn as_ref(&self) -> Backend<'_, &T> {
+    pub(crate) fn as_ref(&self) -> Backend<'_, &T, &D> {
        match self {
            Self::ControlPlane(c, x) => Backend::ControlPlane(MaybeOwned::Borrowed(c), x),
+            Self::ConsoleRedirect(c, x) => Backend::ConsoleRedirect(MaybeOwned::Borrowed(c), x),
            Self::Local(l) => Backend::Local(MaybeOwned::Borrowed(l)),
        }
    }
 }

-impl<'a, T> Backend<'a, T> {
+impl<'a, T, D> Backend<'a, T, D> {
    /// Very similar to [`std::option::Option::map`].
    /// Maps [`Backend<T>`] to [`Backend<R>`] by applying
    /// a function to a contained value.
-    pub(crate) fn map<R>(self, f: impl FnOnce(T) -> R) -> Backend<'a, R> {
+    pub(crate) fn map<R>(self, f: impl FnOnce(T) -> R) -> Backend<'a, R, D> {
        match self {
            Self::ControlPlane(c, x) => Backend::ControlPlane(c, f(x)),
+            Self::ConsoleRedirect(c, x) => Backend::ConsoleRedirect(c, x),
            Self::Local(l) => Backend::Local(l),
        }
    }
 }
-impl<'a, T, E> Backend<'a, Result<T, E>> {
+impl<'a, T, D, E> Backend<'a, Result<T, E>, D> {
    /// Very similar to [`std::option::Option::transpose`].
    /// This is most useful for error handling.
-    pub(crate) fn transpose(self) -> Result<Backend<'a, T>, E> {
+    pub(crate) fn transpose(self) -> Result<Backend<'a, T, D>, E> {
        match self {
            Self::ControlPlane(c, x) => x.map(|x| Backend::ControlPlane(c, x)),
+            Self::ConsoleRedirect(c, x) => Ok(Backend::ConsoleRedirect(c, x)),
            Self::Local(l) => Ok(Backend::Local(l)),
        }
    }
@@ -167,12 +175,10 @@ impl ComputeUserInfo {
    }
 }

-#[cfg_attr(test, derive(Debug))]
 pub(crate) enum ComputeCredentialKeys {
    #[cfg(any(test, feature = "testing"))]
    Password(Vec<u8>),
    AuthKeys(AuthKeys),
-    JwtPayload(Vec<u8>),
    None,
 }

@@ -233,6 +239,7 @@ impl AuthenticationConfig {
    pub(crate) fn check_rate_limit(
        &self,
        ctx: &RequestMonitoring,
+        config: &AuthenticationConfig,
        secret: AuthSecret,
        endpoint: &EndpointId,
        is_cleartext: bool,
@@ -256,7 +263,7 @@ impl AuthenticationConfig {
        let limit_not_exceeded = self.rate_limiter.check(
            (
                endpoint_int,
-                MaskedIp::new(ctx.peer_addr(), self.rate_limit_ip_subnet),
+                MaskedIp::new(ctx.peer_addr(), config.rate_limit_ip_subnet),
            ),
            password_weight,
        );
@@ -330,6 +337,7 @@ async fn auth_quirks(
    let secret = if let Some(secret) = secret {
        config.check_rate_limit(
            ctx,
+            config,
            secret,
            &info.endpoint,
            unauthenticated_password.is_some() || allow_cleartext,
@@ -405,11 +413,12 @@ async fn authenticate_with_secret(
    classic::authenticate(ctx, info, client, config, secret).await
 }

-impl<'a> Backend<'a, ComputeUserInfoMaybeEndpoint> {
+impl<'a> Backend<'a, ComputeUserInfoMaybeEndpoint, &()> {
    /// Get username from the credentials.
    pub(crate) fn get_user(&self) -> &str {
        match self {
            Self::ControlPlane(_, user_info) => &user_info.user,
+            Self::ConsoleRedirect(_, ()) => "web",
            Self::Local(_) => "local",
        }
    }
@@ -423,7 +432,7 @@ impl<'a> Backend<'a, ComputeUserInfoMaybeEndpoint> {
        allow_cleartext: bool,
        config: &'static AuthenticationConfig,
        endpoint_rate_limiter: Arc<EndpointRateLimiter>,
-    ) -> auth::Result<Backend<'a, ComputeCredentials>> {
+    ) -> auth::Result<Backend<'a, ComputeCredentials, NodeInfo>> {
        let res = match self {
            Self::ControlPlane(api, user_info) => {
                info!(
@@ -444,6 +453,14 @@ impl<'a> Backend<'a, ComputeUserInfoMaybeEndpoint> {
                .await?;
                Backend::ControlPlane(api, credentials)
            }
+            // NOTE: this auth backend doesn't use client credentials.
+            Self::ConsoleRedirect(url, ()) => {
+                info!("performing web authentication");
+
+                let info = console_redirect::authenticate(ctx, config, &url, client).await?;
+
+                Backend::ConsoleRedirect(url, info)
+            }
            Self::Local(_) => {
                return Err(auth::AuthError::bad_auth_method("invalid for local proxy"))
            }
@@ -454,13 +471,14 @@ impl<'a> Backend<'a, ComputeUserInfoMaybeEndpoint> {
    }
 }

-impl Backend<'_, ComputeUserInfo> {
+impl Backend<'_, ComputeUserInfo, &()> {
    pub(crate) async fn get_role_secret(
        &self,
        ctx: &RequestMonitoring,
    ) -> Result<CachedRoleSecret, GetAuthInfoError> {
        match self {
            Self::ControlPlane(api, user_info) => api.get_role_secret(ctx, user_info).await,
+            Self::ConsoleRedirect(_, ()) => Ok(Cached::new_uncached(None)),
            Self::Local(_) => Ok(Cached::new_uncached(None)),
        }
    }
@@ -473,19 +491,21 @@ impl Backend<'_, ComputeUserInfo> {
            Self::ControlPlane(api, user_info) => {
                api.get_allowed_ips_and_secret(ctx, user_info).await
            }
+            Self::ConsoleRedirect(_, ()) => Ok((Cached::new_uncached(Arc::new(vec![])), None)),
            Self::Local(_) => Ok((Cached::new_uncached(Arc::new(vec![])), None)),
        }
    }
 }

 #[async_trait::async_trait]
-impl ComputeConnectBackend for Backend<'_, ComputeCredentials> {
+impl ComputeConnectBackend for Backend<'_, ComputeCredentials, NodeInfo> {
    async fn wake_compute(
        &self,
        ctx: &RequestMonitoring,
    ) -> Result<CachedNodeInfo, control_plane::errors::WakeComputeError> {
        match self {
            Self::ControlPlane(api, creds) => api.wake_compute(ctx, &creds.info).await,
+            Self::ConsoleRedirect(_, info) => Ok(Cached::new_uncached(info.clone())),
            Self::Local(local) => Ok(Cached::new_uncached(local.node_info.clone())),
        }
    }
@@ -493,6 +513,31 @@ impl ComputeConnectBackend for Backend<'_, ComputeCredentials> {
    fn get_keys(&self) -> &ComputeCredentialKeys {
        match self {
            Self::ControlPlane(_, creds) => &creds.keys,
+            Self::ConsoleRedirect(_, _) => &ComputeCredentialKeys::None,
+            Self::Local(_) => &ComputeCredentialKeys::None,
+        }
+    }
+}
+
+#[async_trait::async_trait]
+impl ComputeConnectBackend for Backend<'_, ComputeCredentials, &()> {
+    async fn wake_compute(
+        &self,
+        ctx: &RequestMonitoring,
+    ) -> Result<CachedNodeInfo, control_plane::errors::WakeComputeError> {
+        match self {
+            Self::ControlPlane(api, creds) => api.wake_compute(ctx, &creds.info).await,
+            Self::ConsoleRedirect(_, ()) => {
+                unreachable!("web auth flow doesn't support waking the compute")
+            }
+            Self::Local(local) => Ok(Cached::new_uncached(local.node_info.clone())),
+        }
+    }
+
+    fn get_keys(&self) -> &ComputeCredentialKeys {
+        match self {
+            Self::ControlPlane(_, creds) => &creds.keys,
+            Self::ConsoleRedirect(_, ()) => &ComputeCredentialKeys::None,
            Self::Local(_) => &ComputeCredentialKeys::None,
        }
    }
@@ -561,8 +606,7 @@ mod tests {
            &self,
            _ctx: &RequestMonitoring,
            _endpoint: crate::EndpointId,
-        ) -> Result<Vec<super::jwt::AuthRule>, control_plane::errors::GetEndpointJwksError>
-        {
+        ) -> anyhow::Result<Vec<super::jwt::AuthRule>> {
            unimplemented!()
        }

--- a/proxy/src/bin/local_proxy.rs
+++ b/proxy/src/bin/local_proxy.rs
@@ -6,12 +6,9 @@ use compute_api::spec::LocalProxySpec;
 use dashmap::DashMap;
 use futures::future::Either;
 use proxy::{
-    auth::{
-        self,
-        backend::{
-            jwt::JwkCache,
-            local::{LocalBackend, JWKS_ROLE_MAP},
-        },
+    auth::backend::{
+        jwt::JwkCache,
+        local::{LocalBackend, JWKS_ROLE_MAP},
    },
    cancellation::CancellationHandlerMain,
    config::{self, AuthenticationConfig, HttpConfig, ProxyConfig, RetryConfig},
@@ -135,7 +132,6 @@ async fn main() -> anyhow::Result<()> {

    let args = LocalProxyCliArgs::parse();
    let config = build_config(&args)?;
-    let auth_backend = build_auth_backend(&args)?;

    // before we bind to any ports, write the process ID to a file
    // so that compute-ctl can find our process later
@@ -197,7 +193,6 @@ async fn main() -> anyhow::Result<()> {

    let task = serverless::task_main(
        config,
-        auth_backend,
        http_listener,
        shutdown.clone(),
        Arc::new(CancellationHandlerMain::new(
@@ -262,6 +257,9 @@ fn build_config(args: &LocalProxyCliArgs) -> anyhow::Result<&'static ProxyConfig

    Ok(Box::leak(Box::new(ProxyConfig {
        tls_config: None,
+        auth_backend: proxy::auth::Backend::Local(proxy::auth::backend::MaybeOwned::Owned(
+            LocalBackend::new(args.compute),
+        )),
        metric_collection: None,
        allow_self_signed_compute: false,
        http_config,
@@ -288,17 +286,6 @@ fn build_config(args: &LocalProxyCliArgs) -> anyhow::Result<&'static ProxyConfig
    })))
 }

-/// auth::Backend is created at proxy startup, and lives forever.
-fn build_auth_backend(
-    args: &LocalProxyCliArgs,
-) -> anyhow::Result<&'static auth::Backend<'static, ()>> {
-    let auth_backend = proxy::auth::Backend::Local(proxy::auth::backend::MaybeOwned::Owned(
-        LocalBackend::new(args.compute),
-    ));
-
-    Ok(Box::leak(Box::new(auth_backend)))
-}
-
 async fn refresh_config_loop(path: Utf8PathBuf, rx: Arc<Notify>) {
    loop {
        rx.notified().await;
--- a/proxy/src/bin/proxy.rs
+++ b/proxy/src/bin/proxy.rs
@@ -10,7 +10,6 @@ use futures::future::Either;
 use proxy::auth;
 use proxy::auth::backend::jwt::JwkCache;
 use proxy::auth::backend::AuthRateLimiter;
-use proxy::auth::backend::ConsoleRedirectBackend;
 use proxy::auth::backend::MaybeOwned;
 use proxy::cancellation::CancelMap;
 use proxy::cancellation::CancellationHandler;
@@ -312,12 +311,8 @@ async fn main() -> anyhow::Result<()> {

    let args = ProxyCliArgs::parse();
    let config = build_config(&args)?;
-    let auth_backend = build_auth_backend(&args)?;

-    match auth_backend {
-        Either::Left(auth_backend) => info!("Authentication backend: {auth_backend}"),
-        Either::Right(auth_backend) => info!("Authentication backend: {auth_backend:?}"),
-    };
+    info!("Authentication backend: {}", config.auth_backend);
    info!("Using region: {}", args.aws_region);

    let region_provider =
@@ -464,41 +459,24 @@ async fn main() -> anyhow::Result<()> {
    // client facing tasks. these will exit on error or on cancellation
    // cancellation returns Ok(())
    let mut client_tasks = JoinSet::new();
-    match auth_backend {
-        Either::Left(auth_backend) => {
-            if let Some(proxy_listener) = proxy_listener {
-                client_tasks.spawn(proxy::proxy::task_main(
-                    config,
-                    auth_backend,
-                    proxy_listener,
-                    cancellation_token.clone(),
-                    cancellation_handler.clone(),
-                    endpoint_rate_limiter.clone(),
-                ));
-            }
+    if let Some(proxy_listener) = proxy_listener {
+        client_tasks.spawn(proxy::proxy::task_main(
+            config,
+            proxy_listener,
+            cancellation_token.clone(),
+            cancellation_handler.clone(),
+            endpoint_rate_limiter.clone(),
+        ));
+    }

-            if let Some(serverless_listener) = serverless_listener {
-                client_tasks.spawn(serverless::task_main(
-                    config,
-                    auth_backend,
-                    serverless_listener,
-                    cancellation_token.clone(),
-                    cancellation_handler.clone(),
-                    endpoint_rate_limiter.clone(),
-                ));
-            }
-        }
-        Either::Right(auth_backend) => {
-            if let Some(proxy_listener) = proxy_listener {
-                client_tasks.spawn(proxy::console_redirect_proxy::task_main(
-                    config,
-                    auth_backend,
-                    proxy_listener,
-                    cancellation_token.clone(),
-                    cancellation_handler.clone(),
-                ));
-            }
-        }
+    if let Some(serverless_listener) = serverless_listener {
+        client_tasks.spawn(serverless::task_main(
+            config,
+            serverless_listener,
+            cancellation_token.clone(),
+            cancellation_handler.clone(),
+            endpoint_rate_limiter.clone(),
+        ));
    }

    client_tasks.spawn(proxy::context::parquet::worker(
@@ -528,7 +506,7 @@ async fn main() -> anyhow::Result<()> {
        ));
    }

-    if let Either::Left(auth::Backend::ControlPlane(api, _)) = &auth_backend {
+    if let auth::Backend::ControlPlane(api, _) = &config.auth_backend {
        if let proxy::control_plane::provider::ControlPlaneBackend::Management(api) = &**api {
            match (redis_notifications_client, regional_redis_client.clone()) {
                (None, None) => {}
@@ -632,83 +610,7 @@ fn build_config(args: &ProxyCliArgs) -> anyhow::Result<&'static ProxyConfig> {
        bail!("dynamic rate limiter should be disabled");
    }

-    let config::ConcurrencyLockOptions {
-        shards,
-        limiter,
-        epoch,
-        timeout,
-    } = args.connect_compute_lock.parse()?;
-    info!(
-        ?limiter,
-        shards,
-        ?epoch,
-        "Using NodeLocks (connect_compute)"
-    );
-    let connect_compute_locks = control_plane::locks::ApiLocks::new(
-        "connect_compute_lock",
-        limiter,
-        shards,
-        timeout,
-        epoch,
-        &Metrics::get().proxy.connect_compute_lock,
-    )?;
-
-    let http_config = HttpConfig {
-        accept_websockets: !args.is_auth_broker,
-        pool_options: GlobalConnPoolOptions {
-            max_conns_per_endpoint: args.sql_over_http.sql_over_http_pool_max_conns_per_endpoint,
-            gc_epoch: args.sql_over_http.sql_over_http_pool_gc_epoch,
-            pool_shards: args.sql_over_http.sql_over_http_pool_shards,
-            idle_timeout: args.sql_over_http.sql_over_http_idle_timeout,
-            opt_in: args.sql_over_http.sql_over_http_pool_opt_in,
-            max_total_conns: args.sql_over_http.sql_over_http_pool_max_total_conns,
-        },
-        cancel_set: CancelSet::new(args.sql_over_http.sql_over_http_cancel_set_shards),
-        client_conn_threshold: args.sql_over_http.sql_over_http_client_conn_threshold,
-        max_request_size_bytes: args.sql_over_http.sql_over_http_max_request_size_bytes,
-        max_response_size_bytes: args.sql_over_http.sql_over_http_max_response_size_bytes,
-    };
-    let authentication_config = AuthenticationConfig {
-        jwks_cache: JwkCache::default(),
-        thread_pool,
-        scram_protocol_timeout: args.scram_protocol_timeout,
-        rate_limiter_enabled: args.auth_rate_limit_enabled,
-        rate_limiter: AuthRateLimiter::new(args.auth_rate_limit.clone()),
-        rate_limit_ip_subnet: args.auth_rate_limit_ip_subnet,
-        ip_allowlist_check_enabled: !args.is_private_access_proxy,
-        is_auth_broker: args.is_auth_broker,
-        accept_jwts: args.is_auth_broker,
-        webauth_confirmation_timeout: args.webauth_confirmation_timeout,
-    };
-
-    let config = ProxyConfig {
-        tls_config,
-        metric_collection,
-        allow_self_signed_compute: args.allow_self_signed_compute,
-        http_config,
-        authentication_config,
-        proxy_protocol_v2: args.proxy_protocol_v2,
-        handshake_timeout: args.handshake_timeout,
-        region: args.region.clone(),
-        wake_compute_retry_config: config::RetryConfig::parse(&args.wake_compute_retry)?,
-        connect_compute_locks,
-        connect_to_compute_retry_config: config::RetryConfig::parse(
-            &args.connect_to_compute_retry,
-        )?,
-    };
-
-    let config = Box::leak(Box::new(config));
-
-    tokio::spawn(config.connect_compute_locks.garbage_collect_worker());
-
-    Ok(config)
-}
-
-/// auth::Backend is created at proxy startup, and lives forever.
-fn build_auth_backend(
-    args: &ProxyCliArgs,
-) -> anyhow::Result<Either<&'static auth::Backend<'static, ()>, &'static ConsoleRedirectBackend>> {
-    match &args.auth_backend {
+    let auth_backend = match &args.auth_backend {
        AuthBackendType::Console => {
            let wake_compute_cache_config: CacheOptions = args.wake_compute_cache.parse()?;
            let project_info_cache_config: ProjectInfoCacheOptions =
@@ -758,11 +660,12 @@ fn build_auth_backend(
                wake_compute_endpoint_rate_limiter,
            );
            let api = control_plane::provider::ControlPlaneBackend::Management(api);
-            let auth_backend = auth::Backend::ControlPlane(MaybeOwned::Owned(api), ());
+            auth::Backend::ControlPlane(MaybeOwned::Owned(api), ())
+        }

-            let config = Box::leak(Box::new(auth_backend));
-
-            Ok(Either::Left(config))
+        AuthBackendType::Web => {
+            let url = args.uri.parse()?;
+            auth::Backend::ConsoleRedirect(MaybeOwned::Owned(url), ())
        }

        #[cfg(feature = "testing")]
@@ -770,23 +673,79 @@ fn build_auth_backend(
            let url = args.auth_endpoint.parse()?;
            let api = control_plane::provider::mock::Api::new(url, !args.is_private_access_proxy);
            let api = control_plane::provider::ControlPlaneBackend::PostgresMock(api);
-
-            let auth_backend = auth::Backend::ControlPlane(MaybeOwned::Owned(api), ());
-
-            let config = Box::leak(Box::new(auth_backend));
-
-            Ok(Either::Left(config))
+            auth::Backend::ControlPlane(MaybeOwned::Owned(api), ())
        }
+    };

-        AuthBackendType::Web => {
-            let url = args.uri.parse()?;
-            let backend = ConsoleRedirectBackend::new(url);
+    let config::ConcurrencyLockOptions {
+        shards,
+        limiter,
+        epoch,
+        timeout,
+    } = args.connect_compute_lock.parse()?;
+    info!(
+        ?limiter,
+        shards,
+        ?epoch,
+        "Using NodeLocks (connect_compute)"
+    );
+    let connect_compute_locks = control_plane::locks::ApiLocks::new(
+        "connect_compute_lock",
+        limiter,
+        shards,
+        timeout,
+        epoch,
+        &Metrics::get().proxy.connect_compute_lock,
+    )?;

-            let config = Box::leak(Box::new(backend));
+    let http_config = HttpConfig {
+        accept_websockets: !args.is_auth_broker,
+        pool_options: GlobalConnPoolOptions {
+            max_conns_per_endpoint: args.sql_over_http.sql_over_http_pool_max_conns_per_endpoint,
+            gc_epoch: args.sql_over_http.sql_over_http_pool_gc_epoch,
+            pool_shards: args.sql_over_http.sql_over_http_pool_shards,
+            idle_timeout: args.sql_over_http.sql_over_http_idle_timeout,
+            opt_in: args.sql_over_http.sql_over_http_pool_opt_in,
+            max_total_conns: args.sql_over_http.sql_over_http_pool_max_total_conns,
+        },
+        cancel_set: CancelSet::new(args.sql_over_http.sql_over_http_cancel_set_shards),
+        client_conn_threshold: args.sql_over_http.sql_over_http_client_conn_threshold,
+        max_request_size_bytes: args.sql_over_http.sql_over_http_max_request_size_bytes,
+        max_response_size_bytes: args.sql_over_http.sql_over_http_max_response_size_bytes,
+    };
+    let authentication_config = AuthenticationConfig {
+        jwks_cache: JwkCache::default(),
+        thread_pool,
+        scram_protocol_timeout: args.scram_protocol_timeout,
+        rate_limiter_enabled: args.auth_rate_limit_enabled,
+        rate_limiter: AuthRateLimiter::new(args.auth_rate_limit.clone()),
+        rate_limit_ip_subnet: args.auth_rate_limit_ip_subnet,
+        ip_allowlist_check_enabled: !args.is_private_access_proxy,
+        is_auth_broker: args.is_auth_broker,
+        accept_jwts: args.is_auth_broker,
+        webauth_confirmation_timeout: args.webauth_confirmation_timeout,
+    };

-            Ok(Either::Right(config))
-        }
-    }
+    let config = Box::leak(Box::new(ProxyConfig {
+        tls_config,
+        auth_backend,
+        metric_collection,
+        allow_self_signed_compute: args.allow_self_signed_compute,
+        http_config,
+        authentication_config,
+        proxy_protocol_v2: args.proxy_protocol_v2,
+        handshake_timeout: args.handshake_timeout,
+        region: args.region.clone(),
+        wake_compute_retry_config: config::RetryConfig::parse(&args.wake_compute_retry)?,
+        connect_compute_locks,
+        connect_to_compute_retry_config: config::RetryConfig::parse(
+            &args.connect_to_compute_retry,
+        )?,
+    }));
+
+    tokio::spawn(config.connect_compute_locks.garbage_collect_worker());
+
+    Ok(config)
 }

 #[cfg(test)]
--- a/proxy/src/config.rs
+++ b/proxy/src/config.rs
@@ -1,5 +1,8 @@
 use crate::{
-    auth::backend::{jwt::JwkCache, AuthRateLimiter},
+    auth::{
+        self,
+        backend::{jwt::JwkCache, AuthRateLimiter},
+    },
    control_plane::locks::ApiLocks,
    rate_limiter::{RateBucketInfo, RateLimitAlgorithm, RateLimiterConfig},
    scram::threadpool::ThreadPool,
@@ -26,6 +29,7 @@ use x509_parser::oid_registry;

 pub struct ProxyConfig {
    pub tls_config: Option<TlsConfig>,
+    pub auth_backend: auth::Backend<'static, (), ()>,
    pub metric_collection: Option<MetricCollectionConfig>,
    pub allow_self_signed_compute: bool,
    pub http_config: HttpConfig,
--- a/proxy/src/console_redirect_proxy.rs
+++ b/proxy/src/console_redirect_proxy.rs
@@ -1,217 +0,0 @@
-use crate::auth::backend::ConsoleRedirectBackend;
-use crate::config::{ProxyConfig, ProxyProtocolV2};
-use crate::proxy::{
-    prepare_client_connection, run_until_cancelled, ClientRequestError, ErrorSource,
-};
-use crate::{
-    cancellation::{CancellationHandlerMain, CancellationHandlerMainInternal},
-    context::RequestMonitoring,
-    error::ReportableError,
-    metrics::{Metrics, NumClientConnectionsGuard},
-    protocol2::read_proxy_protocol,
-    proxy::handshake::{handshake, HandshakeData},
-};
-use futures::TryFutureExt;
-use std::sync::Arc;
-use tokio::io::{AsyncRead, AsyncWrite, AsyncWriteExt};
-use tokio_util::sync::CancellationToken;
-use tracing::{error, info, Instrument};
-
-use crate::proxy::{
-    connect_compute::{connect_to_compute, TcpMechanism},
-    passthrough::ProxyPassthrough,
-};
-
-pub async fn task_main(
-    config: &'static ProxyConfig,
-    backend: &'static ConsoleRedirectBackend,
-    listener: tokio::net::TcpListener,
-    cancellation_token: CancellationToken,
-    cancellation_handler: Arc<CancellationHandlerMain>,
-) -> anyhow::Result<()> {
-    scopeguard::defer! {
-        info!("proxy has shut down");
-    }
-
-    // When set for the server socket, the keepalive setting
-    // will be inherited by all accepted client sockets.
-    socket2::SockRef::from(&listener).set_keepalive(true)?;
-
-    let connections = tokio_util::task::task_tracker::TaskTracker::new();
-
-    while let Some(accept_result) =
-        run_until_cancelled(listener.accept(), &cancellation_token).await
-    {
-        let (socket, peer_addr) = accept_result?;
-
-        let conn_gauge = Metrics::get()
-            .proxy
-            .client_connections
-            .guard(crate::metrics::Protocol::Tcp);
-
-        let session_id = uuid::Uuid::new_v4();
-        let cancellation_handler = Arc::clone(&cancellation_handler);
-
-        tracing::info!(protocol = "tcp", %session_id, "accepted new TCP connection");
-
-        connections.spawn(async move {
-            let (socket, peer_addr) = match read_proxy_protocol(socket).await {
-                Err(e) => {
-                    error!("per-client task finished with an error: {e:#}");
-                    return;
-                }
-                Ok((_socket, None)) if config.proxy_protocol_v2 == ProxyProtocolV2::Required => {
-                    error!("missing required proxy protocol header");
-                    return;
-                }
-                Ok((_socket, Some(_))) if config.proxy_protocol_v2 == ProxyProtocolV2::Rejected => {
-                    error!("proxy protocol header not supported");
-                    return;
-                }
-                Ok((socket, Some(addr))) => (socket, addr.ip()),
-                Ok((socket, None)) => (socket, peer_addr.ip()),
-            };
-
-            match socket.inner.set_nodelay(true) {
-                Ok(()) => {}
-                Err(e) => {
-                    error!("per-client task finished with an error: failed to set socket option: {e:#}");
-                    return;
-                }
-            };
-
-            let ctx = RequestMonitoring::new(
-                session_id,
-                peer_addr,
-                crate::metrics::Protocol::Tcp,
-                &config.region,
-            );
-            let span = ctx.span();
-
-            let startup = Box::pin(
-                handle_client(
-                    config,
-                    backend,
-                    &ctx,
-                    cancellation_handler,
-                    socket,
-                    conn_gauge,
-                )
-                .instrument(span.clone()),
-            );
-            let res = startup.await;
-
-            match res {
-                Err(e) => {
-                    // todo: log and push to ctx the error kind
-                    ctx.set_error_kind(e.get_error_kind());
-                    error!(parent: &span, "per-client task finished with an error: {e:#}");
-                }
-                Ok(None) => {
-                    ctx.set_success();
-                }
-                Ok(Some(p)) => {
-                    ctx.set_success();
-                    ctx.log_connect();
-                    match p.proxy_pass().instrument(span.clone()).await {
-                        Ok(()) => {}
-                        Err(ErrorSource::Client(e)) => {
-                            error!(parent: &span, "per-client task finished with an IO error from the client: {e:#}");
-                        }
-                        Err(ErrorSource::Compute(e)) => {
-                            error!(parent: &span, "per-client task finished with an IO error from the compute: {e:#}");
-                        }
-                    }
-                }
-            }
-        });
-    }
-
-    connections.close();
-    drop(listener);
-
-    // Drain connections
-    connections.wait().await;
-
-    Ok(())
-}
-
-pub(crate) async fn handle_client<S: AsyncRead + AsyncWrite + Unpin>(
-    config: &'static ProxyConfig,
-    backend: &'static ConsoleRedirectBackend,
-    ctx: &RequestMonitoring,
-    cancellation_handler: Arc<CancellationHandlerMain>,
-    stream: S,
-    conn_gauge: NumClientConnectionsGuard<'static>,
-) -> Result<Option<ProxyPassthrough<CancellationHandlerMainInternal, S>>, ClientRequestError> {
-    info!(
-        protocol = %ctx.protocol(),
-        "handling interactive connection from client"
-    );
-
-    let metrics = &Metrics::get().proxy;
-    let proto = ctx.protocol();
-    let request_gauge = metrics.connection_requests.guard(proto);
-
-    let tls = config.tls_config.as_ref();
-
-    let record_handshake_error = !ctx.has_private_peer_addr();
-    let pause = ctx.latency_timer_pause(crate::metrics::Waiting::Client);
-    let do_handshake = handshake(ctx, stream, tls, record_handshake_error);
-    let (mut stream, params) =
-        match tokio::time::timeout(config.handshake_timeout, do_handshake).await?? {
-            HandshakeData::Startup(stream, params) => (stream, params),
-            HandshakeData::Cancel(cancel_key_data) => {
-                return Ok(cancellation_handler
-                    .cancel_session(cancel_key_data, ctx.session_id())
-                    .await
-                    .map(|()| None)?)
-            }
-        };
-    drop(pause);
-
-    ctx.set_db_options(params.clone());
-
-    let user_info = match backend
-        .authenticate(ctx, &config.authentication_config, &mut stream)
-        .await
-    {
-        Ok(auth_result) => auth_result,
-        Err(e) => {
-            return stream.throw_error(e).await?;
-        }
-    };
-
-    let mut node = connect_to_compute(
-        ctx,
-        &TcpMechanism {
-            params: &params,
-            locks: &config.connect_compute_locks,
-        },
-        &user_info,
-        config.allow_self_signed_compute,
-        config.wake_compute_retry_config,
-        config.connect_to_compute_retry_config,
-    )
-    .or_else(|e| stream.throw_error(e))
-    .await?;
-
-    let session = cancellation_handler.get_session();
-    prepare_client_connection(&node, &session, &mut stream).await?;
-
-    // Before proxy passing, forward to compute whatever data is left in the
-    // PqStream input buffer. Normally there is none, but our serverless npm
-    // driver in pipeline mode sends startup, password and first query
-    // immediately after opening the connection.
-    let (stream, read_buf) = stream.into_inner();
-    node.stream.write_all(&read_buf).await?;
-
-    Ok(Some(ProxyPassthrough {
-        client: stream,
-        aux: node.aux.clone(),
-        compute: node,
-        _req: request_gauge,
-        _conn: conn_gauge,
-        _cancel: session,
-    }))
-}
--- a/proxy/src/control_plane/provider/mock.rs
+++ b/proxy/src/control_plane/provider/mock.rs
@@ -5,8 +5,7 @@ use super::{
    AuthInfo, AuthSecret, CachedNodeInfo, NodeInfo,
 };
 use crate::{
-    auth::backend::jwt::AuthRule, context::RequestMonitoring,
-    control_plane::errors::GetEndpointJwksError, intern::RoleNameInt, RoleName,
+    auth::backend::jwt::AuthRule, context::RequestMonitoring, intern::RoleNameInt, RoleName,
 };
 use crate::{auth::backend::ComputeUserInfo, compute, error::io_error, scram, url::ApiUrl};
 use crate::{auth::IpPattern, cache::Cached};
@@ -121,10 +120,7 @@ impl Api {
        })
    }

-    async fn do_get_endpoint_jwks(
-        &self,
-        endpoint: EndpointId,
-    ) -> Result<Vec<AuthRule>, GetEndpointJwksError> {
+    async fn do_get_endpoint_jwks(&self, endpoint: EndpointId) -> anyhow::Result<Vec<AuthRule>> {
        let (client, connection) =
            tokio_postgres::connect(self.endpoint.as_str(), tokio_postgres::NoTls).await?;

@@ -228,7 +224,7 @@ impl super::Api for Api {
        &self,
        _ctx: &RequestMonitoring,
        endpoint: EndpointId,
-    ) -> Result<Vec<AuthRule>, GetEndpointJwksError> {
+    ) -> anyhow::Result<Vec<AuthRule>> {
        self.do_get_endpoint_jwks(endpoint).await
    }

--- a/proxy/src/control_plane/provider/mod.rs
+++ b/proxy/src/control_plane/provider/mod.rs
@@ -6,7 +6,7 @@ use super::messages::{ControlPlaneError, MetricsAuxInfo};
 use crate::{
    auth::{
        backend::{
-            jwt::{AuthRule, FetchAuthRules, FetchAuthRulesError},
+            jwt::{AuthRule, FetchAuthRules},
            ComputeCredentialKeys, ComputeUserInfo,
        },
        IpPattern,
@@ -44,7 +44,7 @@ pub(crate) mod errors {
    pub(crate) enum ApiError {
        /// Error returned by the console itself.
        #[error("{REQUEST_FAILED} with {0}")]
-        ControlPlane(Box<ControlPlaneError>),
+        ControlPlane(ControlPlaneError),

        /// Various IO errors like broken pipe or malformed payload.
        #[error("{REQUEST_FAILED}: {0}")]
@@ -81,16 +81,16 @@ pub(crate) mod errors {
                    Reason::EndpointNotFound => ErrorKind::User,
                    Reason::BranchNotFound => ErrorKind::User,
                    Reason::RateLimitExceeded => ErrorKind::ServiceRateLimit,
-                    Reason::NonDefaultBranchComputeTimeExceeded => ErrorKind::Quota,
-                    Reason::ActiveTimeQuotaExceeded => ErrorKind::Quota,
-                    Reason::ComputeTimeQuotaExceeded => ErrorKind::Quota,
-                    Reason::WrittenDataQuotaExceeded => ErrorKind::Quota,
-                    Reason::DataTransferQuotaExceeded => ErrorKind::Quota,
-                    Reason::LogicalSizeQuotaExceeded => ErrorKind::Quota,
+                    Reason::NonDefaultBranchComputeTimeExceeded => ErrorKind::User,
+                    Reason::ActiveTimeQuotaExceeded => ErrorKind::User,
+                    Reason::ComputeTimeQuotaExceeded => ErrorKind::User,
+                    Reason::WrittenDataQuotaExceeded => ErrorKind::User,
+                    Reason::DataTransferQuotaExceeded => ErrorKind::User,
+                    Reason::LogicalSizeQuotaExceeded => ErrorKind::User,
                    Reason::ConcurrencyLimitReached => ErrorKind::ControlPlane,
                    Reason::LockAlreadyTaken => ErrorKind::ControlPlane,
                    Reason::RunningOperations => ErrorKind::ControlPlane,
-                    Reason::Unknown => match &**e {
+                    Reason::Unknown => match &e {
                        ControlPlaneError {
                            http_status_code:
                                http::StatusCode::NOT_FOUND | http::StatusCode::NOT_ACCEPTABLE,
@@ -103,7 +103,7 @@ pub(crate) mod errors {
                        } if error
                            .contains("compute time quota of non-primary branches is exceeded") =>
                        {
-                            crate::error::ErrorKind::Quota
+                            crate::error::ErrorKind::User
                        }
                        ControlPlaneError {
                            http_status_code: http::StatusCode::LOCKED,
@@ -112,7 +112,7 @@ pub(crate) mod errors {
                        } if error.contains("quota exceeded")
                            || error.contains("the limit for current plan reached") =>
                        {
-                            crate::error::ErrorKind::Quota
+                            crate::error::ErrorKind::User
                        }
                        ControlPlaneError {
                            http_status_code: http::StatusCode::TOO_MANY_REQUESTS,
@@ -246,33 +246,6 @@ pub(crate) mod errors {
            }
        }
    }
-
-    #[derive(Debug, Error)]
-    pub enum GetEndpointJwksError {
-        #[error("endpoint not found")]
-        EndpointNotFound,
-
-        #[error("failed to build control plane request: {0}")]
-        RequestBuild(#[source] reqwest::Error),
-
-        #[error("failed to send control plane request: {0}")]
-        RequestExecute(#[source] reqwest_middleware::Error),
-
-        #[error(transparent)]
-        ControlPlane(#[from] ApiError),
-
-        #[cfg(any(test, feature = "testing"))]
-        #[error(transparent)]
-        TokioPostgres(#[from] tokio_postgres::Error),
-
-        #[cfg(any(test, feature = "testing"))]
-        #[error(transparent)]
-        ParseUrl(#[from] url::ParseError),
-
-        #[cfg(any(test, feature = "testing"))]
-        #[error(transparent)]
-        TaskJoin(#[from] tokio::task::JoinError),
-    }
 }

 /// Auth secret which is managed by the cloud.
@@ -336,7 +309,7 @@ impl NodeInfo {
            #[cfg(any(test, feature = "testing"))]
            ComputeCredentialKeys::Password(password) => self.config.password(password),
            ComputeCredentialKeys::AuthKeys(auth_keys) => self.config.auth_keys(*auth_keys),
-            ComputeCredentialKeys::JwtPayload(_) | ComputeCredentialKeys::None => &mut self.config,
+            ComputeCredentialKeys::None => &mut self.config,
        };
    }
 }
@@ -369,7 +342,7 @@ pub(crate) trait Api {
        &self,
        ctx: &RequestMonitoring,
        endpoint: EndpointId,
-    ) -> Result<Vec<AuthRule>, errors::GetEndpointJwksError>;
+    ) -> anyhow::Result<Vec<AuthRule>>;

    /// Wake up the compute node and return the corresponding connection info.
    async fn wake_compute(
@@ -428,7 +401,7 @@ impl Api for ControlPlaneBackend {
        &self,
        ctx: &RequestMonitoring,
        endpoint: EndpointId,
-    ) -> Result<Vec<AuthRule>, errors::GetEndpointJwksError> {
+    ) -> anyhow::Result<Vec<AuthRule>> {
        match self {
            Self::Management(api) => api.get_endpoint_jwks(ctx, endpoint).await,
            #[cfg(any(test, feature = "testing"))]
@@ -610,9 +583,7 @@ impl FetchAuthRules for ControlPlaneBackend {
        &self,
        ctx: &RequestMonitoring,
        endpoint: EndpointId,
-    ) -> Result<Vec<AuthRule>, FetchAuthRulesError> {
-        self.get_endpoint_jwks(ctx, endpoint)
-            .await
-            .map_err(FetchAuthRulesError::GetEndpointJwks)
+    ) -> anyhow::Result<Vec<AuthRule>> {
+        self.get_endpoint_jwks(ctx, endpoint).await
    }
 }
--- a/proxy/src/control_plane/provider/neon.rs
+++ b/proxy/src/control_plane/provider/neon.rs
@@ -9,10 +9,7 @@ use super::{
 use crate::{
    auth::backend::{jwt::AuthRule, ComputeUserInfo},
    compute,
-    control_plane::{
-        errors::GetEndpointJwksError,
-        messages::{ColdStartInfo, EndpointJwksResponse, Reason},
-    },
+    control_plane::messages::{ColdStartInfo, EndpointJwksResponse, Reason},
    http,
    metrics::{CacheOutcome, Metrics},
    rate_limiter::WakeComputeRateLimiter,
@@ -20,11 +17,12 @@ use crate::{
 };
 use crate::{cache::Cached, context::RequestMonitoring};
 use ::http::{header::AUTHORIZATION, HeaderName};
+use anyhow::bail;
 use futures::TryFutureExt;
 use std::{sync::Arc, time::Duration};
 use tokio::time::Instant;
 use tokio_postgres::config::SslMode;
-use tracing::{debug, info, info_span, warn, Instrument};
+use tracing::{debug, error, info, info_span, warn, Instrument};

 const X_REQUEST_ID: HeaderName = HeaderName::from_static("x-request-id");

@@ -139,14 +137,14 @@ impl Api {
        &self,
        ctx: &RequestMonitoring,
        endpoint: EndpointId,
-    ) -> Result<Vec<AuthRule>, GetEndpointJwksError> {
+    ) -> anyhow::Result<Vec<AuthRule>> {
        if !self
            .caches
            .endpoints_cache
            .is_valid(ctx, &endpoint.normalize())
            .await
        {
-            return Err(GetEndpointJwksError::EndpointNotFound);
+            bail!("endpoint not found");
        }
        let request_id = ctx.session_id().to_string();
        async {
@@ -161,17 +159,12 @@ impl Api {
                .header(X_REQUEST_ID, &request_id)
                .header(AUTHORIZATION, format!("Bearer {}", &self.jwt))
                .query(&[("session_id", ctx.session_id())])
-                .build()
-                .map_err(GetEndpointJwksError::RequestBuild)?;
+                .build()?;

            info!(url = request.url().as_str(), "sending http request");
            let start = Instant::now();
            let pause = ctx.latency_timer_pause(crate::metrics::Waiting::Cplane);
-            let response = self
-                .endpoint
-                .execute(request)
-                .await
-                .map_err(GetEndpointJwksError::RequestExecute)?;
+            let response = self.endpoint.execute(request).await?;
            drop(pause);
            info!(duration = ?start.elapsed(), "received http response");

@@ -337,7 +330,7 @@ impl super::Api for Api {
        &self,
        ctx: &RequestMonitoring,
        endpoint: EndpointId,
-    ) -> Result<Vec<AuthRule>, GetEndpointJwksError> {
+    ) -> anyhow::Result<Vec<AuthRule>> {
        self.do_get_endpoint_jwks(ctx, endpoint).await
    }

@@ -355,7 +348,7 @@ impl super::Api for Api {
                    let (cached, info) = cached.take_value();
                    let info = info.map_err(|c| {
                        info!(key = &*key, "found cached wake_compute error");
-                        WakeComputeError::ApiError(ApiError::ControlPlane(Box::new(*c)))
+                        WakeComputeError::ApiError(ApiError::ControlPlane(*c))
                    })?;

                    debug!(key = &*key, "found cached compute node info");
@@ -425,7 +418,7 @@ impl super::Api for Api {

                    self.caches.node_info.insert_ttl(
                        key,
-                        Err(err.clone()),
+                        Err(Box::new(err.clone())),
                        Duration::from_secs(30),
                    );

@@ -463,8 +456,8 @@ async fn parse_body<T: for<'a> serde::Deserialize<'a>>(
    });
    body.http_status_code = status;

-    warn!("console responded with an error ({status}): {body:?}");
-    Err(ApiError::ControlPlane(Box::new(body)))
+    error!("console responded with an error ({status}): {body:?}");
+    Err(ApiError::ControlPlane(body))
 }

 fn parse_host_port(input: &str) -> Option<(&str, u16)> {
--- a/proxy/src/error.rs
+++ b/proxy/src/error.rs
@@ -49,10 +49,6 @@ pub enum ErrorKind {
    #[label(rename = "serviceratelimit")]
    ServiceRateLimit,

-    /// Proxy quota limit violation
-    #[label(rename = "quota")]
-    Quota,
-
    /// internal errors
    Service,

@@ -74,7 +70,6 @@ impl ErrorKind {
            ErrorKind::ClientDisconnect => "clientdisconnect",
            ErrorKind::RateLimit => "ratelimit",
            ErrorKind::ServiceRateLimit => "serviceratelimit",
-            ErrorKind::Quota => "quota",
            ErrorKind::Service => "service",
            ErrorKind::ControlPlane => "controlplane",
            ErrorKind::Postgres => "postgres",
--- a/proxy/src/lib.rs
+++ b/proxy/src/lib.rs
@@ -95,7 +95,6 @@ pub mod cache;
 pub mod cancellation;
 pub mod compute;
 pub mod config;
-pub mod console_redirect_proxy;
 pub mod context;
 pub mod control_plane;
 pub mod error;
--- a/proxy/src/proxy/mod.rs
+++ b/proxy/src/proxy/mod.rs
@@ -35,7 +35,7 @@ use std::sync::Arc;
 use thiserror::Error;
 use tokio::io::{AsyncRead, AsyncWrite, AsyncWriteExt};
 use tokio_util::sync::CancellationToken;
-use tracing::{error, info, warn, Instrument};
+use tracing::{error, info, Instrument};

 use self::{
    connect_compute::{connect_to_compute, TcpMechanism},
@@ -61,7 +61,6 @@ pub async fn run_until_cancelled<F: std::future::Future>(

 pub async fn task_main(
    config: &'static ProxyConfig,
-    auth_backend: &'static auth::Backend<'static, ()>,
    listener: tokio::net::TcpListener,
    cancellation_token: CancellationToken,
    cancellation_handler: Arc<CancellationHandlerMain>,
@@ -96,15 +95,15 @@ pub async fn task_main(
        connections.spawn(async move {
            let (socket, peer_addr) = match read_proxy_protocol(socket).await {
                Err(e) => {
-                    warn!("per-client task finished with an error: {e:#}");
+                    error!("per-client task finished with an error: {e:#}");
                    return;
                }
                Ok((_socket, None)) if config.proxy_protocol_v2 == ProxyProtocolV2::Required => {
-                    warn!("missing required proxy protocol header");
+                    error!("missing required proxy protocol header");
                    return;
                }
                Ok((_socket, Some(_))) if config.proxy_protocol_v2 == ProxyProtocolV2::Rejected => {
-                    warn!("proxy protocol header not supported");
+                    error!("proxy protocol header not supported");
                    return;
                }
                Ok((socket, Some(addr))) => (socket, addr.ip()),
@@ -130,7 +129,6 @@ pub async fn task_main(
            let startup = Box::pin(
                handle_client(
                    config,
-                    auth_backend,
                    &ctx,
                    cancellation_handler,
                    socket,
@@ -146,7 +144,7 @@ pub async fn task_main(
                Err(e) => {
                    // todo: log and push to ctx the error kind
                    ctx.set_error_kind(e.get_error_kind());
-                    warn!(parent: &span, "per-client task finished with an error: {e:#}");
+                    error!(parent: &span, "per-client task finished with an error: {e:#}");
                }
                Ok(None) => {
                    ctx.set_success();
@@ -157,7 +155,7 @@ pub async fn task_main(
                    match p.proxy_pass().instrument(span.clone()).await {
                        Ok(()) => {}
                        Err(ErrorSource::Client(e)) => {
-                            warn!(parent: &span, "per-client task finished with an IO error from the client: {e:#}");
+                            error!(parent: &span, "per-client task finished with an IO error from the client: {e:#}");
                        }
                        Err(ErrorSource::Compute(e)) => {
                            error!(parent: &span, "per-client task finished with an IO error from the compute: {e:#}");
@@ -245,10 +243,8 @@ impl ReportableError for ClientRequestError {
    }
 }

-#[allow(clippy::too_many_arguments)]
 pub(crate) async fn handle_client<S: AsyncRead + AsyncWrite + Unpin>(
    config: &'static ProxyConfig,
-    auth_backend: &'static auth::Backend<'static, ()>,
    ctx: &RequestMonitoring,
    cancellation_handler: Arc<CancellationHandlerMain>,
    stream: S,
@@ -289,7 +285,8 @@ pub(crate) async fn handle_client<S: AsyncRead + AsyncWrite + Unpin>(
    let common_names = tls.map(|tls| &tls.common_names);

    // Extract credentials which we're going to use for auth.
-    let result = auth_backend
+    let result = config
+        .auth_backend
        .as_ref()
        .map(|()| auth::ComputeUserInfoMaybeEndpoint::parse(ctx, &params, hostname, common_names))
        .transpose();
@@ -356,7 +353,7 @@ pub(crate) async fn handle_client<S: AsyncRead + AsyncWrite + Unpin>(

 /// Finish client connection initialization: confirm auth success, send params, etc.
 #[tracing::instrument(skip_all)]
-pub(crate) async fn prepare_client_connection<P>(
+async fn prepare_client_connection<P>(
    node: &compute::PostgresConnection,
    session: &cancellation::Session<P>,
    stream: &mut PqStream<impl AsyncRead + AsyncWrite + Unpin>,
--- a/proxy/src/proxy/passthrough.rs
+++ b/proxy/src/proxy/passthrough.rs
@@ -71,7 +71,7 @@ impl<P, S: AsyncRead + AsyncWrite + Unpin> ProxyPassthrough<P, S> {
    pub(crate) async fn proxy_pass(self) -> Result<(), ErrorSource> {
        let res = proxy_pass(self.client, self.compute.stream, self.aux).await;
        if let Err(err) = self.compute.cancel_closure.try_cancel_query().await {
-            tracing::warn!(?err, "could not cancel the query in the database");
+            tracing::error!(?err, "could not cancel the query in the database");
        }
        res
    }
--- a/proxy/src/proxy/tests/mod.rs
+++ b/proxy/src/proxy/tests/mod.rs
@@ -492,32 +492,30 @@ impl TestBackend for TestConnectMechanism {
        match action {
            ConnectAction::Wake => Ok(helper_create_cached_node_info(self.cache)),
            ConnectAction::WakeFail => {
-                let err =
-                    control_plane::errors::ApiError::ControlPlane(Box::new(ControlPlaneError {
-                        http_status_code: StatusCode::BAD_REQUEST,
-                        error: "TEST".into(),
-                        status: None,
-                    }));
+                let err = control_plane::errors::ApiError::ControlPlane(ControlPlaneError {
+                    http_status_code: StatusCode::BAD_REQUEST,
+                    error: "TEST".into(),
+                    status: None,
+                });
                assert!(!err.could_retry());
                Err(control_plane::errors::WakeComputeError::ApiError(err))
            }
            ConnectAction::WakeRetry => {
-                let err =
-                    control_plane::errors::ApiError::ControlPlane(Box::new(ControlPlaneError {
-                        http_status_code: StatusCode::BAD_REQUEST,
-                        error: "TEST".into(),
-                        status: Some(Status {
-                            code: "error".into(),
-                            message: "error".into(),
-                            details: Details {
-                                error_info: None,
-                                retry_info: Some(control_plane::messages::RetryInfo {
-                                    retry_delay_ms: 1,
-                                }),
-                                user_facing_message: None,
-                            },
-                        }),
-                    }));
+                let err = control_plane::errors::ApiError::ControlPlane(ControlPlaneError {
+                    http_status_code: StatusCode::BAD_REQUEST,
+                    error: "TEST".into(),
+                    status: Some(Status {
+                        code: "error".into(),
+                        message: "error".into(),
+                        details: Details {
+                            error_info: None,
+                            retry_info: Some(control_plane::messages::RetryInfo {
+                                retry_delay_ms: 1,
+                            }),
+                            user_facing_message: None,
+                        },
+                    }),
+                });
                assert!(err.could_retry());
                Err(control_plane::errors::WakeComputeError::ApiError(err))
            }
@@ -554,7 +552,7 @@ fn helper_create_cached_node_info(cache: &'static NodeInfoCache) -> CachedNodeIn

 fn helper_create_connect_info(
    mechanism: &TestConnectMechanism,
-) -> auth::Backend<'static, ComputeCredentials> {
+) -> auth::Backend<'static, ComputeCredentials, &()> {
    let user_info = auth::Backend::ControlPlane(
        MaybeOwned::Owned(ControlPlaneBackend::Test(Box::new(mechanism.clone()))),
        ComputeCredentials {
--- a/proxy/src/proxy/wake_compute.rs
+++ b/proxy/src/proxy/wake_compute.rs
@@ -79,7 +79,7 @@ fn report_error(e: &WakeComputeError, retry: bool) {
            Reason::ConcurrencyLimitReached => WakeupFailureKind::ApiConsoleLocked,
            Reason::LockAlreadyTaken => WakeupFailureKind::ApiConsoleLocked,
            Reason::RunningOperations => WakeupFailureKind::ApiConsoleLocked,
-            Reason::Unknown => match **e {
+            Reason::Unknown => match e {
                ControlPlaneError {
                    http_status_code: StatusCode::LOCKED,
                    ref error,
--- a/proxy/src/redis/connection_with_credentials_provider.rs
+++ b/proxy/src/redis/connection_with_credentials_provider.rs
@@ -6,7 +6,7 @@ use redis::{
    ConnectionInfo, IntoConnectionInfo, RedisConnectionInfo, RedisResult,
 };
 use tokio::task::JoinHandle;
-use tracing::{debug, error, info, warn};
+use tracing::{debug, error, info};

 use super::elasticache::CredentialsProvider;

@@ -89,7 +89,7 @@ impl ConnectionWithCredentialsProvider {
                    return Ok(());
                }
                Err(e) => {
-                    warn!("Error during PING: {e:?}");
+                    error!("Error during PING: {e:?}");
                }
            }
        } else {
@@ -121,7 +121,7 @@ impl ConnectionWithCredentialsProvider {
                info!("Connection succesfully established");
            }
            Err(e) => {
-                warn!("Connection is broken. Error during PING: {e:?}");
+                error!("Connection is broken. Error during PING: {e:?}");
            }
        }
        self.con = Some(con);
--- a/proxy/src/redis/notifications.rs
+++ b/proxy/src/redis/notifications.rs
@@ -146,7 +146,7 @@ impl<C: ProjectInfoCache + Send + Sync + 'static> MessageHandler<C> {
                {
                    Ok(()) => {}
                    Err(e) => {
-                        tracing::warn!("failed to cancel session: {e}");
+                        tracing::error!("failed to cancel session: {e}");
                    }
                }
            }
--- a/proxy/src/serverless/backend.rs
+++ b/proxy/src/serverless/backend.rs
@@ -2,19 +2,16 @@ use std::{io, sync::Arc, time::Duration};

 use async_trait::async_trait;
 use hyper_util::rt::{TokioExecutor, TokioIo, TokioTimer};
-use p256::{ecdsa::SigningKey, elliptic_curve::JwkEcKey};
-use rand::rngs::OsRng;
 use tokio::net::{lookup_host, TcpStream};
-use tracing::{debug, field::display, info};
+use tracing::{field::display, info};

 use crate::{
    auth::{
-        self,
        backend::{local::StaticAuthRules, ComputeCredentials, ComputeUserInfo},
        check_peer_addr_is_in_list, AuthError,
    },
    compute,
-    config::ProxyConfig,
+    config::{AuthenticationConfig, ProxyConfig},
    context::RequestMonitoring,
    control_plane::{
        errors::{GetAuthInfoError, WakeComputeError},
@@ -29,21 +26,18 @@ use crate::{
        retry::{CouldRetry, ShouldRetryWakeCompute},
    },
    rate_limiter::EndpointRateLimiter,
-    EndpointId, Host,
+    Host,
 };

 use super::{
    conn_pool::{poll_client, Client, ConnInfo, GlobalConnPool},
    http_conn_pool::{self, poll_http2_client},
-    local_conn_pool::{self, LocalClient, LocalConnPool},
 };

 pub(crate) struct PoolingBackend {
    pub(crate) http_conn_pool: Arc<super::http_conn_pool::GlobalConnPool>,
-    pub(crate) local_pool: Arc<LocalConnPool<tokio_postgres::Client>>,
    pub(crate) pool: Arc<GlobalConnPool<tokio_postgres::Client>>,
    pub(crate) config: &'static ProxyConfig,
-    pub(crate) auth_backend: &'static crate::auth::Backend<'static, ()>,
    pub(crate) endpoint_rate_limiter: Arc<EndpointRateLimiter>,
 }

@@ -51,13 +45,18 @@ impl PoolingBackend {
    pub(crate) async fn authenticate_with_password(
        &self,
        ctx: &RequestMonitoring,
+        config: &AuthenticationConfig,
        user_info: &ComputeUserInfo,
        password: &[u8],
    ) -> Result<ComputeCredentials, AuthError> {
        let user_info = user_info.clone();
-        let backend = self.auth_backend.as_ref().map(|()| user_info.clone());
+        let backend = self
+            .config
+            .auth_backend
+            .as_ref()
+            .map(|()| user_info.clone());
        let (allowed_ips, maybe_secret) = backend.get_allowed_ips_and_secret(ctx).await?;
-        if self.config.authentication_config.ip_allowlist_check_enabled
+        if config.ip_allowlist_check_enabled
            && !check_peer_addr_is_in_list(&ctx.peer_addr(), &allowed_ips)
        {
            return Err(AuthError::ip_address_not_allowed(ctx.peer_addr()));
@@ -76,6 +75,7 @@ impl PoolingBackend {
        let secret = match cached_secret.value.clone() {
            Some(secret) => self.config.authentication_config.check_rate_limit(
                ctx,
+                config,
                secret,
                &user_info.endpoint,
                true,
@@ -87,13 +87,9 @@ impl PoolingBackend {
            }
        };
        let ep = EndpointIdInt::from(&user_info.endpoint);
-        let auth_outcome = crate::auth::validate_password_and_exchange(
-            &self.config.authentication_config.thread_pool,
-            ep,
-            password,
-            secret,
-        )
-        .await?;
+        let auth_outcome =
+            crate::auth::validate_password_and_exchange(&config.thread_pool, ep, password, secret)
+                .await?;
        let res = match auth_outcome {
            crate::sasl::Outcome::Success(key) => {
                info!("user successfully authenticated");
@@ -113,13 +109,13 @@ impl PoolingBackend {
    pub(crate) async fn authenticate_with_jwt(
        &self,
        ctx: &RequestMonitoring,
+        config: &AuthenticationConfig,
        user_info: &ComputeUserInfo,
        jwt: String,
-    ) -> Result<ComputeCredentials, AuthError> {
-        match &self.auth_backend {
+    ) -> Result<(), AuthError> {
+        match &self.config.auth_backend {
            crate::auth::Backend::ControlPlane(console, ()) => {
-                self.config
-                    .authentication_config
+                config
                    .jwks_cache
                    .check_jwt(
                        ctx,
@@ -131,15 +127,13 @@ impl PoolingBackend {
                    .await
                    .map_err(|e| AuthError::auth_failed(e.to_string()))?;

-                Ok(ComputeCredentials {
-                    info: user_info.clone(),
-                    keys: crate::auth::backend::ComputeCredentialKeys::None,
-                })
+                Ok(())
            }
+            crate::auth::Backend::ConsoleRedirect(_, ()) => Err(AuthError::auth_failed(
+                "JWT login over web auth proxy is not supported",
+            )),
            crate::auth::Backend::Local(_) => {
-                let keys = self
-                    .config
-                    .authentication_config
+                config
                    .jwks_cache
                    .check_jwt(
                        ctx,
@@ -151,10 +145,8 @@ impl PoolingBackend {
                    .await
                    .map_err(|e| AuthError::auth_failed(e.to_string()))?;

-                Ok(ComputeCredentials {
-                    info: user_info.clone(),
-                    keys,
-                })
+                // todo: rewrite JWT signature with key shared somehow between local proxy and postgres
+                Ok(())
            }
        }
    }
@@ -184,7 +176,7 @@ impl PoolingBackend {
        let conn_id = uuid::Uuid::new_v4();
        tracing::Span::current().record("conn_id", display(conn_id));
        info!(%conn_id, "pool: opening a new connection '{conn_info}'");
-        let backend = self.auth_backend.as_ref().map(|()| keys);
+        let backend = self.config.auth_backend.as_ref().map(|()| keys);
        crate::proxy::connect_compute::connect_to_compute(
            ctx,
            &TokioMechanism {
@@ -216,14 +208,14 @@ impl PoolingBackend {
        let conn_id = uuid::Uuid::new_v4();
        tracing::Span::current().record("conn_id", display(conn_id));
        info!(%conn_id, "pool: opening a new connection '{conn_info}'");
-        let backend = self.auth_backend.as_ref().map(|()| ComputeCredentials {
-            info: ComputeUserInfo {
-                user: conn_info.user_info.user.clone(),
-                endpoint: EndpointId::from(format!("{}-local-proxy", conn_info.user_info.endpoint)),
-                options: conn_info.user_info.options.clone(),
-            },
-            keys: crate::auth::backend::ComputeCredentialKeys::None,
-        });
+        let backend = self
+            .config
+            .auth_backend
+            .as_ref()
+            .map(|()| ComputeCredentials {
+                info: conn_info.user_info.clone(),
+                keys: crate::auth::backend::ComputeCredentialKeys::None,
+            });
        crate::proxy::connect_compute::connect_to_compute(
            ctx,
            &HyperMechanism {
@@ -239,85 +231,6 @@ impl PoolingBackend {
        )
        .await
    }
-
-    /// Connect to postgres over localhost.
-    ///
-    /// We expect postgres to be started here, so we won't do any retries.
-    ///
-    /// # Panics
-    ///
-    /// Panics if called with a non-local_proxy backend.
-    #[tracing::instrument(fields(pid = tracing::field::Empty), skip_all)]
-    pub(crate) async fn connect_to_local_postgres(
-        &self,
-        ctx: &RequestMonitoring,
-        conn_info: ConnInfo,
-    ) -> Result<LocalClient<tokio_postgres::Client>, HttpConnError> {
-        if let Some(client) = self.local_pool.get(ctx, &conn_info)? {
-            return Ok(client);
-        }
-
-        let conn_id = uuid::Uuid::new_v4();
-        tracing::Span::current().record("conn_id", display(conn_id));
-        info!(%conn_id, "local_pool: opening a new connection '{conn_info}'");
-
-        let mut node_info = match &self.auth_backend {
-            auth::Backend::ControlPlane(_, ()) => {
-                unreachable!("only local_proxy can connect to local postgres")
-            }
-            auth::Backend::Local(local) => local.node_info.clone(),
-        };
-
-        let (key, jwk) = create_random_jwk();
-
-        let config = node_info
-            .config
-            .user(&conn_info.user_info.user)
-            .dbname(&conn_info.dbname)
-            .options(&format!(
-                "-c pg_session_jwt.jwk={}",
-                serde_json::to_string(&jwk).expect("serializing jwk to json should not fail")
-            ));
-
-        let pause = ctx.latency_timer_pause(crate::metrics::Waiting::Compute);
-        let (client, connection) = config.connect(tokio_postgres::NoTls).await?;
-        drop(pause);
-
-        let pid = client.get_process_id();
-        tracing::Span::current().record("pid", pid);
-
-        let mut handle = local_conn_pool::poll_client(
-            self.local_pool.clone(),
-            ctx,
-            conn_info,
-            client,
-            connection,
-            key,
-            conn_id,
-            node_info.aux.clone(),
-        );
-
-        {
-            let (client, mut discard) = handle.inner();
-            debug!("setting up backend session state");
-
-            // initiates the auth session
-            if let Err(e) = client.query("select auth.init()", &[]).await {
-                discard.discard();
-                return Err(e.into());
-            }
-
-            info!("backend session state initialized");
-        }
-
-        Ok(handle)
-    }
-}
-
-fn create_random_jwk() -> (SigningKey, JwkEcKey) {
-    let key = SigningKey::random(&mut OsRng);
-    let jwk = p256::PublicKey::from(key.verifying_key()).to_jwk();
-    (key, jwk)
 }

 #[derive(Debug, thiserror::Error)]
@@ -328,8 +241,6 @@ pub(crate) enum HttpConnError {
    PostgresConnectionError(#[from] tokio_postgres::Error),
    #[error("could not connection to local-proxy in compute")]
    LocalProxyConnectionError(#[from] LocalProxyConnError),
-    #[error("could not parse JWT payload")]
-    JwtPayloadError(serde_json::Error),

    #[error("could not get auth info")]
    GetAuthInfo(#[from] GetAuthInfoError),
@@ -355,7 +266,6 @@ impl ReportableError for HttpConnError {
            HttpConnError::ConnectionClosedAbruptly(_) => ErrorKind::Compute,
            HttpConnError::PostgresConnectionError(p) => p.get_error_kind(),
            HttpConnError::LocalProxyConnectionError(_) => ErrorKind::Compute,
-            HttpConnError::JwtPayloadError(_) => ErrorKind::User,
            HttpConnError::GetAuthInfo(a) => a.get_error_kind(),
            HttpConnError::AuthError(a) => a.get_error_kind(),
            HttpConnError::WakeCompute(w) => w.get_error_kind(),
@@ -370,7 +280,6 @@ impl UserFacingError for HttpConnError {
            HttpConnError::ConnectionClosedAbruptly(_) => self.to_string(),
            HttpConnError::PostgresConnectionError(p) => p.to_string(),
            HttpConnError::LocalProxyConnectionError(p) => p.to_string(),
-            HttpConnError::JwtPayloadError(p) => p.to_string(),
            HttpConnError::GetAuthInfo(c) => c.to_string_client(),
            HttpConnError::AuthError(c) => c.to_string_client(),
            HttpConnError::WakeCompute(c) => c.to_string_client(),
@@ -387,7 +296,6 @@ impl CouldRetry for HttpConnError {
            HttpConnError::PostgresConnectionError(e) => e.could_retry(),
            HttpConnError::LocalProxyConnectionError(e) => e.could_retry(),
            HttpConnError::ConnectionClosedAbruptly(_) => false,
-            HttpConnError::JwtPayloadError(_) => false,
            HttpConnError::GetAuthInfo(_) => false,
            HttpConnError::AuthError(_) => false,
            HttpConnError::WakeCompute(_) => false,
@@ -514,12 +422,8 @@ impl ConnectMechanism for HyperMechanism {

        let pause = ctx.latency_timer_pause(crate::metrics::Waiting::Compute);

-        let port = *node_info.config.get_ports().first().ok_or_else(|| {
-            HttpConnError::WakeCompute(WakeComputeError::BadComputeAddress(
-                "local-proxy port missing on compute address".into(),
-            ))
-        })?;
-        let res = connect_http2(&host, port, timeout).await;
+        // let port = node_info.config.get_ports().first().unwrap_or_else(10432);
+        let res = connect_http2(&host, 10432, timeout).await;
        drop(pause);
        let (client, connection) = permit.release_result(res)?;

--- a/proxy/src/serverless/http_util.rs
+++ b/proxy/src/serverless/http_util.rs
@@ -41,10 +41,6 @@ pub(crate) fn api_error_into_response(this: ApiError) -> Response<BoxBody<Bytes,
            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/proxy/src/serverless/local_conn_pool.rs
+++ b/proxy/src/serverless/local_conn_pool.rs
@@ -1,593 +0,0 @@
-use futures::{future::poll_fn, Future};
-use indexmap::IndexMap;
-use jose_jwk::jose_b64::base64ct::{Base64UrlUnpadded, Encoding};
-use p256::ecdsa::{Signature, SigningKey};
-use parking_lot::RwLock;
-use serde_json::value::RawValue;
-use signature::Signer;
-use std::task::{ready, Poll};
-use std::{collections::HashMap, pin::pin, sync::Arc, sync::Weak, time::Duration};
-use tokio::time::Instant;
-use tokio_postgres::tls::NoTlsStream;
-use tokio_postgres::types::ToSql;
-use tokio_postgres::{AsyncMessage, ReadyForQueryStatus, Socket};
-use tokio_util::sync::CancellationToken;
-
-use crate::control_plane::messages::{ColdStartInfo, MetricsAuxInfo};
-use crate::metrics::Metrics;
-use crate::usage_metrics::{Ids, MetricCounter, USAGE_METRICS};
-use crate::{context::RequestMonitoring, DbName, RoleName};
-
-use tracing::{error, warn, Span};
-use tracing::{info, info_span, Instrument};
-
-use super::backend::HttpConnError;
-use super::conn_pool::{ClientInnerExt, ConnInfo};
-
-struct ConnPoolEntry<C: ClientInnerExt> {
-    conn: ClientInner<C>,
-    _last_access: std::time::Instant,
-}
-
-// /// key id for the pg_session_jwt state
-// static PG_SESSION_JWT_KID: AtomicU64 = AtomicU64::new(1);
-
-// Per-endpoint connection pool, (dbname, username) -> DbUserConnPool
-// Number of open connections is limited by the `max_conns_per_endpoint`.
-pub(crate) struct EndpointConnPool<C: ClientInnerExt> {
-    pools: HashMap<(DbName, RoleName), DbUserConnPool<C>>,
-    total_conns: usize,
-    max_conns: usize,
-    global_pool_size_max_conns: usize,
-}
-
-impl<C: ClientInnerExt> EndpointConnPool<C> {
-    fn get_conn_entry(&mut self, db_user: (DbName, RoleName)) -> Option<ConnPoolEntry<C>> {
-        let Self {
-            pools, total_conns, ..
-        } = self;
-        pools
-            .get_mut(&db_user)
-            .and_then(|pool_entries| pool_entries.get_conn_entry(total_conns))
-    }
-
-    fn remove_client(&mut self, db_user: (DbName, RoleName), conn_id: uuid::Uuid) -> bool {
-        let Self {
-            pools, total_conns, ..
-        } = self;
-        if let Some(pool) = pools.get_mut(&db_user) {
-            let old_len = pool.conns.len();
-            pool.conns.retain(|conn| conn.conn.conn_id != conn_id);
-            let new_len = pool.conns.len();
-            let removed = old_len - new_len;
-            if removed > 0 {
-                Metrics::get()
-                    .proxy
-                    .http_pool_opened_connections
-                    .get_metric()
-                    .dec_by(removed as i64);
-            }
-            *total_conns -= removed;
-            removed > 0
-        } else {
-            false
-        }
-    }
-
-    fn put(pool: &RwLock<Self>, conn_info: &ConnInfo, client: ClientInner<C>) {
-        let conn_id = client.conn_id;
-
-        if client.is_closed() {
-            info!(%conn_id, "local_pool: throwing away connection '{conn_info}' because connection is closed");
-            return;
-        }
-        let global_max_conn = pool.read().global_pool_size_max_conns;
-        if pool.read().total_conns >= global_max_conn {
-            info!(%conn_id, "local_pool: throwing away connection '{conn_info}' because pool is full");
-            return;
-        }
-
-        // return connection to the pool
-        let mut returned = false;
-        let mut per_db_size = 0;
-        let total_conns = {
-            let mut pool = pool.write();
-
-            if pool.total_conns < pool.max_conns {
-                let pool_entries = pool.pools.entry(conn_info.db_and_user()).or_default();
-                pool_entries.conns.push(ConnPoolEntry {
-                    conn: client,
-                    _last_access: std::time::Instant::now(),
-                });
-
-                returned = true;
-                per_db_size = pool_entries.conns.len();
-
-                pool.total_conns += 1;
-                Metrics::get()
-                    .proxy
-                    .http_pool_opened_connections
-                    .get_metric()
-                    .inc();
-            }
-
-            pool.total_conns
-        };
-
-        // do logging outside of the mutex
-        if returned {
-            info!(%conn_id, "local_pool: returning connection '{conn_info}' back to the pool, total_conns={total_conns}, for this (db, user)={per_db_size}");
-        } else {
-            info!(%conn_id, "local_pool: throwing away connection '{conn_info}' because pool is full, total_conns={total_conns}");
-        }
-    }
-}
-
-impl<C: ClientInnerExt> Drop for EndpointConnPool<C> {
-    fn drop(&mut self) {
-        if self.total_conns > 0 {
-            Metrics::get()
-                .proxy
-                .http_pool_opened_connections
-                .get_metric()
-                .dec_by(self.total_conns as i64);
-        }
-    }
-}
-
-pub(crate) struct DbUserConnPool<C: ClientInnerExt> {
-    conns: Vec<ConnPoolEntry<C>>,
-}
-
-impl<C: ClientInnerExt> Default for DbUserConnPool<C> {
-    fn default() -> Self {
-        Self { conns: Vec::new() }
-    }
-}
-
-impl<C: ClientInnerExt> DbUserConnPool<C> {
-    fn clear_closed_clients(&mut self, conns: &mut usize) -> usize {
-        let old_len = self.conns.len();
-
-        self.conns.retain(|conn| !conn.conn.is_closed());
-
-        let new_len = self.conns.len();
-        let removed = old_len - new_len;
-        *conns -= removed;
-        removed
-    }
-
-    fn get_conn_entry(&mut self, conns: &mut usize) -> Option<ConnPoolEntry<C>> {
-        let mut removed = self.clear_closed_clients(conns);
-        let conn = self.conns.pop();
-        if conn.is_some() {
-            *conns -= 1;
-            removed += 1;
-        }
-        Metrics::get()
-            .proxy
-            .http_pool_opened_connections
-            .get_metric()
-            .dec_by(removed as i64);
-        conn
-    }
-}
-
-pub(crate) struct LocalConnPool<C: ClientInnerExt> {
-    global_pool: RwLock<EndpointConnPool<C>>,
-
-    config: &'static crate::config::HttpConfig,
-}
-
-impl<C: ClientInnerExt> LocalConnPool<C> {
-    pub(crate) fn new(config: &'static crate::config::HttpConfig) -> Arc<Self> {
-        Arc::new(Self {
-            global_pool: RwLock::new(EndpointConnPool {
-                pools: HashMap::new(),
-                total_conns: 0,
-                max_conns: config.pool_options.max_conns_per_endpoint,
-                global_pool_size_max_conns: config.pool_options.max_total_conns,
-            }),
-            config,
-        })
-    }
-
-    pub(crate) fn get_idle_timeout(&self) -> Duration {
-        self.config.pool_options.idle_timeout
-    }
-
-    // pub(crate) fn shutdown(&self) {
-    //     let mut pool = self.global_pool.write();
-    //     pool.pools.clear();
-    //     pool.total_conns = 0;
-    // }
-
-    pub(crate) fn get(
-        self: &Arc<Self>,
-        ctx: &RequestMonitoring,
-        conn_info: &ConnInfo,
-    ) -> Result<Option<LocalClient<C>>, HttpConnError> {
-        let mut client: Option<ClientInner<C>> = None;
-        if let Some(entry) = self
-            .global_pool
-            .write()
-            .get_conn_entry(conn_info.db_and_user())
-        {
-            client = Some(entry.conn);
-        }
-
-        // ok return cached connection if found and establish a new one otherwise
-        if let Some(client) = client {
-            if client.is_closed() {
-                info!("local_pool: cached connection '{conn_info}' is closed, opening a new one");
-                return Ok(None);
-            }
-            tracing::Span::current().record("conn_id", tracing::field::display(client.conn_id));
-            tracing::Span::current().record(
-                "pid",
-                tracing::field::display(client.inner.get_process_id()),
-            );
-            info!(
-                cold_start_info = ColdStartInfo::HttpPoolHit.as_str(),
-                "local_pool: reusing connection '{conn_info}'"
-            );
-            client.session.send(ctx.session_id())?;
-            ctx.set_cold_start_info(ColdStartInfo::HttpPoolHit);
-            ctx.success();
-            return Ok(Some(LocalClient::new(
-                client,
-                conn_info.clone(),
-                Arc::downgrade(self),
-            )));
-        }
-        Ok(None)
-    }
-}
-
-#[allow(clippy::too_many_arguments)]
-pub(crate) fn poll_client(
-    global_pool: Arc<LocalConnPool<tokio_postgres::Client>>,
-    ctx: &RequestMonitoring,
-    conn_info: ConnInfo,
-    client: tokio_postgres::Client,
-    mut connection: tokio_postgres::Connection<Socket, NoTlsStream>,
-    key: SigningKey,
-    conn_id: uuid::Uuid,
-    aux: MetricsAuxInfo,
-) -> LocalClient<tokio_postgres::Client> {
-    let conn_gauge = Metrics::get().proxy.db_connections.guard(ctx.protocol());
-    let mut session_id = ctx.session_id();
-    let (tx, mut rx) = tokio::sync::watch::channel(session_id);
-
-    let span = info_span!(parent: None, "connection", %conn_id);
-    let cold_start_info = ctx.cold_start_info();
-    span.in_scope(|| {
-        info!(cold_start_info = cold_start_info.as_str(), %conn_info, %session_id, "new connection");
-    });
-    let pool = Arc::downgrade(&global_pool);
-    let pool_clone = pool.clone();
-
-    let db_user = conn_info.db_and_user();
-    let idle = global_pool.get_idle_timeout();
-    let cancel = CancellationToken::new();
-    let cancelled = cancel.clone().cancelled_owned();
-
-    tokio::spawn(
-    async move {
-        let _conn_gauge = conn_gauge;
-        let mut idle_timeout = pin!(tokio::time::sleep(idle));
-        let mut cancelled = pin!(cancelled);
-
-        poll_fn(move |cx| {
-            if cancelled.as_mut().poll(cx).is_ready() {
-                info!("connection dropped");
-                return Poll::Ready(())
-            }
-
-            match rx.has_changed() {
-                Ok(true) => {
-                    session_id = *rx.borrow_and_update();
-                    info!(%session_id, "changed session");
-                    idle_timeout.as_mut().reset(Instant::now() + idle);
-                }
-                Err(_) => {
-                    info!("connection dropped");
-                    return Poll::Ready(())
-                }
-                _ => {}
-            }
-
-            // 5 minute idle connection timeout
-            if idle_timeout.as_mut().poll(cx).is_ready() {
-                idle_timeout.as_mut().reset(Instant::now() + idle);
-                info!("connection idle");
-                if let Some(pool) = pool.clone().upgrade() {
-                    // remove client from pool - should close the connection if it's idle.
-                    // does nothing if the client is currently checked-out and in-use
-                    if pool.global_pool.write().remove_client(db_user.clone(), conn_id) {
-                        info!("idle connection removed");
-                    }
-                }
-            }
-
-            loop {
-                let message = ready!(connection.poll_message(cx));
-
-                match message {
-                    Some(Ok(AsyncMessage::Notice(notice))) => {
-                        info!(%session_id, "notice: {}", notice);
-                    }
-                    Some(Ok(AsyncMessage::Notification(notif))) => {
-                        warn!(%session_id, pid = notif.process_id(), channel = notif.channel(), "notification received");
-                    }
-                    Some(Ok(_)) => {
-                        warn!(%session_id, "unknown message");
-                    }
-                    Some(Err(e)) => {
-                        error!(%session_id, "connection error: {}", e);
-                        break
-                    }
-                    None => {
-                        info!("connection closed");
-                        break
-                    }
-                }
-            }
-
-            // remove from connection pool
-            if let Some(pool) = pool.clone().upgrade() {
-                if pool.global_pool.write().remove_client(db_user.clone(), conn_id) {
-                    info!("closed connection removed");
-                }
-            }
-
-            Poll::Ready(())
-        }).await;
-
-    }
-    .instrument(span));
-
-    let inner = ClientInner {
-        inner: client,
-        session: tx,
-        cancel,
-        aux,
-        conn_id,
-        key,
-        jti: 0,
-    };
-    LocalClient::new(inner, conn_info, pool_clone)
-}
-
-struct ClientInner<C: ClientInnerExt> {
-    inner: C,
-    session: tokio::sync::watch::Sender<uuid::Uuid>,
-    cancel: CancellationToken,
-    aux: MetricsAuxInfo,
-    conn_id: uuid::Uuid,
-
-    // needed for pg_session_jwt state
-    key: SigningKey,
-    jti: u64,
-}
-
-impl<C: ClientInnerExt> Drop for ClientInner<C> {
-    fn drop(&mut self) {
-        // on client drop, tell the conn to shut down
-        self.cancel.cancel();
-    }
-}
-
-impl<C: ClientInnerExt> ClientInner<C> {
-    pub(crate) fn is_closed(&self) -> bool {
-        self.inner.is_closed()
-    }
-}
-
-impl<C: ClientInnerExt> LocalClient<C> {
-    pub(crate) fn metrics(&self) -> Arc<MetricCounter> {
-        let aux = &self.inner.as_ref().unwrap().aux;
-        USAGE_METRICS.register(Ids {
-            endpoint_id: aux.endpoint_id,
-            branch_id: aux.branch_id,
-        })
-    }
-}
-
-pub(crate) struct LocalClient<C: ClientInnerExt> {
-    span: Span,
-    inner: Option<ClientInner<C>>,
-    conn_info: ConnInfo,
-    pool: Weak<LocalConnPool<C>>,
-}
-
-pub(crate) struct Discard<'a, C: ClientInnerExt> {
-    conn_info: &'a ConnInfo,
-    pool: &'a mut Weak<LocalConnPool<C>>,
-}
-
-impl<C: ClientInnerExt> LocalClient<C> {
-    pub(self) fn new(
-        inner: ClientInner<C>,
-        conn_info: ConnInfo,
-        pool: Weak<LocalConnPool<C>>,
-    ) -> Self {
-        Self {
-            inner: Some(inner),
-            span: Span::current(),
-            conn_info,
-            pool,
-        }
-    }
-    pub(crate) fn inner(&mut self) -> (&mut C, Discard<'_, C>) {
-        let Self {
-            inner,
-            pool,
-            conn_info,
-            span: _,
-        } = self;
-        let inner = inner.as_mut().expect("client inner should not be removed");
-        (&mut inner.inner, Discard { conn_info, pool })
-    }
-}
-
-impl LocalClient<tokio_postgres::Client> {
-    pub(crate) async fn set_jwt_session(&mut self, payload: &[u8]) -> Result<(), HttpConnError> {
-        let inner = self
-            .inner
-            .as_mut()
-            .expect("client inner should not be removed");
-
-        inner.jti += 1;
-        let token = resign_jwt(&inner.key, payload, inner.jti)?;
-
-        // initiates the auth session
-        inner.inner.simple_query("discard all").await?;
-        inner
-            .inner
-            .query(
-                "select auth.jwt_session_init($1)",
-                &[&token as &(dyn ToSql + Sync)],
-            )
-            .await?;
-
-        let pid = inner.inner.get_process_id();
-        info!(pid, jti = inner.jti, "user session state init");
-
-        Ok(())
-    }
-}
-
-/// implements relatively efficient in-place json object key upserting
-///
-/// only supports top-level keys
-fn upsert_json_object(
-    payload: &[u8],
-    key: &str,
-    value: &RawValue,
-) -> Result<String, serde_json::Error> {
-    let mut payload = serde_json::from_slice::<IndexMap<&str, &RawValue>>(payload)?;
-    payload.insert(key, value);
-    serde_json::to_string(&payload)
-}
-
-fn resign_jwt(sk: &SigningKey, payload: &[u8], jti: u64) -> Result<String, HttpConnError> {
-    let mut buffer = itoa::Buffer::new();
-
-    // encode the jti integer to a json rawvalue
-    let jti = serde_json::from_str::<&RawValue>(buffer.format(jti)).unwrap();
-
-    // update the jti in-place
-    let payload =
-        upsert_json_object(payload, "jti", jti).map_err(HttpConnError::JwtPayloadError)?;
-
-    // sign the jwt
-    let token = sign_jwt(sk, payload.as_bytes());
-
-    Ok(token)
-}
-
-fn sign_jwt(sk: &SigningKey, payload: &[u8]) -> String {
-    let header_len = 20;
-    let payload_len = Base64UrlUnpadded::encoded_len(payload);
-    let signature_len = Base64UrlUnpadded::encoded_len(&[0; 64]);
-    let total_len = header_len + payload_len + signature_len + 2;
-
-    let mut jwt = String::with_capacity(total_len);
-    let cap = jwt.capacity();
-
-    // we only need an empty header with the alg specified.
-    // base64url(r#"{"alg":"ES256"}"#) == "eyJhbGciOiJFUzI1NiJ9"
-    jwt.push_str("eyJhbGciOiJFUzI1NiJ9.");
-
-    // encode the jwt payload in-place
-    base64::encode_config_buf(payload, base64::URL_SAFE_NO_PAD, &mut jwt);
-
-    // create the signature from the encoded header || payload
-    let sig: Signature = sk.sign(jwt.as_bytes());
-
-    jwt.push('.');
-
-    // encode the jwt signature in-place
-    base64::encode_config_buf(sig.to_bytes(), base64::URL_SAFE_NO_PAD, &mut jwt);
-
-    debug_assert_eq!(
-        jwt.len(),
-        total_len,
-        "the jwt len should match our expected len"
-    );
-    debug_assert_eq!(jwt.capacity(), cap, "the jwt capacity should not change");
-
-    jwt
-}
-
-impl<C: ClientInnerExt> Discard<'_, C> {
-    pub(crate) fn check_idle(&mut self, status: ReadyForQueryStatus) {
-        let conn_info = &self.conn_info;
-        if status != ReadyForQueryStatus::Idle && std::mem::take(self.pool).strong_count() > 0 {
-            info!(
-                "local_pool: throwing away connection '{conn_info}' because connection is not idle"
-            );
-        }
-    }
-    pub(crate) fn discard(&mut self) {
-        let conn_info = &self.conn_info;
-        if std::mem::take(self.pool).strong_count() > 0 {
-            info!("local_pool: throwing away connection '{conn_info}' because connection is potentially in a broken state");
-        }
-    }
-}
-
-impl<C: ClientInnerExt> LocalClient<C> {
-    fn do_drop(&mut self) -> Option<impl FnOnce()> {
-        let conn_info = self.conn_info.clone();
-        let client = self
-            .inner
-            .take()
-            .expect("client inner should not be removed");
-        if let Some(conn_pool) = std::mem::take(&mut self.pool).upgrade() {
-            let current_span = self.span.clone();
-            // return connection to the pool
-            return Some(move || {
-                let _span = current_span.enter();
-                EndpointConnPool::put(&conn_pool.global_pool, &conn_info, client);
-            });
-        }
-        None
-    }
-}
-
-impl<C: ClientInnerExt> Drop for LocalClient<C> {
-    fn drop(&mut self) {
-        if let Some(drop) = self.do_drop() {
-            tokio::task::spawn_blocking(drop);
-        }
-    }
-}
-
-#[cfg(test)]
-mod tests {
-    use p256::ecdsa::SigningKey;
-    use typed_json::json;
-
-    use super::resign_jwt;
-
-    #[test]
-    fn jwt_token_snapshot() {
-        let key = SigningKey::from_bytes(&[1; 32].into()).unwrap();
-        let data =
-            json!({"foo":"bar","jti":"foo\nbar","nested":{"jti":"tricky nesting"}}).to_string();
-
-        let jwt = resign_jwt(&key, data.as_bytes(), 2).unwrap();
-
-        // To validate the JWT, copy the JWT string and paste it into https://jwt.io/.
-        // In the public-key box, paste the following jwk public key
-        // `{"kty":"EC","crv":"P-256","x":"b_A7lJJBzh2t1DUZ5pYOCoW0GmmgXDKBA6orzhWUyhY","y":"PE91OlW_AdxT9sCwx-7ni0DG_30lqW4igrmJzvccFEo"}`
-
-        // let pub_key = p256::ecdsa::VerifyingKey::from(&key);
-        // let pub_key = p256::PublicKey::from(pub_key);
-        // println!("{}", pub_key.to_jwk_string());
-
-        assert_eq!(jwt, "eyJhbGciOiJFUzI1NiJ9.eyJmb28iOiJiYXIiLCJqdGkiOjIsIm5lc3RlZCI6eyJqdGkiOiJ0cmlja3kgbmVzdGluZyJ9fQ.pYf0LxoJ8sDgpmsYOgrbNecOSipnPBEGwnZzB-JhW2cONrKlqRsgXwK8_cOsyolGy-hTTe8GXbWTl_UdpF5RyA");
-    }
-}
--- a/proxy/src/serverless/mod.rs
+++ b/proxy/src/serverless/mod.rs
@@ -8,7 +8,6 @@ mod conn_pool;
 mod http_conn_pool;
 mod http_util;
 mod json;
-mod local_conn_pool;
 mod sql_over_http;
 mod websocket;

@@ -48,14 +47,13 @@ use std::pin::{pin, Pin};
 use std::sync::Arc;
 use tokio::net::{TcpListener, TcpStream};
 use tokio_util::sync::CancellationToken;
-use tracing::{info, warn, Instrument};
+use tracing::{error, info, warn, Instrument};
 use utils::http::error::ApiError;

 pub(crate) const SERVERLESS_DRIVER_SNI: &str = "api";

 pub async fn task_main(
    config: &'static ProxyConfig,
-    auth_backend: &'static crate::auth::Backend<'static, ()>,
    ws_listener: TcpListener,
    cancellation_token: CancellationToken,
    cancellation_handler: Arc<CancellationHandlerMain>,
@@ -65,7 +63,6 @@ pub async fn task_main(
        info!("websocket server has shut down");
    }

-    let local_pool = local_conn_pool::LocalConnPool::new(&config.http_config);
    let conn_pool = conn_pool::GlobalConnPool::new(&config.http_config);
    {
        let conn_pool = Arc::clone(&conn_pool);
@@ -108,10 +105,8 @@ pub async fn task_main(

    let backend = Arc::new(PoolingBackend {
        http_conn_pool: Arc::clone(&http_conn_pool),
-        local_pool,
        pool: Arc::clone(&conn_pool),
        config,
-        auth_backend,
        endpoint_rate_limiter: Arc::clone(&endpoint_rate_limiter),
    });
    let tls_acceptor: Arc<dyn MaybeTlsAcceptor> = match config.tls_config.as_ref() {
@@ -243,7 +238,7 @@ async fn connection_startup(
    let (conn, peer) = match read_proxy_protocol(conn).await {
        Ok(c) => c,
        Err(e) => {
-            tracing::warn!(?session_id, %peer_addr, "failed to accept TCP connection: invalid PROXY protocol V2 header: {e:#}");
+            tracing::error!(?session_id, %peer_addr, "failed to accept TCP connection: invalid PROXY protocol V2 header: {e:#}");
            return None;
        }
    };
@@ -399,7 +394,6 @@ async fn request_handler(
            async move {
                if let Err(e) = websocket::serve_websocket(
                    config,
-                    backend.auth_backend,
                    ctx,
                    websocket,
                    cancellation_handler,
@@ -408,7 +402,7 @@ async fn request_handler(
                )
                .await
                {
-                    warn!("error in websocket connection: {e:#}");
+                    error!("error in websocket connection: {e:#}");
                }
            }
            .instrument(span),
--- a/proxy/src/serverless/sql_over_http.rs
+++ b/proxy/src/serverless/sql_over_http.rs
@@ -40,12 +40,11 @@ use url::Url;
 use urlencoding;
 use utils::http::error::ApiError;

-use crate::auth::backend::ComputeCredentialKeys;
+use crate::auth::backend::ComputeCredentials;
 use crate::auth::backend::ComputeUserInfo;
 use crate::auth::endpoint_sni;
 use crate::auth::ComputeUserInfoParseError;
 use crate::config::AuthenticationConfig;
-use crate::config::HttpConfig;
 use crate::config::ProxyConfig;
 use crate::config::TlsConfig;
 use crate::context::RequestMonitoring;
@@ -57,22 +56,20 @@ use crate::metrics::Metrics;
 use crate::proxy::run_until_cancelled;
 use crate::proxy::NeonOptions;
 use crate::serverless::backend::HttpConnError;
-use crate::usage_metrics::MetricCounter;
 use crate::usage_metrics::MetricCounterRecorder;
 use crate::DbName;
 use crate::RoleName;

 use super::backend::LocalProxyConnError;
 use super::backend::PoolingBackend;
-use super::conn_pool;
 use super::conn_pool::AuthData;
+use super::conn_pool::Client;
 use super::conn_pool::ConnInfo;
 use super::conn_pool::ConnInfoWithAuth;
 use super::http_util::json_response;
 use super::json::json_to_pg_text;
 use super::json::pg_text_row_to_json;
 use super::json::JsonConversionError;
-use super::local_conn_pool;

 #[derive(serde::Deserialize)]
 #[serde(rename_all = "camelCase")]
@@ -555,7 +552,7 @@ async fn handle_inner(

    match conn_info.auth {
        AuthData::Jwt(jwt) if config.authentication_config.is_auth_broker => {
-            handle_auth_broker_inner(ctx, request, conn_info.conn_info, jwt, backend).await
+            handle_auth_broker_inner(config, ctx, request, conn_info.conn_info, jwt, backend).await
        }
        auth => {
            handle_db_inner(
@@ -623,35 +620,37 @@ async fn handle_db_inner(

    let authenticate_and_connect = Box::pin(
        async {
-            let is_local_proxy = matches!(backend.auth_backend, crate::auth::Backend::Local(_));
-
            let keys = match auth {
                AuthData::Password(pw) => {
                    backend
-                        .authenticate_with_password(ctx, &conn_info.user_info, &pw)
+                        .authenticate_with_password(
+                            ctx,
+                            &config.authentication_config,
+                            &conn_info.user_info,
+                            &pw,
+                        )
                        .await?
                }
                AuthData::Jwt(jwt) => {
                    backend
-                        .authenticate_with_jwt(ctx, &conn_info.user_info, jwt)
-                        .await?
-                }
-            };
-
-            let client = match keys.keys {
-                ComputeCredentialKeys::JwtPayload(payload) if is_local_proxy => {
-                    let mut client = backend.connect_to_local_postgres(ctx, conn_info).await?;
-                    client.set_jwt_session(&payload).await?;
-                    Client::Local(client)
-                }
-                _ => {
-                    let client = backend
-                        .connect_to_compute(ctx, conn_info, keys, !allow_pool)
+                        .authenticate_with_jwt(
+                            ctx,
+                            &config.authentication_config,
+                            &conn_info.user_info,
+                            jwt,
+                        )
                        .await?;
-                    Client::Remote(client)
+
+                    ComputeCredentials {
+                        info: conn_info.user_info.clone(),
+                        keys: crate::auth::backend::ComputeCredentialKeys::None,
+                    }
                }
            };

+            let client = backend
+                .connect_to_compute(ctx, conn_info, keys, !allow_pool)
+                .await?;
            // not strictly necessary to mark success here,
            // but it's just insurance for if we forget it somewhere else
            ctx.success();
@@ -681,7 +680,7 @@ async fn handle_db_inner(
    // Now execute the query and return the result.
    let json_output = match payload {
        Payload::Single(stmt) => {
-            stmt.process(&config.http_config, cancel, &mut client, parsed_headers)
+            stmt.process(config, cancel, &mut client, parsed_headers)
                .await?
        }
        Payload::Batch(statements) => {
@@ -699,7 +698,7 @@ async fn handle_db_inner(
            }

            statements
-                .process(&config.http_config, cancel, &mut client, parsed_headers)
+                .process(config, cancel, &mut client, parsed_headers)
                .await?
        }
    };
@@ -739,6 +738,7 @@ static HEADERS_TO_FORWARD: &[&HeaderName] = &[
 ];

 async fn handle_auth_broker_inner(
+    config: &'static ProxyConfig,
    ctx: &RequestMonitoring,
    request: Request<Incoming>,
    conn_info: ConnInfo,
@@ -746,7 +746,12 @@ async fn handle_auth_broker_inner(
    backend: Arc<PoolingBackend>,
 ) -> Result<Response<BoxBody<Bytes, hyper::Error>>, SqlOverHttpError> {
    backend
-        .authenticate_with_jwt(ctx, &conn_info.user_info, jwt)
+        .authenticate_with_jwt(
+            ctx,
+            &config.authentication_config,
+            &conn_info.user_info,
+            jwt,
+        )
        .await
        .map_err(HttpConnError::from)?;

@@ -784,9 +789,9 @@ async fn handle_auth_broker_inner(
 impl QueryData {
    async fn process(
        self,
-        config: &'static HttpConfig,
+        config: &'static ProxyConfig,
        cancel: CancellationToken,
-        client: &mut Client,
+        client: &mut Client<tokio_postgres::Client>,
        parsed_headers: HttpHeaders,
    ) -> Result<String, SqlOverHttpError> {
        let (inner, mut discard) = client.inner();
@@ -815,7 +820,7 @@ impl QueryData {
            Either::Right((_cancelled, query)) => {
                tracing::info!("cancelling query");
                if let Err(err) = cancel_token.cancel_query(NoTls).await {
-                    tracing::warn!(?err, "could not cancel query");
+                    tracing::error!(?err, "could not cancel query");
                }
                // wait for the query cancellation
                match time::timeout(time::Duration::from_millis(100), query).await {
@@ -858,9 +863,9 @@ impl QueryData {
 impl BatchQueryData {
    async fn process(
        self,
-        config: &'static HttpConfig,
+        config: &'static ProxyConfig,
        cancel: CancellationToken,
-        client: &mut Client,
+        client: &mut Client<tokio_postgres::Client>,
        parsed_headers: HttpHeaders,
    ) -> Result<String, SqlOverHttpError> {
        info!("starting transaction");
@@ -904,7 +909,7 @@ impl BatchQueryData {
            }
            Err(SqlOverHttpError::Cancelled(_)) => {
                if let Err(err) = cancel_token.cancel_query(NoTls).await {
-                    tracing::warn!(?err, "could not cancel query");
+                    tracing::error!(?err, "could not cancel query");
                }
                // TODO: after cancelling, wait to see if we can get a status. maybe the connection is still safe.
                discard.discard();
@@ -928,7 +933,7 @@ impl BatchQueryData {
 }

 async fn query_batch(
-    config: &'static HttpConfig,
+    config: &'static ProxyConfig,
    cancel: CancellationToken,
    transaction: &Transaction<'_>,
    queries: BatchQueryData,
@@ -967,7 +972,7 @@ async fn query_batch(
 }

 async fn query_to_json<T: GenericClient>(
-    config: &'static HttpConfig,
+    config: &'static ProxyConfig,
    client: &T,
    data: QueryData,
    current_size: &mut usize,
@@ -988,9 +993,9 @@ async fn query_to_json<T: GenericClient>(
        rows.push(row);
        // we don't have a streaming response support yet so this is to prevent OOM
        // from a malicious query (eg a cross join)
-        if *current_size > config.max_response_size_bytes {
+        if *current_size > config.http_config.max_response_size_bytes {
            return Err(SqlOverHttpError::ResponseTooLarge(
-                config.max_response_size_bytes,
+                config.http_config.max_response_size_bytes,
            ));
        }
    }
@@ -1053,50 +1058,3 @@ async fn query_to_json<T: GenericClient>(

    Ok((ready, results))
 }
-
-enum Client {
-    Remote(conn_pool::Client<tokio_postgres::Client>),
-    Local(local_conn_pool::LocalClient<tokio_postgres::Client>),
-}
-
-enum Discard<'a> {
-    Remote(conn_pool::Discard<'a, tokio_postgres::Client>),
-    Local(local_conn_pool::Discard<'a, tokio_postgres::Client>),
-}
-
-impl Client {
-    fn metrics(&self) -> Arc<MetricCounter> {
-        match self {
-            Client::Remote(client) => client.metrics(),
-            Client::Local(local_client) => local_client.metrics(),
-        }
-    }
-
-    fn inner(&mut self) -> (&mut tokio_postgres::Client, Discard<'_>) {
-        match self {
-            Client::Remote(client) => {
-                let (c, d) = client.inner();
-                (c, Discard::Remote(d))
-            }
-            Client::Local(local_client) => {
-                let (c, d) = local_client.inner();
-                (c, Discard::Local(d))
-            }
-        }
-    }
-}
-
-impl Discard<'_> {
-    fn check_idle(&mut self, status: ReadyForQueryStatus) {
-        match self {
-            Discard::Remote(discard) => discard.check_idle(status),
-            Discard::Local(discard) => discard.check_idle(status),
-        }
-    }
-    fn discard(&mut self) {
-        match self {
-            Discard::Remote(discard) => discard.discard(),
-            Discard::Local(discard) => discard.discard(),
-        }
-    }
-}
--- a/proxy/src/serverless/websocket.rs
+++ b/proxy/src/serverless/websocket.rs
@@ -129,7 +129,6 @@ impl<S: AsyncRead + AsyncWrite + Unpin> AsyncBufRead for WebSocketRw<S> {

 pub(crate) async fn serve_websocket(
    config: &'static ProxyConfig,
-    auth_backend: &'static crate::auth::Backend<'static, ()>,
    ctx: RequestMonitoring,
    websocket: OnUpgrade,
    cancellation_handler: Arc<CancellationHandlerMain>,
@@ -146,7 +145,6 @@ pub(crate) async fn serve_websocket(

    let res = Box::pin(handle_client(
        config,
-        auth_backend,
        &ctx,
        cancellation_handler,
        WebSocketRw::new(websocket),
--- a/proxy/src/usage_metrics.rs
+++ b/proxy/src/usage_metrics.rs
@@ -27,7 +27,7 @@ use std::{
 };
 use tokio::io::AsyncWriteExt;
 use tokio_util::sync::CancellationToken;
-use tracing::{error, info, instrument, trace, warn};
+use tracing::{error, info, instrument, trace};
 use utils::backoff;
 use uuid::{NoContext, Timestamp};

@@ -346,7 +346,7 @@ async fn collect_metrics_iteration(
            error!("metrics endpoint refused the sent metrics: {:?}", res);
            for metric in chunk.events.iter().filter(|e| e.value > (1u64 << 40)) {
                // Report if the metric value is suspiciously large
-                warn!("potentially abnormal metric value: {:?}", metric);
+                error!("potentially abnormal metric value: {:?}", metric);
            }
        }
    }
--- a/safekeeper/src/auth.rs
+++ b/safekeeper/src/auth.rs
@@ -15,20 +15,15 @@ pub fn check_permission(claims: &Claims, tenant_id: Option<TenantId>) -> Result<
            }
            Ok(())
        }
-        (
-            Scope::Admin
-            | Scope::PageServerApi
-            | Scope::GenerationsApi
-            | Scope::Infra
-            | Scope::Scrubber,
-            _,
-        ) => Err(AuthError(
-            format!(
-                "JWT scope '{:?}' is ineligible for Safekeeper auth",
-                claims.scope
-            )
-            .into(),
-        )),
+        (Scope::Admin | Scope::PageServerApi | Scope::GenerationsApi | Scope::Scrubber, _) => {
+            Err(AuthError(
+                format!(
+                    "JWT scope '{:?}' is ineligible for Safekeeper auth",
+                    claims.scope
+                )
+                .into(),
+            ))
+        }
        (Scope::SafekeeperData, _) => Ok(()),
    }
 }
--- a/storage_controller/src/http.rs
+++ b/storage_controller/src/http.rs
@@ -636,7 +636,7 @@ async fn handle_tenant_list(
 }

 async fn handle_node_register(req: Request<Body>) -> Result<Response<Body>, ApiError> {
-    check_permissions(&req, Scope::Infra)?;
+    check_permissions(&req, Scope::Admin)?;

    let mut req = match maybe_forward(req).await {
        ForwardOutcome::Forwarded(res) => {
@@ -1182,7 +1182,7 @@ async fn handle_get_safekeeper(req: Request<Body>) -> Result<Response<Body>, Api
 /// Assumes information is only relayed to storage controller after first selecting an unique id on
 /// control plane database, which means we have an id field in the request and payload.
 async fn handle_upsert_safekeeper(mut req: Request<Body>) -> Result<Response<Body>, ApiError> {
-    check_permissions(&req, Scope::Infra)?;
+    check_permissions(&req, Scope::Admin)?;

    let body = json_request::<SafekeeperPersistence>(&mut req).await?;
    let id = parse_request_param::<i64>(&req, "id")?;
--- a/Show More
+++ b/Show More