Remove VPC endpoint operations as requested

Co-Authored-By: peterbendel@neon.tech <peterbendel@neon.tech>
Fix Python formatting issues
2026-07-09 15:10:38 +00:00 · 2025-03-19 09:16:03 +00:00 · 2025-03-19 09:11:59 +00:00 · 2025-03-19 09:07:42 +00:00 · 2025-03-19 00:13:36 +00:00 · 2025-03-18 18:48:43 +00:00
151 changed files with 4393 additions and 1780 deletions
--- a/.github/scripts/generate_image_maps.py
+++ b/.github/scripts/generate_image_maps.py
@@ -49,10 +49,10 @@ target_stages = (
 for component_name, component_images in components.items():
    for stage in target_stages:
        outputs[f"{component_name}-{stage}"] = {
-            f"docker.io/neondatabase/{component_image}:{source_tag}": [
+            f"ghcr.io/neondatabase/{component_image}:{source_tag}": [
                f"{registry}/{component_image}:{tag}"
                for registry, tag in itertools.product(registries[stage], target_tags)
-                if not (registry == "docker.io/neondatabase" and tag == source_tag)
+                if not (registry == "ghcr.io/neondatabase" and tag == source_tag)
            ]
            for component_image in component_images
        }
--- a/.github/workflows/_benchmarking_preparation.yml
+++ b/.github/workflows/_benchmarking_preparation.yml
@@ -27,10 +27,10 @@ jobs:

    runs-on: [ self-hosted, us-east-2, x64 ]
    container:
-      image: neondatabase/build-tools:pinned-bookworm
+      image: ghcr.io/neondatabase/build-tools:pinned-bookworm
      credentials:
-        username: ${{ secrets.NEON_DOCKERHUB_USERNAME }}
-        password: ${{ secrets.NEON_DOCKERHUB_PASSWORD }}
+        username: ${{ github.actor }}
+        password: ${{ secrets.GITHUB_TOKEN }}
      options: --init

    steps:
--- a/.github/workflows/_build-and-test-locally.yml
+++ b/.github/workflows/_build-and-test-locally.yml
@@ -39,15 +39,15 @@ env:

 jobs:
  build-neon:
-    runs-on: ${{ fromJson(format('["self-hosted", "{0}"]', inputs.arch == 'arm64' && 'large-arm64' || 'large')) }}
+    runs-on: ${{ fromJSON(format('["self-hosted", "{0}"]', inputs.arch == 'arm64' && 'large-arm64' || 'large')) }}
    permissions:
      id-token: write # aws-actions/configure-aws-credentials
      contents: read
    container:
      image: ${{ inputs.build-tools-image }}
      credentials:
-        username: ${{ secrets.NEON_DOCKERHUB_USERNAME }}
-        password: ${{ secrets.NEON_DOCKERHUB_PASSWORD }}
+        username: ${{ github.actor }}
+        password: ${{ secrets.GITHUB_TOKEN }}
      # Raise locked memory limit for tokio-epoll-uring.
      # On 5.10 LTS kernels < 5.10.162 (and generally mainline kernels < 5.12),
      # io_uring will account the memory of the CQ and SQ as locked.
@@ -318,12 +318,12 @@ jobs:
      contents: read
      statuses: write
    needs: [ build-neon ]
-    runs-on: ${{ fromJson(format('["self-hosted", "{0}"]', inputs.arch == 'arm64' && 'large-arm64' || 'large')) }}
+    runs-on: ${{ fromJSON(format('["self-hosted", "{0}"]', inputs.arch == 'arm64' && 'large-arm64' || 'large')) }}
    container:
      image: ${{ inputs.build-tools-image }}
      credentials:
-        username: ${{ secrets.NEON_DOCKERHUB_USERNAME }}
-        password: ${{ secrets.NEON_DOCKERHUB_PASSWORD }}
+        username: ${{ github.actor }}
+        password: ${{ secrets.GITHUB_TOKEN }}
      # for changed limits, see comments on `options:` earlier in this file
      options: --init --shm-size=512mb --ulimit memlock=67108864:67108864
    strategy:
--- a/.github/workflows/_check-codestyle-python.yml
+++ b/.github/workflows/_check-codestyle-python.yml
@@ -15,11 +15,15 @@ defaults:
 jobs:
  check-codestyle-python:
    runs-on: [ self-hosted, small ]
+
+    permissions:
+      packages: read
+
    container:
      image: ${{ inputs.build-tools-image }}
      credentials:
-        username: ${{ secrets.NEON_DOCKERHUB_USERNAME }}
-        password: ${{ secrets.NEON_DOCKERHUB_PASSWORD }}
+        username: ${{ github.actor }}
+        password: ${{ secrets.GITHUB_TOKEN }}
      options: --init

    steps:
--- a/.github/workflows/_check-codestyle-rust.yml
+++ b/.github/workflows/_check-codestyle-rust.yml
@@ -23,14 +23,17 @@ jobs:
  check-codestyle-rust:
    strategy:
      matrix:
-        arch: ${{ fromJson(inputs.archs) }}
-    runs-on: ${{ fromJson(format('["self-hosted", "{0}"]', matrix.arch == 'arm64' && 'small-arm64' || 'small')) }}
+        arch: ${{ fromJSON(inputs.archs) }}
+    runs-on: ${{ fromJSON(format('["self-hosted", "{0}"]', matrix.arch == 'arm64' && 'small-arm64' || 'small')) }}
+
+    permissions:
+      packages: read

    container:
      image: ${{ inputs.build-tools-image }}
      credentials:
-        username: ${{ secrets.NEON_DOCKERHUB_USERNAME }}
-        password: ${{ secrets.NEON_DOCKERHUB_PASSWORD }}
+        username: ${{ github.actor }}
+        password: ${{ secrets.GITHUB_TOKEN }}
      options: --init

    steps:
--- a/.github/workflows/_meta.yml
+++ b/.github/workflows/_meta.yml
@@ -120,10 +120,10 @@ jobs:

      - name: Get the release PR run ID
        id: release-pr-run-id
-        if: ${{ contains(fromJson('["storage-release", "compute-release", "proxy-release"]'), steps.run-kind.outputs.run-kind) }}
+        if: ${{ contains(fromJSON('["storage-release", "compute-release", "proxy-release"]'), steps.run-kind.outputs.run-kind) }}
        env:
          GH_TOKEN: ${{ secrets.GITHUB_TOKEN }}
          CURRENT_SHA: ${{ github.event.pull_request.head.sha || github.sha }}
        run: |
-          RELEASE_PR_RUN_ID=$(gh api "/repos/${GITHUB_REPOSITORY}/actions/runs?head_sha=$CURRENT_SHA" | jq '[.workflow_runs[] | select(.name == "Build and Test") | select(.head_branch | test("^rc/release(-(proxy)|(compute))?/[0-9]{4}-[0-9]{2}-[0-9]{2}$"; "s"))] | first | .id // "Faied to find Build and Test run from  RC PR!" | halt_error(1)')
+          RELEASE_PR_RUN_ID=$(gh api "/repos/${GITHUB_REPOSITORY}/actions/runs?head_sha=$CURRENT_SHA" | jq '[.workflow_runs[] | select(.name == "Build and Test") | select(.head_branch | test("^rc/release(-(proxy|compute))?/[0-9]{4}-[0-9]{2}-[0-9]{2}$"; "s"))] | first | .id // ("Failed to find Build and Test run from  RC PR!" | halt_error(1))')
          echo "release-pr-run-id=$RELEASE_PR_RUN_ID" | tee -a $GITHUB_OUTPUT
--- a/.github/workflows/_push-to-container-registry.yml
+++ b/.github/workflows/_push-to-container-registry.yml
@@ -89,7 +89,7 @@ jobs:
        uses: docker/login-action@v3
        with:
          registry: ghcr.io
-          username: ${{ github.repository_owner }}
+          username: ${{ github.actor }}
          password: ${{ secrets.GITHUB_TOKEN }}

      - name: Log in to Docker Hub
--- a/.github/workflows/benchmarking.yml
+++ b/.github/workflows/benchmarking.yml
@@ -87,10 +87,10 @@ jobs:

    runs-on: ${{ matrix.RUNNER }}
    container:
-      image: neondatabase/build-tools:pinned-bookworm
+      image: ghcr.io/neondatabase/build-tools:pinned-bookworm
      credentials:
-        username: ${{ secrets.NEON_DOCKERHUB_USERNAME }}
-        password: ${{ secrets.NEON_DOCKERHUB_PASSWORD }}
+        username: ${{ github.actor }}
+        password: ${{ secrets.GITHUB_TOKEN }}
      options: --init

    steps:
@@ -190,10 +190,10 @@ jobs:

    runs-on: [ self-hosted, us-east-2, x64 ]
    container:
-      image: neondatabase/build-tools:pinned-bookworm
+      image: ghcr.io/neondatabase/build-tools:pinned-bookworm
      credentials:
-        username: ${{ secrets.NEON_DOCKERHUB_USERNAME }}
-        password: ${{ secrets.NEON_DOCKERHUB_PASSWORD }}
+        username: ${{ github.actor }}
+        password: ${{ secrets.GITHUB_TOKEN }}
      options: --init

    steps:
@@ -245,10 +245,10 @@ jobs:

    runs-on: [ self-hosted, us-east-2, x64 ]
    container:
-      image: neondatabase/build-tools:pinned-bookworm
+      image: ghcr.io/neondatabase/build-tools:pinned-bookworm
      credentials:
-        username: ${{ secrets.NEON_DOCKERHUB_USERNAME }}
-        password: ${{ secrets.NEON_DOCKERHUB_PASSWORD }}
+        username: ${{ github.actor }}
+        password: ${{ secrets.GITHUB_TOKEN }}
      options: --init

    steps:
@@ -352,7 +352,7 @@ jobs:
        region_id_default=${{ env.DEFAULT_REGION_ID }}
        runner_default='["self-hosted", "us-east-2", "x64"]'
        runner_azure='["self-hosted", "eastus2", "x64"]'
-        image_default="neondatabase/build-tools:pinned-bookworm"
+        image_default="ghcr.io/neondatabase/build-tools:pinned-bookworm"
        matrix='{
          "pg_version" : [
            16
@@ -368,18 +368,18 @@ jobs:
          "db_size": [ "10gb" ],
          "runner": ['"$runner_default"'],
          "image": [ "'"$image_default"'" ],
-          "include": [{ "pg_version": 16, "region_id": "'"$region_id_default"'", "platform": "neonvm-captest-freetier",       "db_size": "3gb" ,"runner": '"$runner_default"', "image": "'"$image_default"'" },
-                      { "pg_version": 16, "region_id": "'"$region_id_default"'", "platform": "neonvm-captest-new",            "db_size": "10gb","runner": '"$runner_default"', "image": "'"$image_default"'" },
-                      { "pg_version": 16, "region_id": "'"$region_id_default"'", "platform": "neonvm-captest-new-many-tables","db_size": "10gb","runner": '"$runner_default"', "image": "'"$image_default"'" },
-                      { "pg_version": 16, "region_id": "'"$region_id_default"'", "platform": "neonvm-captest-new",            "db_size": "50gb","runner": '"$runner_default"', "image": "'"$image_default"'" },
-                      { "pg_version": 16, "region_id": "azure-eastus2",          "platform": "neonvm-azure-captest-freetier", "db_size": "3gb" ,"runner": '"$runner_azure"',   "image": "neondatabase/build-tools:pinned-bookworm" },
-                      { "pg_version": 16, "region_id": "azure-eastus2",          "platform": "neonvm-azure-captest-new",      "db_size": "10gb","runner": '"$runner_azure"',   "image": "neondatabase/build-tools:pinned-bookworm" },
-                      { "pg_version": 16, "region_id": "azure-eastus2",          "platform": "neonvm-azure-captest-new",      "db_size": "50gb","runner": '"$runner_azure"',   "image": "neondatabase/build-tools:pinned-bookworm" },
-                      { "pg_version": 16, "region_id": "'"$region_id_default"'", "platform": "neonvm-captest-sharding-reuse", "db_size": "50gb","runner": '"$runner_default"', "image": "'"$image_default"'" },
-                      { "pg_version": 17, "region_id": "'"$region_id_default"'", "platform": "neonvm-captest-freetier",       "db_size": "3gb" ,"runner": '"$runner_default"', "image": "'"$image_default"'" },
-                      { "pg_version": 17, "region_id": "'"$region_id_default"'", "platform": "neonvm-captest-new",            "db_size": "10gb","runner": '"$runner_default"', "image": "'"$image_default"'" },
-                      { "pg_version": 17, "region_id": "'"$region_id_default"'", "platform": "neonvm-captest-new-many-tables","db_size": "10gb","runner": '"$runner_default"', "image": "'"$image_default"'" },
-                      { "pg_version": 17, "region_id": "'"$region_id_default"'", "platform": "neonvm-captest-new",            "db_size": "50gb","runner": '"$runner_default"', "image": "'"$image_default"'" }]
+          "include": [{ "pg_version": 16, "region_id": "'"$region_id_default"'", "platform": "neonvm-captest-freetier",       "db_size": "3gb" ,"runner": '"$runner_default"', "image": "'"$image_default"'"                               },
+                      { "pg_version": 16, "region_id": "'"$region_id_default"'", "platform": "neonvm-captest-new",            "db_size": "10gb","runner": '"$runner_default"', "image": "'"$image_default"'"                               },
+                      { "pg_version": 16, "region_id": "'"$region_id_default"'", "platform": "neonvm-captest-new-many-tables","db_size": "10gb","runner": '"$runner_default"', "image": "'"$image_default"'"                               },
+                      { "pg_version": 16, "region_id": "'"$region_id_default"'", "platform": "neonvm-captest-new",            "db_size": "50gb","runner": '"$runner_default"', "image": "'"$image_default"'"                               },
+                      { "pg_version": 16, "region_id": "azure-eastus2",          "platform": "neonvm-azure-captest-freetier", "db_size": "3gb" ,"runner": '"$runner_azure"',   "image": "ghcr.io/neondatabase/build-tools:pinned-bookworm" },
+                      { "pg_version": 16, "region_id": "azure-eastus2",          "platform": "neonvm-azure-captest-new",      "db_size": "10gb","runner": '"$runner_azure"',   "image": "ghcr.io/neondatabase/build-tools:pinned-bookworm" },
+                      { "pg_version": 16, "region_id": "azure-eastus2",          "platform": "neonvm-azure-captest-new",      "db_size": "50gb","runner": '"$runner_azure"',   "image": "ghcr.io/neondatabase/build-tools:pinned-bookworm" },
+                      { "pg_version": 16, "region_id": "'"$region_id_default"'", "platform": "neonvm-captest-sharding-reuse", "db_size": "50gb","runner": '"$runner_default"', "image": "'"$image_default"'"                               },
+                      { "pg_version": 17, "region_id": "'"$region_id_default"'", "platform": "neonvm-captest-freetier",       "db_size": "3gb" ,"runner": '"$runner_default"', "image": "'"$image_default"'"                               },
+                      { "pg_version": 17, "region_id": "'"$region_id_default"'", "platform": "neonvm-captest-new",            "db_size": "10gb","runner": '"$runner_default"', "image": "'"$image_default"'"                               },
+                      { "pg_version": 17, "region_id": "'"$region_id_default"'", "platform": "neonvm-captest-new-many-tables","db_size": "10gb","runner": '"$runner_default"', "image": "'"$image_default"'"                               },
+                      { "pg_version": 17, "region_id": "'"$region_id_default"'", "platform": "neonvm-captest-new",            "db_size": "50gb","runner": '"$runner_default"', "image": "'"$image_default"'"                               }]
        }'

        if [ "$(date +%A)" = "Saturday" ] || [ ${RUN_AWS_RDS_AND_AURORA} = "true" ]; then
@@ -441,7 +441,7 @@ jobs:

    strategy:
      fail-fast: false
-      matrix: ${{fromJson(needs.generate-matrices.outputs.pgbench-compare-matrix)}}
+      matrix: ${{fromJSON(needs.generate-matrices.outputs.pgbench-compare-matrix)}}

    env:
      TEST_PG_BENCH_DURATIONS_MATRIX: "60m"
@@ -457,8 +457,8 @@ jobs:
    container:
      image: ${{ matrix.image }}
      credentials:
-        username: ${{ secrets.NEON_DOCKERHUB_USERNAME }}
-        password: ${{ secrets.NEON_DOCKERHUB_PASSWORD }}
+        username: ${{ github.actor }}
+        password: ${{ secrets.GITHUB_TOKEN }}
      options: --init

    # Increase timeout to 8h, default timeout is 6h
@@ -483,7 +483,7 @@ jobs:
        aws-oicd-role-arn: ${{ vars.DEV_AWS_OIDC_ROLE_ARN }}

    - name: Create Neon Project
-      if: contains(fromJson('["neonvm-captest-new", "neonvm-captest-new-many-tables", "neonvm-captest-freetier", "neonvm-azure-captest-freetier", "neonvm-azure-captest-new"]'), matrix.platform)
+      if: contains(fromJSON('["neonvm-captest-new", "neonvm-captest-new-many-tables", "neonvm-captest-freetier", "neonvm-azure-captest-freetier", "neonvm-azure-captest-new"]'), matrix.platform)
      id: create-neon-project
      uses: ./.github/actions/neon-project-create
      with:
@@ -523,7 +523,7 @@ jobs:
    # without (neonvm-captest-new)
    # and with (neonvm-captest-new-many-tables) many relations in the database
    - name: Create many relations before the run
-      if: contains(fromJson('["neonvm-captest-new-many-tables"]'), matrix.platform)
+      if: contains(fromJSON('["neonvm-captest-new-many-tables"]'), matrix.platform)
      uses: ./.github/actions/run-python-test-set
      with:
        build_type: ${{ env.BUILD_TYPE }}
@@ -642,10 +642,10 @@ jobs:

    runs-on: ${{ matrix.RUNNER }}
    container:
-      image: neondatabase/build-tools:pinned-bookworm
+      image: ghcr.io/neondatabase/build-tools:pinned-bookworm
      credentials:
-        username: ${{ secrets.NEON_DOCKERHUB_USERNAME }}
-        password: ${{ secrets.NEON_DOCKERHUB_PASSWORD }}
+        username: ${{ github.actor }}
+        password: ${{ secrets.GITHUB_TOKEN }}
      options: --init

    steps:
@@ -753,7 +753,7 @@ jobs:

    strategy:
      fail-fast: false
-      matrix: ${{ fromJson(needs.generate-matrices.outputs.olap-compare-matrix) }}
+      matrix: ${{ fromJSON(needs.generate-matrices.outputs.olap-compare-matrix) }}

    env:
      POSTGRES_DISTRIB_DIR: /tmp/neon/pg_install
@@ -767,10 +767,10 @@ jobs:

    runs-on: [ self-hosted, us-east-2, x64 ]
    container:
-      image: neondatabase/build-tools:pinned-bookworm
+      image: ghcr.io/neondatabase/build-tools:pinned-bookworm
      credentials:
-        username: ${{ secrets.NEON_DOCKERHUB_USERNAME }}
-        password: ${{ secrets.NEON_DOCKERHUB_PASSWORD }}
+        username: ${{ github.actor }}
+        password: ${{ secrets.GITHUB_TOKEN }}
      options: --init

    # Increase timeout to 12h, default timeout is 6h
@@ -880,7 +880,7 @@ jobs:

    strategy:
      fail-fast: false
-      matrix: ${{ fromJson(needs.generate-matrices.outputs.tpch-compare-matrix) }}
+      matrix: ${{ fromJSON(needs.generate-matrices.outputs.tpch-compare-matrix) }}

    env:
      POSTGRES_DISTRIB_DIR: /tmp/neon/pg_install
@@ -892,10 +892,10 @@ jobs:

    runs-on: [ self-hosted, us-east-2, x64 ]
    container:
-      image: neondatabase/build-tools:pinned-bookworm
+      image: ghcr.io/neondatabase/build-tools:pinned-bookworm
      credentials:
-        username: ${{ secrets.NEON_DOCKERHUB_USERNAME }}
-        password: ${{ secrets.NEON_DOCKERHUB_PASSWORD }}
+        username: ${{ github.actor }}
+        password: ${{ secrets.GITHUB_TOKEN }}
      options: --init

    steps:
@@ -999,7 +999,7 @@ jobs:

    strategy:
      fail-fast: false
-      matrix: ${{ fromJson(needs.generate-matrices.outputs.olap-compare-matrix) }}
+      matrix: ${{ fromJSON(needs.generate-matrices.outputs.olap-compare-matrix) }}

    env:
      POSTGRES_DISTRIB_DIR: /tmp/neon/pg_install
@@ -1011,10 +1011,10 @@ jobs:

    runs-on: [ self-hosted, us-east-2, x64 ]
    container:
-      image: neondatabase/build-tools:pinned-bookworm
+      image: ghcr.io/neondatabase/build-tools:pinned-bookworm
      credentials:
-        username: ${{ secrets.NEON_DOCKERHUB_USERNAME }}
-        password: ${{ secrets.NEON_DOCKERHUB_PASSWORD }}
+        username: ${{ github.actor }}
+        password: ${{ secrets.GITHUB_TOKEN }}
      options: --init

    steps:
--- a/.github/workflows/build-build-tools-image.yml
+++ b/.github/workflows/build-build-tools-image.yml
@@ -19,7 +19,7 @@ on:
        value: ${{ jobs.check-image.outputs.tag }}
      image:
        description: "build-tools image"
-        value: neondatabase/build-tools:${{ jobs.check-image.outputs.tag }}
+        value: ghcr.io/neondatabase/build-tools:${{ jobs.check-image.outputs.tag }}

 defaults:
  run:
@@ -49,9 +49,18 @@ jobs:
      everything: ${{ steps.set-more-variables.outputs.everything }}
      found: ${{ steps.set-more-variables.outputs.found }}

+    permissions:
+      packages: read
+
    steps:
      - uses: actions/checkout@v4

+      - uses: docker/login-action@v3
+        with:
+          registry: ghcr.io
+          username: ${{ github.actor }}
+          password: ${{ secrets.GITHUB_TOKEN }}
+
      - name: Set variables
        id: set-variables
        env:
@@ -70,12 +79,12 @@ jobs:
        env:
          IMAGE_TAG: ${{ steps.set-variables.outputs.image-tag }}
          EVERYTHING: |
-            ${{ contains(fromJson(steps.set-variables.outputs.archs), 'x64') &&
-                contains(fromJson(steps.set-variables.outputs.archs), 'arm64') &&
-                contains(fromJson(steps.set-variables.outputs.debians), 'bullseye') &&
-                contains(fromJson(steps.set-variables.outputs.debians), 'bookworm') }}
+            ${{ contains(fromJSON(steps.set-variables.outputs.archs), 'x64') &&
+                contains(fromJSON(steps.set-variables.outputs.archs), 'arm64') &&
+                contains(fromJSON(steps.set-variables.outputs.debians), 'bullseye') &&
+                contains(fromJSON(steps.set-variables.outputs.debians), 'bookworm') }}
        run: |
-          if docker manifest inspect neondatabase/build-tools:${IMAGE_TAG}; then
+          if docker manifest inspect ghcr.io/neondatabase/build-tools:${IMAGE_TAG}; then
            found=true
          else
            found=false
@@ -90,10 +99,13 @@ jobs:

    strategy:
      matrix:
-        arch: ${{ fromJson(needs.check-image.outputs.archs) }}
-        debian: ${{ fromJson(needs.check-image.outputs.debians) }}
+        arch: ${{ fromJSON(needs.check-image.outputs.archs) }}
+        debian: ${{ fromJSON(needs.check-image.outputs.debians) }}

-    runs-on: ${{ fromJson(format('["self-hosted", "{0}"]', matrix.arch == 'arm64' && 'large-arm64' || 'large')) }}
+    permissions:
+      packages: write
+
+    runs-on: ${{ fromJSON(format('["self-hosted", "{0}"]', matrix.arch == 'arm64' && 'large-arm64' || 'large')) }}

    steps:
      - uses: actions/checkout@v4
@@ -108,6 +120,12 @@ jobs:
          username: ${{ secrets.NEON_DOCKERHUB_USERNAME }}
          password: ${{ secrets.NEON_DOCKERHUB_PASSWORD }}

+      - uses: docker/login-action@v3
+        with:
+          registry: ghcr.io
+          username: ${{ github.actor }}
+          password: ${{ secrets.GITHUB_TOKEN }}
+
      - uses: docker/login-action@v3
        with:
          registry: cache.neon.build
@@ -126,35 +144,44 @@ jobs:
          cache-from: type=registry,ref=cache.neon.build/build-tools:cache-${{ matrix.debian }}-${{ matrix.arch }}
          cache-to: ${{ github.ref_name == 'main' && format('type=registry,ref=cache.neon.build/build-tools:cache-{0}-{1},mode=max', matrix.debian, matrix.arch) || '' }}
          tags: |
-            neondatabase/build-tools:${{ needs.check-image.outputs.tag }}-${{ matrix.debian }}-${{ matrix.arch }}
+            ghcr.io/neondatabase/build-tools:${{ needs.check-image.outputs.tag }}-${{ matrix.debian }}-${{ matrix.arch }}

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

+    permissions:
+      packages: write
+
    steps:
      - uses: docker/login-action@v3
        with:
          username: ${{ secrets.NEON_DOCKERHUB_USERNAME }}
          password: ${{ secrets.NEON_DOCKERHUB_PASSWORD }}

+      - uses: docker/login-action@v3
+        with:
+          registry: ghcr.io
+          username: ${{ github.actor }}
+          password: ${{ secrets.GITHUB_TOKEN }}
+
      - name: Create multi-arch image
        env:
          DEFAULT_DEBIAN_VERSION: bookworm
-          ARCHS: ${{ join(fromJson(needs.check-image.outputs.archs), ' ') }}
-          DEBIANS: ${{ join(fromJson(needs.check-image.outputs.debians), ' ') }}
+          ARCHS: ${{ join(fromJSON(needs.check-image.outputs.archs), ' ') }}
+          DEBIANS: ${{ join(fromJSON(needs.check-image.outputs.debians), ' ') }}
          EVERYTHING: ${{ needs.check-image.outputs.everything }}
          IMAGE_TAG: ${{ needs.check-image.outputs.tag }}
        run: |
          for debian in ${DEBIANS}; do
-            tags=("-t" "neondatabase/build-tools:${IMAGE_TAG}-${debian}")
+            tags=("-t" "ghcr.io/neondatabase/build-tools:${IMAGE_TAG}-${debian}")

            if [ "${EVERYTHING}" == "true" ] && [ "${debian}" == "${DEFAULT_DEBIAN_VERSION}" ]; then
-              tags+=("-t" "neondatabase/build-tools:${IMAGE_TAG}")
+              tags+=("-t" "ghcr.io/neondatabase/build-tools:${IMAGE_TAG}")
            fi

            for arch in ${ARCHS}; do
-              tags+=("neondatabase/build-tools:${IMAGE_TAG}-${debian}-${arch}")
+              tags+=("ghcr.io/neondatabase/build-tools:${IMAGE_TAG}-${debian}-${arch}")
            done

            docker buildx imagetools create "${tags[@]}"
--- a/.github/workflows/build-macos.yml
+++ b/.github/workflows/build-macos.yml
@@ -40,7 +40,7 @@ jobs:
    runs-on: macos-15
    strategy:
      matrix:
-        postgres-version: ${{ inputs.rebuild_everything && fromJson('["v14", "v15", "v16", "v17"]') || fromJSON(inputs.pg_versions) }}
+        postgres-version: ${{ inputs.rebuild_everything && fromJSON('["v14", "v15", "v16", "v17"]') || fromJSON(inputs.pg_versions) }}
    env:
      # Use release build only, to have less debug info around
      # Hence keeping target/ (and general cache size) smaller
--- a/.github/workflows/build_and_test.yml
+++ b/.github/workflows/build_and_test.yml
@@ -77,20 +77,23 @@ jobs:
    secrets: inherit

  check-codestyle-python:
-    needs: [ check-permissions, build-build-tools-image ]
+    needs: [ meta, check-permissions, build-build-tools-image ]
+    # No need to run on `main` because we this in the merge queue
+    if: ${{ needs.meta.outputs.run-kind == 'pr' }}
    uses: ./.github/workflows/_check-codestyle-python.yml
    with:
      build-tools-image: ${{ needs.build-build-tools-image.outputs.image }}-bookworm
    secrets: inherit

  check-codestyle-jsonnet:
-    needs: [ check-permissions, build-build-tools-image ]
+    needs: [ meta, check-permissions, build-build-tools-image ]
+    if: ${{ contains(fromJSON('["pr", "push-main"]'), needs.meta.outputs.run-kind) }}
    runs-on: [ self-hosted, small ]
    container:
      image: ${{ needs.build-build-tools-image.outputs.image }}
      credentials:
-        username: ${{ secrets.NEON_DOCKERHUB_USERNAME }}
-        password: ${{ secrets.NEON_DOCKERHUB_PASSWORD }}
+        username: ${{ github.actor }}
+        password: ${{ secrets.GITHUB_TOKEN }}
      options: --init

    steps:
@@ -156,7 +159,9 @@ jobs:
          pass_if_unchanged: true

  check-codestyle-rust:
-    needs: [ check-permissions, build-build-tools-image ]
+    needs: [ meta, check-permissions, build-build-tools-image ]
+    # No need to run on `main` because we this in the merge queue
+    if: ${{ needs.meta.outputs.run-kind == 'pr' }}
    uses: ./.github/workflows/_check-codestyle-rust.yml
    with:
      build-tools-image: ${{ needs.build-build-tools-image.outputs.image }}-bookworm
@@ -164,8 +169,8 @@ jobs:
    secrets: inherit

  check-dependencies-rust:
-    needs: [ files-changed, build-build-tools-image ]
-    if: ${{ needs.files-changed.outputs.check-rust-dependencies == 'true' }}
+    needs: [ meta, files-changed, build-build-tools-image ]
+    if: ${{ needs.files-changed.outputs.check-rust-dependencies == 'true' && needs.meta.outputs.run-kind == 'pr' }}
    uses: ./.github/workflows/cargo-deny.yml
    with:
      build-tools-image: ${{ needs.build-build-tools-image.outputs.image }}-bookworm
@@ -173,12 +178,13 @@ jobs:

  build-and-test-locally:
    needs: [ meta, build-build-tools-image ]
+    if: ${{ contains(fromJSON('["pr", "push-main"]'), needs.meta.outputs.run-kind) }}
    strategy:
      fail-fast: false
      matrix:
        arch: [ x64, arm64 ]
        # Do not build or run tests in debug for release branches
-        build-type: ${{ fromJson((startsWith(github.ref_name, 'release') && github.event_name == 'push') && '["release"]' || '["debug", "release"]') }}
+        build-type: ${{ fromJSON((startsWith(github.ref_name, 'release') && github.event_name == 'push') && '["release"]' || '["debug", "release"]') }}
        include:
          - build-type: release
            arch: arm64
@@ -209,8 +215,8 @@ jobs:
    container:
      image: ${{ needs.build-build-tools-image.outputs.image }}-bookworm
      credentials:
-        username: ${{ secrets.NEON_DOCKERHUB_USERNAME }}
-        password: ${{ secrets.NEON_DOCKERHUB_PASSWORD }}
+        username: ${{ github.actor }}
+        password: ${{ secrets.GITHUB_TOKEN }}
      options: --init
    steps:
      - name: Checkout
@@ -248,8 +254,8 @@ jobs:
    container:
      image: ${{ needs.build-build-tools-image.outputs.image }}-bookworm
      credentials:
-        username: ${{ secrets.NEON_DOCKERHUB_USERNAME }}
-        password: ${{ secrets.NEON_DOCKERHUB_PASSWORD }}
+        username: ${{ github.actor }}
+        password: ${{ secrets.GITHUB_TOKEN }}
      # for changed limits, see comments on `options:` earlier in this file
      options: --init --shm-size=512mb --ulimit memlock=67108864:67108864
    strategy:
@@ -314,8 +320,8 @@ jobs:
    container:
      image: ${{ needs.build-build-tools-image.outputs.image }}-bookworm
      credentials:
-        username: ${{ secrets.NEON_DOCKERHUB_USERNAME }}
-        password: ${{ secrets.NEON_DOCKERHUB_PASSWORD }}
+        username: ${{ github.actor }}
+        password: ${{ secrets.GITHUB_TOKEN }}
      options: --init

    steps:
@@ -367,8 +373,8 @@ jobs:
    container:
      image: ${{ needs.build-build-tools-image.outputs.image }}-bookworm
      credentials:
-        username: ${{ secrets.NEON_DOCKERHUB_USERNAME }}
-        password: ${{ secrets.NEON_DOCKERHUB_PASSWORD }}
+        username: ${{ github.actor }}
+        password: ${{ secrets.GITHUB_TOKEN }}
      options: --init
    strategy:
      fail-fast: false
@@ -470,14 +476,20 @@ jobs:
            })

  trigger-e2e-tests:
-    # Depends on jobs that can get skipped
+    # !failure() && !cancelled() because it depends on jobs that can get skipped
    if: >-
      ${{
        (
-          !github.event.pull_request.draft
-          || contains( github.event.pull_request.labels.*.name, 'run-e2e-tests-in-draft')
-          || needs.meta.outputs.run-kind == 'push-main'
-        ) && !failure() && !cancelled()
+          (
+            needs.meta.outputs.run-kind == 'pr'
+            && (
+              !github.event.pull_request.draft
+              || contains(github.event.pull_request.labels.*.name, 'run-e2e-tests-in-draft')
+            )
+          )
+          || contains(fromJSON('["push-main", "storage-rc-pr", "proxy-rc-pr", "compute-rc-pr"]'), needs.meta.outputs.run-kind)
+        )
+        && !failure() && !cancelled()
      }}
    needs: [ check-permissions, push-neon-image-dev, push-compute-image-dev, meta ]
    uses: ./.github/workflows/trigger-e2e-tests.yml
@@ -492,7 +504,10 @@ jobs:
      matrix:
        arch: [ x64, arm64 ]

-    runs-on: ${{ fromJson(format('["self-hosted", "{0}"]', matrix.arch == 'arm64' && 'large-arm64' || 'large')) }}
+    runs-on: ${{ fromJSON(format('["self-hosted", "{0}"]', matrix.arch == 'arm64' && 'large-arm64' || 'large')) }}
+
+    permissions:
+      packages: write

    steps:
      - uses: actions/checkout@v4
@@ -509,6 +524,12 @@ jobs:
          username: ${{ secrets.NEON_DOCKERHUB_USERNAME }}
          password: ${{ secrets.NEON_DOCKERHUB_PASSWORD }}

+      - uses: docker/login-action@v3
+        with:
+          registry: ghcr.io
+          username: ${{ github.actor }}
+          password: ${{ secrets.GITHUB_TOKEN }}
+
      - uses: docker/login-action@v3
        with:
          registry: cache.neon.build
@@ -533,7 +554,7 @@ jobs:
          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) || '' }}
          tags: |
-            neondatabase/neon:${{ needs.meta.outputs.build-tag }}-bookworm-${{ matrix.arch }}
+            ghcr.io/neondatabase/neon:${{ needs.meta.outputs.build-tag }}-bookworm-${{ matrix.arch }}

  neon-image:
    needs: [ neon-image-arch, meta ]
@@ -543,19 +564,21 @@ jobs:
      id-token: write # aws-actions/configure-aws-credentials
      statuses: write
      contents: read
+      packages: write

    steps:
      - uses: docker/login-action@v3
        with:
-          username: ${{ secrets.NEON_DOCKERHUB_USERNAME }}
-          password: ${{ secrets.NEON_DOCKERHUB_PASSWORD }}
+          registry: ghcr.io
+          username: ${{ github.actor }}
+          password: ${{ secrets.GITHUB_TOKEN }}

      - name: Create multi-arch image
        run: |
-          docker buildx imagetools create -t neondatabase/neon:${{ needs.meta.outputs.build-tag }} \
-                                          -t neondatabase/neon:${{ needs.meta.outputs.build-tag }}-bookworm \
-                                             neondatabase/neon:${{ needs.meta.outputs.build-tag }}-bookworm-x64 \
-                                             neondatabase/neon:${{ needs.meta.outputs.build-tag }}-bookworm-arm64
+          docker buildx imagetools create -t ghcr.io/neondatabase/neon:${{ needs.meta.outputs.build-tag }} \
+                                          -t ghcr.io/neondatabase/neon:${{ needs.meta.outputs.build-tag }}-bookworm \
+                                             ghcr.io/neondatabase/neon:${{ needs.meta.outputs.build-tag }}-bookworm-x64 \
+                                             ghcr.io/neondatabase/neon:${{ needs.meta.outputs.build-tag }}-bookworm-arm64

  compute-node-image-arch:
    needs: [ check-permissions, build-build-tools-image, meta ]
@@ -564,6 +587,7 @@ jobs:
      id-token: write # aws-actions/configure-aws-credentials
      statuses: write
      contents: read
+      packages: write
    strategy:
      fail-fast: false
      matrix:
@@ -582,7 +606,7 @@ jobs:
            debian: bookworm
        arch: [ x64, arm64 ]

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

    steps:
      - uses: actions/checkout@v4
@@ -604,6 +628,12 @@ jobs:
          username: ${{ secrets.NEON_DOCKERHUB_USERNAME }}
          password: ${{ secrets.NEON_DOCKERHUB_PASSWORD }}

+      - uses: docker/login-action@v3
+        with:
+          registry: ghcr.io
+          username: ${{ github.actor }}
+          password: ${{ secrets.GITHUB_TOKEN }}
+
      - uses: docker/login-action@v3
        with:
          registry: cache.neon.build
@@ -627,7 +657,7 @@ jobs:
          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) || '' }}
          tags: |
-            neondatabase/compute-node-${{ matrix.version.pg }}:${{ needs.meta.outputs.build-tag }}-${{ matrix.version.debian }}-${{ matrix.arch }}
+            ghcr.io/neondatabase/compute-node-${{ matrix.version.pg }}:${{ needs.meta.outputs.build-tag }}-${{ matrix.version.debian }}-${{ matrix.arch }}

      - name: Build neon extensions test image
        if: matrix.version.pg >= 'v16'
@@ -647,7 +677,7 @@ jobs:
          target: extension-tests
          cache-from: type=registry,ref=cache.neon.build/compute-node-${{ matrix.version.pg }}:cache-${{ matrix.version.debian }}-${{ matrix.arch }}
          tags: |
-            neondatabase/neon-test-extensions-${{ matrix.version.pg }}:${{needs.meta.outputs.build-tag}}-${{ matrix.version.debian }}-${{ matrix.arch }}
+            ghcr.io/neondatabase/neon-test-extensions-${{ matrix.version.pg }}:${{needs.meta.outputs.build-tag}}-${{ matrix.version.debian }}-${{ matrix.arch }}

  compute-node-image:
    needs: [ compute-node-image-arch, meta ]
@@ -656,6 +686,7 @@ jobs:
      id-token: write # aws-actions/configure-aws-credentials
      statuses: write
      contents: read
+      packages: write
    runs-on: ubuntu-22.04

    strategy:
@@ -674,28 +705,32 @@ jobs:
    steps:
      - uses: docker/login-action@v3
        with:
-          username: ${{ secrets.NEON_DOCKERHUB_USERNAME }}
-          password: ${{ secrets.NEON_DOCKERHUB_PASSWORD }}
+          registry: ghcr.io
+          username: ${{ github.actor }}
+          password: ${{ secrets.GITHUB_TOKEN }}

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

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

  vm-compute-node-image-arch:
    needs: [ check-permissions, meta, compute-node-image ]
    if: ${{ contains(fromJSON('["push-main", "pr", "compute-rc-pr"]'), needs.meta.outputs.run-kind) }}
-    runs-on: ${{ fromJson(format('["self-hosted", "{0}"]', matrix.arch == 'arm64' && 'large-arm64' || 'large')) }}
+    runs-on: ${{ fromJSON(format('["self-hosted", "{0}"]', matrix.arch == 'arm64' && 'large-arm64' || 'large')) }}
+    permissions:
+      contents: read
+      packages: write
    strategy:
      fail-fast: false
      matrix:
@@ -723,31 +758,34 @@ jobs:
      - uses: neondatabase/dev-actions/set-docker-config-dir@6094485bf440001c94a94a3f9e221e81ff6b6193
      - uses: docker/login-action@v3
        with:
-          username: ${{ secrets.NEON_DOCKERHUB_USERNAME }}
-          password: ${{ secrets.NEON_DOCKERHUB_PASSWORD }}
+          registry: ghcr.io
+          username: ${{ github.actor }}
+          password: ${{ secrets.GITHUB_TOKEN }}

      # Note: we need a separate pull step here because otherwise vm-builder will try to pull, and
      # 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.meta.outputs.build-tag }}
+          docker pull ghcr.io/neondatabase/compute-node-${{ matrix.version.pg }}:${{ needs.meta.outputs.build-tag }}

      - name: Build vm image
        run: |
          ./vm-builder \
            -size=2G \
            -spec=compute/vm-image-spec-${{ matrix.version.debian }}.yaml \
-            -src=neondatabase/compute-node-${{ matrix.version.pg }}:${{ needs.meta.outputs.build-tag }} \
-            -dst=neondatabase/vm-compute-node-${{ matrix.version.pg }}:${{ needs.meta.outputs.build-tag }}-${{ matrix.arch }} \
+            -src=ghcr.io/neondatabase/compute-node-${{ matrix.version.pg }}:${{ needs.meta.outputs.build-tag }} \
+            -dst=ghcr.io/neondatabase/vm-compute-node-${{ matrix.version.pg }}:${{ needs.meta.outputs.build-tag }}-${{ matrix.arch }} \
            -target-arch=linux/${{ matrix.arch }}

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

  vm-compute-node-image:
    needs: [ vm-compute-node-image-arch, meta ]
    if: ${{ contains(fromJSON('["push-main", "pr", "compute-rc-pr"]'), needs.meta.outputs.run-kind) }}
+    permissions:
+      packages: write
    runs-on: ubuntu-22.04
    strategy:
      matrix:
@@ -760,14 +798,15 @@ jobs:
    steps:
      - uses: docker/login-action@v3
        with:
-          username: ${{ secrets.NEON_DOCKERHUB_USERNAME }}
-          password: ${{ secrets.NEON_DOCKERHUB_PASSWORD }}
+          registry: ghcr.io
+          username: ${{ github.actor }}
+          password: ${{ secrets.GITHUB_TOKEN }}

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


  test-images:
@@ -785,18 +824,28 @@ jobs:
        arch: [ x64, arm64 ]
        pg_version: [v16, v17]

-    runs-on: ${{ fromJson(format('["self-hosted", "{0}"]', matrix.arch == 'arm64' && 'small-arm64' || 'small')) }}
+    permissions:
+      packages: read
+
+    runs-on: ${{ fromJSON(format('["self-hosted", "{0}"]', matrix.arch == 'arm64' && 'small-arm64' || 'small')) }}

    steps:
      - uses: actions/checkout@v4

      - uses: neondatabase/dev-actions/set-docker-config-dir@6094485bf440001c94a94a3f9e221e81ff6b6193
+
      - uses: docker/login-action@v3
        with:
          username: ${{ secrets.NEON_DOCKERHUB_USERNAME }}
          password: ${{ secrets.NEON_DOCKERHUB_PASSWORD }}

-      # `neondatabase/neon` contains multiple binaries, all of them use the same input for the version into the same version formatting library.
+      - uses: docker/login-action@v3
+        with:
+          registry: ghcr.io
+          username: ${{ github.actor }}
+          password: ${{ secrets.GITHUB_TOKEN }}
+
+      # `ghcr.io/neondatabase/neon` contains multiple binaries, all of them use the same input for the version into the same version formatting library.
      # Pick pageserver as currently the only binary with extra "version" features printed in the string to verify.
      # Regular pageserver version string looks like
      #   Neon page server git-env:32d14403bd6ab4f4520a94cbfd81a6acef7a526c failpoints: true, features: []
@@ -807,7 +856,7 @@ jobs:
        shell: bash # ensure no set -e for better error messages
        if: ${{ contains(fromJSON('["push-main", "pr", "storage-rc-pr", "proxy-rc-pr"]'), needs.meta.outputs.run-kind) }}
        run: |
-          pageserver_version=$(docker run --rm neondatabase/neon:${{ needs.meta.outputs.build-tag }} "/bin/sh" "-c" "/usr/local/bin/pageserver --version")
+          pageserver_version=$(docker run --rm ghcr.io/neondatabase/neon:${{ needs.meta.outputs.build-tag }} "/bin/sh" "-c" "/usr/local/bin/pageserver --version")

          echo "Pageserver version string: $pageserver_version"

@@ -892,7 +941,7 @@ jobs:
        env:
          SOURCE_TAG: >-
            ${{
-              contains(fromJson('["storage-release", "compute-release", "proxy-release"]'), needs.meta.outputs.run-kind)
+              contains(fromJSON('["storage-release", "compute-release", "proxy-release"]'), needs.meta.outputs.run-kind)
              && needs.meta.outputs.release-pr-run-id
              || needs.meta.outputs.build-tag
            }}
@@ -978,16 +1027,64 @@ jobs:
      acr-registry-name: ${{ vars.AZURE_PROD_REGISTRY_NAME }}
    secrets: inherit

-  # This is a bit of a special case so we're not using a generated image map.
-  add-latest-tag-to-neon-extensions-test-image:
-    if: github.ref_name == 'main'
+  push-neon-test-extensions-image-dockerhub:
+    if: ${{ contains(fromJSON('["push-main", "pr", "compute-rc-pr"]'), needs.meta.outputs.run-kind) }}
    needs: [ meta, compute-node-image ]
    uses: ./.github/workflows/_push-to-container-registry.yml
+    permissions:
+      packages: write
+      id-token: write
    with:
      image-map: |
        {
-          "docker.io/neondatabase/neon-test-extensions-v16:${{ needs.meta.outputs.build-tag }}": ["docker.io/neondatabase/neon-test-extensions-v16:latest"],
-          "docker.io/neondatabase/neon-test-extensions-v17:${{ needs.meta.outputs.build-tag }}": ["docker.io/neondatabase/neon-test-extensions-v17:latest"]
+          "ghcr.io/neondatabase/neon-test-extensions-v16:${{ needs.meta.outputs.build-tag }}": [
+            "docker.io/neondatabase/neon-test-extensions-v16:${{ needs.meta.outputs.build-tag }}"
+          ],
+          "ghcr.io/neondatabase/neon-test-extensions-v17:${{ needs.meta.outputs.build-tag }}": [
+            "docker.io/neondatabase/neon-test-extensions-v17:${{ needs.meta.outputs.build-tag }}"
+          ]
+        }
+    secrets: inherit
+
+  add-latest-tag-to-neon-test-extensions-image:
+    if: ${{ needs.meta.outputs.run-kind == 'push-main' }}
+    needs: [ meta, compute-node-image ]
+    uses: ./.github/workflows/_push-to-container-registry.yml
+    permissions:
+      packages: write
+      id-token: write
+    with:
+      image-map: |
+        {
+          "ghcr.io/neondatabase/neon-test-extensions-v16:${{ needs.meta.outputs.build-tag }}": [
+            "docker.io/neondatabase/neon-test-extensions-v16:latest",
+            "ghcr.io/neondatabase/neon-test-extensions-v16:latest"
+          ],
+          "ghcr.io/neondatabase/neon-test-extensions-v17:${{ needs.meta.outputs.build-tag }}": [
+            "docker.io/neondatabase/neon-test-extensions-v17:latest",
+            "ghcr.io/neondatabase/neon-test-extensions-v17:latest"
+          ]
+        }
+    secrets: inherit
+
+  add-release-tag-to-neon-test-extensions-image:
+    if: ${{ needs.meta.outputs.run-kind == 'compute-release' }}
+    needs: [ meta ]
+    uses: ./.github/workflows/_push-to-container-registry.yml
+    permissions:
+      packages: write
+      id-token: write
+    with:
+      image-map: |
+        {
+          "ghcr.io/neondatabase/neon-test-extensions-v16:${{ needs.meta.outputs.release-pr-run-id }}": [
+            "docker.io/neondatabase/neon-test-extensions-v16:${{ needs.meta.outputs.build-tag }}",
+            "ghcr.io/neondatabase/neon-test-extensions-v16:${{ needs.meta.outputs.build-tag }}"
+          ],
+          "ghcr.io/neondatabase/neon-test-extensions-v17:${{ needs.meta.outputs.release-pr-run-id }}": [
+            "docker.io/neondatabase/neon-test-extensions-v17:${{ needs.meta.outputs.build-tag }}",
+            "ghcr.io/neondatabase/neon-test-extensions-v17:${{ needs.meta.outputs.build-tag }}"
+          ]
        }
    secrets: inherit

@@ -1349,10 +1446,10 @@ jobs:
        if: |
          contains(needs.*.result, 'failure')
          || contains(needs.*.result, 'cancelled')
-          || (needs.check-dependencies-rust.result == 'skipped' && needs.files-changed.outputs.check-rust-dependencies == 'true')
-          || needs.build-and-test-locally.result == 'skipped'
-          || needs.check-codestyle-python.result == 'skipped'
-          || needs.check-codestyle-rust.result == 'skipped'
+          || (needs.check-dependencies-rust.result == 'skipped' && needs.files-changed.outputs.check-rust-dependencies == 'true' && needs.meta.outputs.run-kind == 'pr')
+          || (needs.build-and-test-locally.result == 'skipped' && needs.meta.outputs.run-kind == 'pr')
+          || (needs.check-codestyle-python.result == 'skipped' && needs.meta.outputs.run-kind == 'pr')
+          || (needs.check-codestyle-rust.result == 'skipped' && needs.meta.outputs.run-kind == 'pr')
          || needs.files-changed.result == 'skipped'
          || (needs.push-compute-image-dev.result == 'skipped' && contains(fromJSON('["push-main", "pr", "compute-release", "compute-rc-pr"]'), needs.meta.outputs.run-kind))
          || (needs.push-neon-image-dev.result == 'skipped' && contains(fromJSON('["push-main", "pr", "storage-release", "storage-rc-pr", "proxy-release", "proxy-rc-pr"]'), needs.meta.outputs.run-kind))
--- a/.github/workflows/build_and_test_with_sanitizers.yml
+++ b/.github/workflows/build_and_test_with_sanitizers.yml
@@ -94,8 +94,8 @@ jobs:
    container:
      image: ${{ needs.build-build-tools-image.outputs.image }}-bookworm
      credentials:
-        username: ${{ secrets.NEON_DOCKERHUB_USERNAME }}
-        password: ${{ secrets.NEON_DOCKERHUB_PASSWORD }}
+        username: ${{ github.actor }}
+        password: ${{ secrets.GITHUB_TOKEN }}
      options: --init

    steps:
--- a/.github/workflows/cargo-deny.yml
+++ b/.github/workflows/cargo-deny.yml
@@ -24,11 +24,14 @@ jobs:

    runs-on: [self-hosted, small]

+    permissions:
+      packages: read
+
    container:
-      image: ${{ inputs.build-tools-image || 'neondatabase/build-tools:pinned' }}
+      image: ${{ inputs.build-tools-image || 'ghcr.io/neondatabase/build-tools:pinned' }}
      credentials:
-        username: ${{ secrets.NEON_DOCKERHUB_USERNAME }}
-        password: ${{ secrets.NEON_DOCKERHUB_PASSWORD }}
+        username: ${{ github.actor }}
+        password: ${{ secrets.GITHUB_TOKEN }}
      options: --init

    steps:
--- a/.github/workflows/cloud-regress.yml
+++ b/.github/workflows/cloud-regress.yml
@@ -37,10 +37,10 @@ jobs:

    runs-on: us-east-2
    container:
-      image: neondatabase/build-tools:pinned-bookworm
+      image: ghcr.io/neondatabase/build-tools:pinned-bookworm
      credentials:
-        username: ${{ secrets.NEON_DOCKERHUB_USERNAME }}
-        password: ${{ secrets.NEON_DOCKERHUB_PASSWORD }}
+        username: ${{ github.actor }}
+        password: ${{ secrets.GITHUB_TOKEN }}
      options: --init

    steps:
--- a/.github/workflows/ingest_benchmark.yml
+++ b/.github/workflows/ingest_benchmark.yml
@@ -67,10 +67,10 @@ jobs:
      PGCOPYDB_LIB_PATH: /pgcopydb/lib
    runs-on: [ self-hosted, us-east-2, x64 ]
    container:
-      image: neondatabase/build-tools:pinned-bookworm
+      image: ghcr.io/neondatabase/build-tools:pinned-bookworm
      credentials:
-        username: ${{ secrets.NEON_DOCKERHUB_USERNAME }}
-        password: ${{ secrets.NEON_DOCKERHUB_PASSWORD }}
+        username: ${{ github.actor }}
+        password: ${{ secrets.GITHUB_TOKEN }}
      options: --init
    timeout-minutes: 1440

--- a/.github/workflows/large_oltp_benchmark.yml
+++ b/.github/workflows/large_oltp_benchmark.yml
@@ -2,8 +2,8 @@ name: large oltp benchmark

 on:
  # uncomment to run on push for debugging your PR
-  push:
-    branches: [ bodobolero/synthetic_oltp_workload ]
+  #push:
+  #  branches: [ bodobolero/synthetic_oltp_workload ]

  schedule:
    # * is a special character in YAML so you have to quote this string
@@ -12,7 +12,7 @@ on:
    #          │ │  ┌───────────── day of the month (1 - 31)
    #          │ │  │ ┌───────────── month (1 - 12 or JAN-DEC)
    #          │ │  │ │ ┌───────────── day of the week (0 - 6 or SUN-SAT)
-    - cron:   '0 15 * * *' # run once a day, timezone is utc, avoid conflict with other benchmarks
+    - cron:   '0 15 * * 0,2,4' # run on Sunday, Tuesday, Thursday at 3 PM UTC
  workflow_dispatch: # adds ability to run this manually

 defaults:
@@ -22,7 +22,7 @@ defaults:
 concurrency:
  # Allow only one workflow globally because we need dedicated resources which only exist once
  group: large-oltp-bench-workflow
-  cancel-in-progress: true
+  cancel-in-progress: false

 jobs:
  oltp:
@@ -31,9 +31,9 @@ jobs:
      matrix:
        include:
          - target: new_branch 
-            custom_scripts: insert_webhooks.sql@2 select_any_webhook_with_skew.sql@4 select_recent_webhook.sql@4 
+            custom_scripts: insert_webhooks.sql@200 select_any_webhook_with_skew.sql@300 select_recent_webhook.sql@397 select_prefetch_webhook.sql@3 IUD_one_transaction.sql@100
          - target: reuse_branch 
-            custom_scripts: insert_webhooks.sql@2 select_any_webhook_with_skew.sql@4 select_recent_webhook.sql@4 
+            custom_scripts: insert_webhooks.sql@200 select_any_webhook_with_skew.sql@300 select_recent_webhook.sql@397 select_prefetch_webhook.sql@3 IUD_one_transaction.sql@100
      max-parallel: 1 # we want to run each stripe size sequentially to be able to compare the results
    permissions:
      contents: write
@@ -46,19 +46,20 @@ jobs:
      PG_VERSION: 16 # pre-determined by pre-determined project
      TEST_OUTPUT: /tmp/test_output
      BUILD_TYPE: remote
-      SAVE_PERF_REPORT: ${{ github.ref_name == 'main' }}
      PLATFORM: ${{ matrix.target }}

    runs-on: [ self-hosted, us-east-2, x64 ]
    container:
-      image: neondatabase/build-tools:pinned-bookworm
+      image: ghcr.io/neondatabase/build-tools:pinned-bookworm
      credentials:
-        username: ${{ secrets.NEON_DOCKERHUB_USERNAME }}
-        password: ${{ secrets.NEON_DOCKERHUB_PASSWORD }}
+        username: ${{ github.actor }}
+        password: ${{ secrets.GITHUB_TOKEN }}
      options: --init

-    # Increase timeout to 8h, default timeout is 6h
-    timeout-minutes: 480
+    # Increase timeout to 2 days, default timeout is 6h - database maintenance can take a long time
+    # (normally 1h pgbench, 3h vacuum analyze 3.5h re-index) x 2 = 15h, leave some buffer for regressions
+    # in one run vacuum didn't finish within 12 hours
+    timeout-minutes: 2880

    steps:
    - uses: actions/checkout@v4
@@ -89,29 +90,45 @@ jobs:
    - name: Set up Connection String
      id: set-up-connstr
      run: |
-          case "${{ matrix.target }}" in
-              new_branch)
-              CONNSTR=${{ steps.create-neon-branch-oltp-target.outputs.dsn }}
-              ;;
-              reuse_branch)
-              CONNSTR=${{ secrets.BENCHMARK_LARGE_OLTP_REUSE_CONNSTR }}
-              ;;
-              *)
-              echo >&2 "Unknown target=${{ matrix.target }}"
-              exit 1
-              ;;
-          esac
+        case "${{ matrix.target }}" in
+          new_branch)
+          CONNSTR=${{ steps.create-neon-branch-oltp-target.outputs.dsn }}
+          ;;
+          reuse_branch)
+          CONNSTR=${{ secrets.BENCHMARK_LARGE_OLTP_REUSE_CONNSTR }}
+          ;;
+          *)
+          echo >&2 "Unknown target=${{ matrix.target }}"
+          exit 1
+          ;;
+        esac

-          echo "connstr=${CONNSTR}" >> $GITHUB_OUTPUT
+        CONNSTR_WITHOUT_POOLER="${CONNSTR//-pooler/}"

-    - name: Benchmark pgbench with custom-scripts
+        echo "connstr=${CONNSTR}" >> $GITHUB_OUTPUT
+        echo "connstr_without_pooler=${CONNSTR_WITHOUT_POOLER}" >> $GITHUB_OUTPUT
+
+    - name: Delete rows from prior runs in reuse branch
+      if: ${{ matrix.target == 'reuse_branch' }}
+      env:
+          BENCHMARK_CONNSTR: ${{ steps.set-up-connstr.outputs.connstr_without_pooler }}
+          PG_CONFIG: /tmp/neon/pg_install/v16/bin/pg_config
+          PSQL: /tmp/neon/pg_install/v16/bin/psql
+          PG_16_LIB_PATH: /tmp/neon/pg_install/v16/lib
+      run: |
+        echo "$(date '+%Y-%m-%d %H:%M:%S') - Deleting rows in table webhook.incoming_webhooks from prior runs"
+        export LD_LIBRARY_PATH=${PG_16_LIB_PATH}
+        ${PSQL} "${BENCHMARK_CONNSTR}" -c "SET statement_timeout = 0; DELETE FROM webhook.incoming_webhooks WHERE created_at > '2025-02-27 23:59:59+00';"
+        echo "$(date '+%Y-%m-%d %H:%M:%S') - Finished deleting rows in table webhook.incoming_webhooks from prior runs"
+
+    - name: Benchmark pgbench with custom-scripts 
      uses: ./.github/actions/run-python-test-set
      with:
        build_type: ${{ env.BUILD_TYPE }}
        test_selection: performance
        run_in_parallel: false
-        save_perf_report: ${{ env.SAVE_PERF_REPORT }}
-        extra_params: -m remote_cluster --timeout 21600 -k test_perf_oltp_large_tenant
+        save_perf_report: true
+        extra_params: -m remote_cluster --timeout 7200 -k test_perf_oltp_large_tenant_pgbench
        pg_version: ${{ env.PG_VERSION }}
        aws-oicd-role-arn: ${{ vars.DEV_AWS_OIDC_ROLE_ARN }}
      env:
@@ -119,6 +136,21 @@ jobs:
        VIP_VAP_ACCESS_TOKEN: "${{ secrets.VIP_VAP_ACCESS_TOKEN }}"
        PERF_TEST_RESULT_CONNSTR: "${{ secrets.PERF_TEST_RESULT_CONNSTR }}"

+    - name: Benchmark database maintenance
+      uses: ./.github/actions/run-python-test-set
+      with:
+        build_type: ${{ env.BUILD_TYPE }}
+        test_selection: performance
+        run_in_parallel: false
+        save_perf_report: true
+        extra_params: -m remote_cluster --timeout 172800 -k test_perf_oltp_large_tenant_maintenance
+        pg_version: ${{ env.PG_VERSION }}
+        aws-oicd-role-arn: ${{ vars.DEV_AWS_OIDC_ROLE_ARN }}
+      env:
+        BENCHMARK_CONNSTR: ${{ steps.set-up-connstr.outputs.connstr_without_pooler }}
+        VIP_VAP_ACCESS_TOKEN: "${{ secrets.VIP_VAP_ACCESS_TOKEN }}"
+        PERF_TEST_RESULT_CONNSTR: "${{ secrets.PERF_TEST_RESULT_CONNSTR }}"
+
    - name: Delete Neon Branch for large tenant
      if: ${{ always() && matrix.target == 'new_branch' }}
      uses: ./.github/actions/neon-branch-delete
@@ -127,6 +159,13 @@ jobs:
        branch_id: ${{ steps.create-neon-branch-oltp-target.outputs.branch_id }}
        api_key: ${{ secrets.NEON_STAGING_API_KEY }}

+    - name: Configure AWS credentials # again because prior steps could have exceeded 5 hours
+      uses: aws-actions/configure-aws-credentials@v4
+      with:
+        aws-region: eu-central-1
+        role-to-assume: ${{ vars.DEV_AWS_OIDC_ROLE_ARN }}
+        role-duration-seconds: 18000 # 5 hours
+
    - name: Create Allure report
      id: create-allure-report
      if: ${{ !cancelled() }}
--- a/.github/workflows/neon_extra_builds.yml
+++ b/.github/workflows/neon_extra_builds.yml
@@ -71,8 +71,8 @@ jobs:
    uses: ./.github/workflows/build-macos.yml
    with:
      pg_versions: ${{ needs.files-changed.outputs.postgres_changes }}
-      rebuild_rust_code: ${{ fromJson(needs.files-changed.outputs.rebuild_rust_code) }}
-      rebuild_everything: ${{ fromJson(needs.files-changed.outputs.rebuild_everything) }}
+      rebuild_rust_code: ${{ fromJSON(needs.files-changed.outputs.rebuild_rust_code) }}
+      rebuild_everything: ${{ fromJSON(needs.files-changed.outputs.rebuild_everything) }}

  gather-rust-build-stats:
    needs: [ check-permissions, build-build-tools-image, files-changed ]
@@ -90,8 +90,8 @@ jobs:
    container:
      image: ${{ needs.build-build-tools-image.outputs.image }}-bookworm
      credentials:
-        username: ${{ secrets.NEON_DOCKERHUB_USERNAME }}
-        password: ${{ secrets.NEON_DOCKERHUB_PASSWORD }}
+        username: ${{ github.actor }}
+        password: ${{ secrets.GITHUB_TOKEN }}
      options: --init

    env:
--- a/.github/workflows/periodic_pagebench.yml
+++ b/.github/workflows/periodic_pagebench.yml
@@ -34,10 +34,10 @@ jobs:
      pull-requests: write
    runs-on: [ self-hosted, small ]
    container:
-      image: neondatabase/build-tools:pinned-bookworm
+      image: ghcr.io/neondatabase/build-tools:pinned-bookworm
      credentials:
-        username: ${{ secrets.NEON_DOCKERHUB_USERNAME }}
-        password: ${{ secrets.NEON_DOCKERHUB_PASSWORD }}
+        username: ${{ github.actor }}
+        password: ${{ secrets.GITHUB_TOKEN }}
      options: --init
    timeout-minutes: 360  # Set the timeout to 6 hours
    env:
--- a/.github/workflows/pg-clients.yml
+++ b/.github/workflows/pg-clients.yml
@@ -53,8 +53,8 @@ jobs:
    container:
      image: ${{ needs.build-build-tools-image.outputs.image }}-bookworm
      credentials:
-        username: ${{ secrets.NEON_DOCKERHUB_USERNAME }}
-        password: ${{ secrets.NEON_DOCKERHUB_PASSWORD }}
+        username: ${{ github.actor }}
+        password: ${{ secrets.GITHUB_TOKEN }}
      options: --init --user root
    services:
      clickhouse:
@@ -153,8 +153,8 @@ jobs:
    container:
      image: ${{ needs.build-build-tools-image.outputs.image }}-bookworm
      credentials:
-        username: ${{ secrets.NEON_DOCKERHUB_USERNAME }}
-        password: ${{ secrets.NEON_DOCKERHUB_PASSWORD }}
+        username: ${{ github.actor }}
+        password: ${{ secrets.GITHUB_TOKEN }}
      options: --init --user root

    steps:
--- a/.github/workflows/pin-build-tools-image.yml
+++ b/.github/workflows/pin-build-tools-image.yml
@@ -46,8 +46,8 @@ jobs:
          FROM_TAG: ${{ inputs.from-tag }}
          TO_TAG: pinned
        run: |
-          docker manifest inspect "docker.io/neondatabase/build-tools:${FROM_TAG}" > "${FROM_TAG}.json"
-          docker manifest inspect "docker.io/neondatabase/build-tools:${TO_TAG}"   > "${TO_TAG}.json"
+          docker manifest inspect "ghcr.io/neondatabase/build-tools:${FROM_TAG}" > "${FROM_TAG}.json"
+          docker manifest inspect "ghcr.io/neondatabase/build-tools:${TO_TAG}"   > "${TO_TAG}.json"

          if diff "${FROM_TAG}.json" "${TO_TAG}.json"; then
            skip=true
@@ -71,13 +71,13 @@ jobs:
    with:
      image-map: |
        {
-          "docker.io/neondatabase/build-tools:${{ inputs.from-tag }}-bullseye": [
+          "ghcr.io/neondatabase/build-tools:${{ inputs.from-tag }}-bullseye": [
            "docker.io/neondatabase/build-tools:pinned-bullseye",
            "ghcr.io/neondatabase/build-tools:pinned-bullseye",
            "${{ vars.NEON_DEV_AWS_ACCOUNT_ID }}.dkr.ecr.${{ vars.AWS_ECR_REGION }}.amazonaws.com/build-tools:pinned-bullseye",
            "${{ vars.AZURE_DEV_REGISTRY_NAME }}.azurecr.io/neondatabase/build-tools:pinned-bullseye"
          ],
-          "docker.io/neondatabase/build-tools:${{ inputs.from-tag }}-bookworm": [
+          "ghcr.io/neondatabase/build-tools:${{ inputs.from-tag }}-bookworm": [
            "docker.io/neondatabase/build-tools:pinned-bookworm",
            "docker.io/neondatabase/build-tools:pinned",
            "ghcr.io/neondatabase/build-tools:pinned-bookworm",
--- a/.github/workflows/pre-merge-checks.yml
+++ b/.github/workflows/pre-merge-checks.yml
@@ -19,6 +19,8 @@ permissions: {}
 jobs:
  meta:
    runs-on: ubuntu-22.04
+    permissions:
+      contents: read
    outputs:
      python-changed: ${{ steps.python-src.outputs.any_changed }}
      rust-changed: ${{ steps.rust-src.outputs.any_changed }}
@@ -27,7 +29,7 @@ jobs:
    steps:
      - uses: actions/checkout@v4

-      - uses: tj-actions/changed-files@4edd678ac3f81e2dc578756871e4d00c19191daf # v45.0.4
+      - uses: step-security/changed-files@3dbe17c78367e7d60f00d78ae6781a35be47b4a1 # v45.0.1
        id: python-src
        with:
          files: |
@@ -38,7 +40,7 @@ jobs:
            poetry.lock
            pyproject.toml

-      - uses: tj-actions/changed-files@4edd678ac3f81e2dc578756871e4d00c19191daf # v45.0.4
+      - uses: step-security/changed-files@3dbe17c78367e7d60f00d78ae6781a35be47b4a1 # v45.0.1
        id: rust-src
        with:
          files: |
@@ -72,6 +74,9 @@ jobs:
      || needs.meta.outputs.python-changed == 'true'
      || needs.meta.outputs.rust-changed == 'true'
    needs: [ meta ]
+    permissions:
+      contents: read
+      packages: write
    uses: ./.github/workflows/build-build-tools-image.yml
    with:
      # Build only one combination to save time
@@ -82,6 +87,9 @@ jobs:
  check-codestyle-python:
    if: needs.meta.outputs.python-changed == 'true'
    needs: [ meta, build-build-tools-image ]
+    permissions:
+      contents: read
+      packages: read
    uses: ./.github/workflows/_check-codestyle-python.yml
    with:
      # `-bookworm-x64` suffix should match the combination in `build-build-tools-image`
@@ -91,6 +99,9 @@ jobs:
  check-codestyle-rust:
    if: needs.meta.outputs.rust-changed == 'true'
    needs: [ meta, build-build-tools-image ]
+    permissions:
+      contents: read
+      packages: read
    uses: ./.github/workflows/_check-codestyle-rust.yml
    with:
      # `-bookworm-x64` suffix should match the combination in `build-build-tools-image`
@@ -148,7 +159,7 @@ jobs:
          ${{
            always()
            && github.event_name == 'merge_group'
-            && contains(fromJson('["release", "release-proxy", "release-compute"]'), github.base_ref)
+            && contains(fromJSON('["release", "release-proxy", "release-compute"]'), needs.meta.outputs.branch)
          }}
        env:
          GH_TOKEN: ${{ secrets.CI_ACCESS_TOKEN }}
--- a/Cargo.lock
+++ b/Cargo.lock
@@ -167,45 +167,6 @@ version = "0.7.6"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "7c02d123df017efcdfbd739ef81735b36c5ba83ec3c59c80a9d7ecc718f92e50"

-[[package]]
-name = "asn1-rs"
-version = "0.6.2"
-source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "5493c3bedbacf7fd7382c6346bbd66687d12bbaad3a89a2d2c303ee6cf20b048"
-dependencies = [
- "asn1-rs-derive",
- "asn1-rs-impl",
- "displaydoc",
- "nom",
- "num-traits",
- "rusticata-macros",
- "thiserror 1.0.69",
- "time",
-]
-
-[[package]]
-name = "asn1-rs-derive"
-version = "0.5.1"
-source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "965c2d33e53cb6b267e148a4cb0760bc01f4904c1cd4bb4002a085bb016d1490"
-dependencies = [
- "proc-macro2",
- "quote",
- "syn 2.0.100",
- "synstructure",
-]
-
-[[package]]
-name = "asn1-rs-impl"
-version = "0.2.0"
-source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "7b18050c2cd6fe86c3a76584ef5e0baf286d038cda203eb6223df2cc413565f7"
-dependencies = [
- "proc-macro2",
- "quote",
- "syn 2.0.100",
-]
-
 [[package]]
 name = "assert-json-diff"
 version = "2.0.2"
@@ -1309,6 +1270,7 @@ version = "0.1.0"
 dependencies = [
 "anyhow",
 "chrono",
+ "indexmap 2.0.1",
 "jsonwebtoken",
 "regex",
 "remote_storage",
@@ -1339,6 +1301,7 @@ dependencies = [
 "flate2",
 "futures",
 "http 1.1.0",
+ "indexmap 2.0.1",
 "jsonwebtoken",
 "metrics",
 "nix 0.27.1",
@@ -1347,17 +1310,20 @@ dependencies = [
 "once_cell",
 "opentelemetry",
 "opentelemetry_sdk",
+ "p256 0.13.2",
 "postgres",
 "postgres_initdb",
 "regex",
 "remote_storage",
 "reqwest",
+ "ring",
 "rlimit",
 "rust-ini",
 "serde",
 "serde_json",
 "serde_with",
 "signal-hook",
+ "spki 0.7.3",
 "tar",
 "thiserror 1.0.69",
 "tokio",
@@ -1377,6 +1343,7 @@ dependencies = [
 "vm_monitor",
 "walkdir",
 "workspace_hack",
+ "x509-cert",
 "zstd",
 ]

@@ -1801,22 +1768,21 @@ source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "fffa369a668c8af7dbf8b5e56c9f744fbd399949ed171606040001947de40b1c"
 dependencies = [
 "const-oid",
+ "der_derive",
+ "flagset",
 "pem-rfc7468",
 "zeroize",
 ]

 [[package]]
-name = "der-parser"
-version = "9.0.0"
+name = "der_derive"
+version = "0.7.3"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "5cd0a5c643689626bec213c4d8bd4d96acc8ffdb4ad4bb6bc16abf27d5f4b553"
+checksum = "8034092389675178f570469e6c3b0465d3d30b4505c294a6550db47f3c17ad18"
 dependencies = [
- "asn1-rs",
- "displaydoc",
- "nom",
- "num-bigint",
- "num-traits",
- "rusticata-macros",
+ "proc-macro2",
+ "quote",
+ "syn 2.0.100",
 ]

 [[package]]
@@ -2282,6 +2248,12 @@ version = "0.4.2"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "0ce7134b9999ecaf8bcd65542e436736ef32ddca1b3e06094cb6ec5755203b80"

+[[package]]
+name = "flagset"
+version = "0.4.6"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "b3ea1ec5f8307826a5b71094dd91fc04d4ae75d5709b20ad351c7fb4815c86ec"
+
 [[package]]
 name = "flate2"
 version = "1.0.26"
@@ -2838,6 +2810,7 @@ version = "0.1.0"
 dependencies = [
 "anyhow",
 "bytes",
+ "camino",
 "fail",
 "futures",
 "hyper 0.14.30",
@@ -2848,6 +2821,7 @@ dependencies = [
 "pprof",
 "regex",
 "routerify",
+ "rustls-pemfile 2.1.1",
 "serde",
 "serde_json",
 "serde_path_to_error",
@@ -2877,9 +2851,9 @@ checksum = "c4a1e36c821dbe04574f602848a19f742f4fb3c98d40449f11bcad18d6b17421"

 [[package]]
 name = "humantime"
-version = "2.1.0"
+version = "2.2.0"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "9a3a5bfb195931eeb336b2a7b4d761daec841b97f947d34394601737a7bba5e4"
+checksum = "9b112acc8b3adf4b107a8ec20977da0273a8c386765a3ec0229bd500a1443f9f"

 [[package]]
 name = "humantime-serde"
@@ -4017,15 +3991,6 @@ dependencies = [
 "memchr",
 ]

-[[package]]
-name = "oid-registry"
-version = "0.7.1"
-source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "a8d8034d9489cdaf79228eb9f6a3b8d7bb32ba00d6645ebd48eef4077ceb5bd9"
-dependencies = [
- "asn1-rs",
-]
-
 [[package]]
 name = "once_cell"
 version = "1.20.2"
@@ -4302,8 +4267,6 @@ dependencies = [
 "reqwest",
 "rpds",
 "rustls 0.23.18",
- "rustls-pemfile 2.1.1",
- "rustls-pki-types",
 "scopeguard",
 "send-future",
 "serde",
@@ -5202,7 +5165,7 @@ dependencies = [
 "uuid",
 "walkdir",
 "workspace_hack",
- "x509-parser",
+ "x509-cert",
 "zerocopy",
 ]

@@ -5823,15 +5786,6 @@ dependencies = [
 "semver",
 ]

-[[package]]
-name = "rusticata-macros"
-version = "4.1.0"
-source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "faf0c4a6ece9950b9abdb62b1cfcf2a68b3b67a10ba445b3bb85be2a293d0632"
-dependencies = [
- "nom",
-]
-
 [[package]]
 name = "rustix"
 version = "0.38.41"
@@ -6019,6 +5973,7 @@ dependencies = [
 "regex",
 "remote_storage",
 "reqwest",
+ "rustls 0.23.18",
 "safekeeper_api",
 "safekeeper_client",
 "scopeguard",
@@ -6035,6 +5990,7 @@ dependencies = [
 "tokio",
 "tokio-io-timeout",
 "tokio-postgres",
+ "tokio-rustls 0.26.0",
 "tokio-stream",
 "tokio-tar",
 "tokio-util",
@@ -6425,9 +6381,9 @@ dependencies = [

 [[package]]
 name = "sha1"
-version = "0.10.5"
+version = "0.10.6"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "f04293dc80c3993519f2d7f6f511707ee7094fe0c6d3406feb330cdb3540eba3"
+checksum = "e3bf829a2d51ab4a5ddf1352d8470c140cadc8301b2ae1789db023f01cedd6ba"
 dependencies = [
 "cfg-if",
 "cpufeatures",
@@ -7135,6 +7091,27 @@ version = "0.1.1"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "1f3ccbac311fea05f86f61904b462b55fb3df8837a366dfc601a0161d0532f20"

+[[package]]
+name = "tls_codec"
+version = "0.4.2"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "0de2e01245e2bb89d6f05801c564fa27624dbd7b1846859876c7dad82e90bf6b"
+dependencies = [
+ "tls_codec_derive",
+ "zeroize",
+]
+
+[[package]]
+name = "tls_codec_derive"
+version = "0.4.2"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "2d2e76690929402faae40aebdda620a2c0e25dd6d3b9afe48867dfd95991f4bd"
+dependencies = [
+ "proc-macro2",
+ "quote",
+ "syn 2.0.100",
+]
+
 [[package]]
 name = "tokio"
 version = "1.43.0"
@@ -8387,12 +8364,14 @@ dependencies = [
 "chrono",
 "clap",
 "clap_builder",
+ "const-oid",
 "crypto-bigint 0.5.5",
 "der 0.7.8",
 "deranged",
 "digest",
- "displaydoc",
+ "ecdsa 0.16.9",
 "either",
+ "elliptic-curve 0.13.8",
 "env_filter",
 "env_logger",
 "fail",
@@ -8427,6 +8406,7 @@ dependencies = [
 "num-rational",
 "num-traits",
 "once_cell",
+ "p256 0.13.2",
 "parquet",
 "prettyplease",
 "proc-macro2",
@@ -8439,6 +8419,7 @@ dependencies = [
 "reqwest",
 "rustls 0.23.18",
 "scopeguard",
+ "sec1 0.7.3",
 "serde",
 "serde_json",
 "sha2",
@@ -8484,6 +8465,18 @@ version = "0.5.5"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "1e9df38ee2d2c3c5948ea468a8406ff0db0b29ae1ffde1bcf20ef305bcc95c51"

+[[package]]
+name = "x509-cert"
+version = "0.2.5"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "1301e935010a701ae5f8655edc0ad17c44bad3ac5ce8c39185f75453b720ae94"
+dependencies = [
+ "const-oid",
+ "der 0.7.8",
+ "spki 0.7.3",
+ "tls_codec",
+]
+
 [[package]]
 name = "x509-certificate"
 version = "0.23.1"
@@ -8503,23 +8496,6 @@ dependencies = [
 "zeroize",
 ]

-[[package]]
-name = "x509-parser"
-version = "0.16.0"
-source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "fcbc162f30700d6f3f82a24bf7cc62ffe7caea42c0b2cba8bf7f3ae50cf51f69"
-dependencies = [
- "asn1-rs",
- "data-encoding",
- "der-parser",
- "lazy_static",
- "nom",
- "oid-registry",
- "rusticata-macros",
- "thiserror 1.0.69",
- "time",
-]
-
 [[package]]
 name = "xattr"
 version = "1.0.0"
@@ -8612,9 +8588,9 @@ dependencies = [

 [[package]]
 name = "zeroize"
-version = "1.7.0"
+version = "1.8.1"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "525b4ec142c6b68a2d10f01f7bbf6755599ca3f81ea53b8431b7dd348f5fdb2d"
+checksum = "ced3678a2879b30306d323f4542626697a464a97c0a07c9aebf7ebca65cd4dde"
 dependencies = [
 "serde",
 "zeroize_derive",
--- a/Cargo.toml
+++ b/Cargo.toml
@@ -106,13 +106,13 @@ hostname = "0.4"
 http = {version = "1.1.0", features = ["std"]}
 http-types = { version = "2", default-features = false }
 http-body-util = "0.1.2"
-humantime = "2.1"
+humantime = "2.2"
 humantime-serde = "1.1.1"
 hyper0 = { package = "hyper", version = "0.14" }
 hyper = "1.4"
 hyper-util = "0.1"
 tokio-tungstenite = "0.21.0"
-indexmap = "2"
+indexmap = { version = "2", features = ["serde"] }
 indoc = "2"
 ipnet = "2.10.0"
 itertools = "0.10"
@@ -215,10 +215,10 @@ urlencoding = "2.1"
 uuid = { version = "1.6.1", features = ["v4", "v7", "serde"] }
 walkdir = "2.3.2"
 rustls-native-certs = "0.8"
-x509-parser = "0.16"
 whoami = "1.5.1"
 zerocopy = { version = "0.7", features = ["derive"] }
 json-structural-diff = { version = "0.2.0" }
+x509-cert = { version = "0.2.5" }

 ## TODO replace this with tracing
 env_logger = "0.11"
--- a/2
+++ b/2
@@ -2,7 +2,7 @@
 ### The image itself is mainly used as a container for the binaries and for starting e2e tests with custom parameters.
 ### By default, the binaries inside the image have some mock parameters and can start, but are not intended to be used
 ### inside this image in the real deployments.
-ARG REPOSITORY=neondatabase
+ARG REPOSITORY=ghcr.io/neondatabase
 ARG IMAGE=build-tools
 ARG TAG=pinned
 ARG DEFAULT_PG_VERSION=17
--- a/compute/compute-node.Dockerfile
+++ b/compute/compute-node.Dockerfile
@@ -77,7 +77,7 @@
 # build_and_test.yml github workflow for how that's done.

 ARG PG_VERSION
-ARG REPOSITORY=neondatabase
+ARG REPOSITORY=ghcr.io/neondatabase
 ARG IMAGE=build-tools
 ARG TAG=pinned
 ARG BUILD_TAG
@@ -1735,6 +1735,8 @@ RUN set -e \
        libevent-dev \
        libtool \
        pkg-config \
+        libcurl4-openssl-dev \
+        libssl-dev \
    && apt clean && rm -rf /var/lib/apt/lists/*

 # Use `dist_man_MANS=` to skip manpage generation (which requires python3/pandoc)
@@ -1743,7 +1745,7 @@ RUN set -e \
    && git clone --recurse-submodules --depth 1 --branch ${PGBOUNCER_TAG} https://github.com/pgbouncer/pgbouncer.git pgbouncer \
    && cd pgbouncer \
    && ./autogen.sh \
-    && ./configure --prefix=/usr/local/pgbouncer --without-openssl \
+    && ./configure --prefix=/usr/local/pgbouncer \
    && make -j $(nproc) dist_man_MANS= \
    && make install dist_man_MANS=

--- a/compute/vm-image-spec-bookworm.yaml
+++ b/compute/vm-image-spec-bookworm.yaml
@@ -39,6 +39,13 @@ commands:
    user: nobody
    sysvInitAction: respawn
    shell: '/bin/sql_exporter -config.file=/etc/sql_exporter_autoscaling.yml -web.listen-address=:9499'
+  # Rsyslog by default creates a unix socket under /dev/log . That's where Postgres sends logs also.
+  # We run syslog with postgres user so it can't create /dev/log. Instead we configure rsyslog to
+  # use a different path for the socket. The symlink actually points to our custom path.
+  - name: rsyslogd-socket-symlink
+    user: root
+    sysvInitAction: sysinit
+    shell: "ln -s /var/db/postgres/rsyslogpipe /dev/log"
  - name: rsyslogd
    user: postgres
    sysvInitAction: respawn
@@ -77,6 +84,9 @@ files:
 # compute_ctl will rewrite this file with the actual configuration, if needed.
  - filename: compute_rsyslog.conf
    content: |
+      # Syslock.Name specifies a non-default pipe location that is writeable for the postgres user.
+      module(load="imuxsock" SysSock.Name="/var/db/postgres/rsyslogpipe") # provides support for local system logging
+
      *.*    /dev/null
      $IncludeConfig /etc/rsyslog.d/*.conf
 build: |
@@ -145,7 +155,7 @@ merge: |

  COPY compute_rsyslog.conf /etc/compute_rsyslog.conf
  RUN chmod 0666 /etc/compute_rsyslog.conf
-  RUN chmod 0666 /var/log/
+  RUN mkdir /var/log/rsyslog && chown -R postgres /var/log/rsyslog


  COPY --from=libcgroup-builder /libcgroup-install/bin/*  /usr/bin/
--- a/compute/vm-image-spec-bullseye.yaml
+++ b/compute/vm-image-spec-bullseye.yaml
@@ -39,6 +39,13 @@ commands:
    user: nobody
    sysvInitAction: respawn
    shell: '/bin/sql_exporter -config.file=/etc/sql_exporter_autoscaling.yml -web.listen-address=:9499'
+  # Rsyslog by default creates a unix socket under /dev/log . That's where Postgres sends logs also.
+  # We run syslog with postgres user so it can't create /dev/log. Instead we configure rsyslog to
+  # use a different path for the socket. The symlink actually points to our custom path.
+  - name: rsyslogd-socket-symlink
+    user: root
+    sysvInitAction: sysinit
+    shell: "ln -s /var/db/postgres/rsyslogpipe /dev/log"
  - name: rsyslogd
    user: postgres
    sysvInitAction: respawn
@@ -77,6 +84,9 @@ files:
 # compute_ctl will rewrite this file with the actual configuration, if needed.
  - filename: compute_rsyslog.conf
    content: |
+      # Syslock.Name specifies a non-default pipe location that is writeable for the postgres user.
+      module(load="imuxsock" SysSock.Name="/var/db/postgres/rsyslogpipe") # provides support for local system logging
+
      *.*    /dev/null
      $IncludeConfig /etc/rsyslog.d/*.conf
 build: |
@@ -140,7 +150,7 @@ merge: |

  COPY compute_rsyslog.conf /etc/compute_rsyslog.conf
  RUN chmod 0666 /etc/compute_rsyslog.conf
-  RUN chmod 0666 /var/log/
+  RUN mkdir /var/log/rsyslog && chown -R postgres /var/log/rsyslog


  COPY --from=libcgroup-builder /libcgroup-install/bin/*  /usr/bin/
--- a/compute_tools/Cargo.toml
+++ b/compute_tools/Cargo.toml
@@ -26,6 +26,7 @@ fail.workspace = true
 flate2.workspace = true
 futures.workspace = true
 http.workspace = true
+indexmap.workspace = true
 jsonwebtoken.workspace = true
 metrics.workspace = true
 nix.workspace = true
@@ -34,16 +35,19 @@ num_cpus.workspace = true
 once_cell.workspace = true
 opentelemetry.workspace = true
 opentelemetry_sdk.workspace = true
+p256 = { version = "0.13", features = ["pem"] }
 postgres.workspace = true
 regex.workspace = true
+reqwest = { workspace = true, features = ["json"] }
+ring = "0.17"
 serde.workspace = true
 serde_with.workspace = true
 serde_json.workspace = true
 signal-hook.workspace = true
+spki = { version = "0.7.3", features = ["std"] }
 tar.workspace = true
 tower.workspace = true
 tower-http.workspace = true
-reqwest = { workspace = true, features = ["json"] }
 tokio = { workspace = true, features = ["rt", "rt-multi-thread"] }
 tokio-postgres.workspace = true
 tokio-util.workspace = true
@@ -57,6 +61,7 @@ thiserror.workspace = true
 url.workspace = true
 uuid.workspace = true
 walkdir.workspace = true
+x509-cert.workspace = true

 postgres_initdb.workspace = true
 compute_api.workspace = true
--- a/compute_tools/src/compute.rs
+++ b/compute_tools/src/compute.rs
@@ -37,10 +37,14 @@ use crate::logger::startup_context_from_env;
 use crate::lsn_lease::launch_lsn_lease_bg_task_for_static;
 use crate::monitor::launch_monitor;
 use crate::pg_helpers::*;
-use crate::rsyslog::configure_audit_rsyslog;
+use crate::rsyslog::{
+    PostgresLogsRsyslogConfig, configure_audit_rsyslog, configure_postgres_logs_export,
+    launch_pgaudit_gc,
+};
 use crate::spec::*;
 use crate::swap::resize_swap;
 use crate::sync_sk::{check_if_synced, ping_safekeeper};
+use crate::tls::watch_cert_for_changes;
 use crate::{config, extension_server, local_proxy};

 pub static SYNC_SAFEKEEPERS_PID: AtomicU32 = AtomicU32::new(0);
@@ -112,6 +116,7 @@ pub struct ComputeNode {

    // key: ext_archive_name, value: started download time, download_completed?
    pub ext_download_progress: RwLock<HashMap<String, (DateTime<Utc>, bool)>>,
+    pub compute_ctl_config: ComputeCtlConfig,
 }

 // store some metrics about download size that might impact startup time
@@ -135,8 +140,6 @@ pub struct ComputeState {
    /// passed by the control plane with a /configure HTTP request.
    pub pspec: Option<ParsedSpec>,

-    pub compute_ctl_config: ComputeCtlConfig,
-
    /// If the spec is passed by a /configure request, 'startup_span' is the
    /// /configure request's tracing span. The main thread enters it when it
    /// processes the compute startup, so that the compute startup is considered
@@ -160,7 +163,6 @@ impl ComputeState {
            last_active: None,
            error: None,
            pspec: None,
-            compute_ctl_config: ComputeCtlConfig::default(),
            startup_span: None,
            metrics: ComputeMetrics::default(),
        }
@@ -314,7 +316,6 @@ impl ComputeNode {
            let pspec = ParsedSpec::try_from(cli_spec).map_err(|msg| anyhow::anyhow!(msg))?;
            new_state.pspec = Some(pspec);
        }
-        new_state.compute_ctl_config = compute_ctl_config;

        Ok(ComputeNode {
            params,
@@ -323,6 +324,7 @@ impl ComputeNode {
            state: Mutex::new(new_state),
            state_changed: Condvar::new(),
            ext_download_progress: RwLock::new(HashMap::new()),
+            compute_ctl_config,
        })
    }

@@ -345,7 +347,7 @@ impl ComputeNode {
        // requests while configuration is still in progress.
        crate::http::server::Server::External {
            port: this.params.external_http_port,
-            jwks: this.state.lock().unwrap().compute_ctl_config.jwks.clone(),
+            config: this.compute_ctl_config.clone(),
            compute_id: this.params.compute_id.clone(),
        }
        .launch(&this);
@@ -524,6 +526,16 @@ impl ComputeNode {
        // Collect all the tasks that must finish here
        let mut pre_tasks = tokio::task::JoinSet::new();

+        // Make sure TLS certificates are properly loaded and in the right place.
+        if self.compute_ctl_config.tls.is_some() {
+            let this = self.clone();
+            pre_tasks.spawn(async move {
+                this.watch_cert_for_changes().await;
+
+                Ok::<(), anyhow::Error>(())
+            });
+        }
+
        // If there are any remote extensions in shared_preload_libraries, start downloading them
        if pspec.spec.remote_extensions.is_some() {
            let (this, spec) = (self.clone(), pspec.spec.clone());
@@ -579,11 +591,13 @@ impl ComputeNode {
        if let Some(pgbouncer_settings) = &pspec.spec.pgbouncer_settings {
            info!("tuning pgbouncer");

+            let pgbouncer_settings = pgbouncer_settings.clone();
+            let tls_config = self.compute_ctl_config.tls.clone();
+
            // Spawn a background task to do the tuning,
            // so that we don't block the main thread that starts Postgres.
-            let pgbouncer_settings = pgbouncer_settings.clone();
            let _handle = tokio::spawn(async move {
-                let res = tune_pgbouncer(pgbouncer_settings).await;
+                let res = tune_pgbouncer(pgbouncer_settings, tls_config).await;
                if let Err(err) = res {
                    error!("error while tuning pgbouncer: {err:?}");
                    // Continue with the startup anyway
@@ -606,7 +620,7 @@ impl ComputeNode {
            });
        }

-        // Configure and start rsyslog if necessary
+        // Configure and start rsyslog for HIPAA if necessary
        if let ComputeAudit::Hipaa = pspec.spec.audit_log_level {
            let remote_endpoint = std::env::var("AUDIT_LOGGING_ENDPOINT").unwrap_or("".to_string());
            if remote_endpoint.is_empty() {
@@ -614,13 +628,22 @@ impl ComputeNode {
            }

            let log_directory_path = Path::new(&self.params.pgdata).join("log");
-            // TODO: make this more robust
-            // now rsyslog starts once and there is no monitoring or restart if it fails
-            configure_audit_rsyslog(
-                log_directory_path.to_str().unwrap(),
-                "hipaa",
-                &remote_endpoint,
-            )?;
+            let log_directory_path = log_directory_path.to_string_lossy().to_string();
+            configure_audit_rsyslog(log_directory_path.clone(), "hipaa", &remote_endpoint)?;
+
+            // Launch a background task to clean up the audit logs
+            launch_pgaudit_gc(log_directory_path);
+        }
+
+        // Configure and start rsyslog for Postgres logs export
+        if self.has_feature(ComputeFeature::PostgresLogsExport) {
+            if let Some(ref project_id) = pspec.spec.cluster.cluster_id {
+                let host = PostgresLogsRsyslogConfig::default_host(project_id);
+                let conf = PostgresLogsRsyslogConfig::new(Some(&host));
+                configure_postgres_logs_export(conf)?;
+            } else {
+                warn!("not configuring rsyslog for Postgres logs export: project ID is missing")
+            }
        }

        // Launch remaining service threads
@@ -1105,9 +1128,10 @@ impl ComputeNode {
        // Remove/create an empty pgdata directory and put configuration there.
        self.create_pgdata()?;
        config::write_postgres_conf(
-            &pgdata_path.join("postgresql.conf"),
+            pgdata_path,
            &pspec.spec,
            self.params.internal_http_port,
+            &self.compute_ctl_config.tls,
        )?;

        // Syncing safekeepers is only safe with primary nodes: if a primary
@@ -1489,11 +1513,13 @@ impl ComputeNode {
        if let Some(ref pgbouncer_settings) = spec.pgbouncer_settings {
            info!("tuning pgbouncer");

+            let pgbouncer_settings = pgbouncer_settings.clone();
+            let tls_config = self.compute_ctl_config.tls.clone();
+
            // Spawn a background task to do the tuning,
            // so that we don't block the main thread that starts Postgres.
-            let pgbouncer_settings = pgbouncer_settings.clone();
            tokio::spawn(async move {
-                let res = tune_pgbouncer(pgbouncer_settings).await;
+                let res = tune_pgbouncer(pgbouncer_settings, tls_config).await;
                if let Err(err) = res {
                    error!("error while tuning pgbouncer: {err:?}");
                }
@@ -1505,7 +1531,8 @@ impl ComputeNode {

            // Spawn a background task to do the configuration,
            // so that we don't block the main thread that starts Postgres.
-            let local_proxy = local_proxy.clone();
+            let mut local_proxy = local_proxy.clone();
+            local_proxy.tls = self.compute_ctl_config.tls.clone();
            tokio::spawn(async move {
                if let Err(err) = local_proxy::configure(&local_proxy) {
                    error!("error while configuring local_proxy: {err:?}");
@@ -1515,8 +1542,12 @@ impl ComputeNode {

        // Write new config
        let pgdata_path = Path::new(&self.params.pgdata);
-        let postgresql_conf_path = pgdata_path.join("postgresql.conf");
-        config::write_postgres_conf(&postgresql_conf_path, &spec, self.params.internal_http_port)?;
+        config::write_postgres_conf(
+            pgdata_path,
+            &spec,
+            self.params.internal_http_port,
+            &self.compute_ctl_config.tls,
+        )?;

        if !spec.skip_pg_catalog_updates {
            let max_concurrent_connections = spec.reconfigure_concurrency;
@@ -1587,6 +1618,56 @@ impl ComputeNode {
        Ok(())
    }

+    pub async fn watch_cert_for_changes(self: Arc<Self>) {
+        // update status on cert renewal
+        if let Some(tls_config) = &self.compute_ctl_config.tls {
+            let tls_config = tls_config.clone();
+
+            // wait until the cert exists.
+            let mut cert_watch = watch_cert_for_changes(tls_config.cert_path.clone()).await;
+
+            tokio::task::spawn_blocking(move || {
+                let handle = tokio::runtime::Handle::current();
+                'cert_update: loop {
+                    // let postgres/pgbouncer/local_proxy know the new cert/key exists.
+                    // we need to wait until it's configurable first.
+
+                    let mut state = self.state.lock().unwrap();
+                    'status_update: loop {
+                        match state.status {
+                            // let's update the state to config pending
+                            ComputeStatus::ConfigurationPending | ComputeStatus::Running => {
+                                state.set_status(
+                                    ComputeStatus::ConfigurationPending,
+                                    &self.state_changed,
+                                );
+                                break 'status_update;
+                            }
+
+                            // exit loop
+                            ComputeStatus::Failed
+                            | ComputeStatus::TerminationPending
+                            | ComputeStatus::Terminated => break 'cert_update,
+
+                            // wait
+                            ComputeStatus::Init
+                            | ComputeStatus::Configuration
+                            | ComputeStatus::Empty => {
+                                state = self.state_changed.wait(state).unwrap();
+                            }
+                        }
+                    }
+                    drop(state);
+
+                    // wait for a new certificate update
+                    if handle.block_on(cert_watch.changed()).is_err() {
+                        break;
+                    }
+                }
+            });
+        }
+    }
+
    /// Update the `last_active` in the shared state, but ensure that it's a more recent one.
    pub fn update_last_active(&self, last_active: Option<DateTime<Utc>>) {
        let mut state = self.state.lock().unwrap();
--- a/compute_tools/src/config.rs
+++ b/compute_tools/src/config.rs
@@ -6,11 +6,13 @@ use std::io::Write;
 use std::io::prelude::*;
 use std::path::Path;

-use compute_api::spec::{ComputeAudit, ComputeMode, ComputeSpec, GenericOption};
+use compute_api::responses::TlsConfig;
+use compute_api::spec::{ComputeAudit, ComputeFeature, ComputeMode, ComputeSpec, GenericOption};

 use crate::pg_helpers::{
    GenericOptionExt, GenericOptionsSearch, PgOptionsSerialize, escape_conf_value,
 };
+use crate::tls::{self, SERVER_CRT, SERVER_KEY};

 /// Check that `line` is inside a text file and put it there if it is not.
 /// Create file if it doesn't exist.
@@ -38,10 +40,12 @@ pub fn line_in_file(path: &Path, line: &str) -> Result<bool> {

 /// Create or completely rewrite configuration file specified by `path`
 pub fn write_postgres_conf(
-    path: &Path,
+    pgdata_path: &Path,
    spec: &ComputeSpec,
    extension_server_port: u16,
+    tls_config: &Option<TlsConfig>,
 ) -> Result<()> {
+    let path = pgdata_path.join("postgresql.conf");
    // File::create() destroys the file content if it exists.
    let mut file = File::create(path)?;

@@ -86,6 +90,20 @@ pub fn write_postgres_conf(
        )?;
    }

+    // tls
+    if let Some(tls_config) = tls_config {
+        writeln!(file, "ssl = on")?;
+
+        // postgres requires the keyfile to be in a secure file,
+        // currently too complicated to ensure that at the VM level,
+        // so we just copy them to another file instead. :shrug:
+        tls::update_key_path_blocking(pgdata_path, tls_config);
+
+        // these are the default, but good to be explicit.
+        writeln!(file, "ssl_cert_file = '{}'", SERVER_CRT)?;
+        writeln!(file, "ssl_key_file = '{}'", SERVER_KEY)?;
+    }
+
    // Locales
    if cfg!(target_os = "macos") {
        writeln!(file, "lc_messages='C'")?;
@@ -140,52 +158,89 @@ pub fn write_postgres_conf(
        writeln!(file, "# Managed by compute_ctl: end")?;
    }

-    // If audit logging is enabled, configure pgaudit.
+    // If base audit logging is enabled, configure it.
+    // In this setup, the audit log will be written to the standard postgresql log.
+    //
+    // If compliance audit logging is enabled, configure pgaudit.
    //
    // Note, that this is called after the settings from spec are written.
    // This way we always override the settings from the spec
    // and don't allow the user or the control plane admin to change them.
-    if let ComputeAudit::Hipaa = spec.audit_log_level {
-        writeln!(file, "# Managed by compute_ctl audit settings: begin")?;
-        // This log level is very verbose
-        // but this is necessary for HIPAA compliance.
-        writeln!(file, "pgaudit.log='all'")?;
-        writeln!(file, "pgaudit.log_parameter=on")?;
-        // Disable logging of catalog queries
-        // The catalog doesn't contain sensitive data, so we don't need to audit it.
-        writeln!(file, "pgaudit.log_catalog=off")?;
-        // Set log rotation to 5 minutes
-        // TODO: tune this after performance testing
-        writeln!(file, "pgaudit.log_rotation_age=5")?;
+    match spec.audit_log_level {
+        ComputeAudit::Disabled => {}
+        ComputeAudit::Log => {
+            writeln!(file, "# Managed by compute_ctl base audit settings: start")?;
+            writeln!(file, "pgaudit.log='ddl,role'")?;
+            // Disable logging of catalog queries to reduce the noise
+            writeln!(file, "pgaudit.log_catalog=off")?;

-        // Add audit shared_preload_libraries, if they are not present.
-        //
-        // The caller who sets the flag is responsible for ensuring that the necessary
-        // shared_preload_libraries are present in the compute image,
-        // otherwise the compute start will fail.
-        if let Some(libs) = spec.cluster.settings.find("shared_preload_libraries") {
-            let mut extra_shared_preload_libraries = String::new();
-            if !libs.contains("pgaudit") {
-                extra_shared_preload_libraries.push_str(",pgaudit");
-            }
-            if !libs.contains("pgauditlogtofile") {
-                extra_shared_preload_libraries.push_str(",pgauditlogtofile");
+            if let Some(libs) = spec.cluster.settings.find("shared_preload_libraries") {
+                let mut extra_shared_preload_libraries = String::new();
+                if !libs.contains("pgaudit") {
+                    extra_shared_preload_libraries.push_str(",pgaudit");
+                }
+                writeln!(
+                    file,
+                    "shared_preload_libraries='{}{}'",
+                    libs, extra_shared_preload_libraries
+                )?;
+            } else {
+                // Typically, this should be unreacheable,
+                // because we always set at least some shared_preload_libraries in the spec
+                // but let's handle it explicitly anyway.
+                writeln!(file, "shared_preload_libraries='neon,pgaudit'")?;
            }
+            writeln!(file, "# Managed by compute_ctl base audit settings: end")?;
+        }
+        ComputeAudit::Hipaa => {
            writeln!(
                file,
-                "shared_preload_libraries='{}{}'",
-                libs, extra_shared_preload_libraries
+                "# Managed by compute_ctl compliance audit settings: begin"
            )?;
-        } else {
-            // Typically, this should be unreacheable,
-            // because we always set at least some shared_preload_libraries in the spec
-            // but let's handle it explicitly anyway.
+            // This log level is very verbose
+            // but this is necessary for HIPAA compliance.
+            // Exclude 'misc' category, because it doesn't contain anythig relevant.
+            writeln!(file, "pgaudit.log='all, -misc'")?;
+            writeln!(file, "pgaudit.log_parameter=on")?;
+            // Disable logging of catalog queries
+            // The catalog doesn't contain sensitive data, so we don't need to audit it.
+            writeln!(file, "pgaudit.log_catalog=off")?;
+            // Set log rotation to 5 minutes
+            // TODO: tune this after performance testing
+            writeln!(file, "pgaudit.log_rotation_age=5")?;
+
+            // Add audit shared_preload_libraries, if they are not present.
+            //
+            // The caller who sets the flag is responsible for ensuring that the necessary
+            // shared_preload_libraries are present in the compute image,
+            // otherwise the compute start will fail.
+            if let Some(libs) = spec.cluster.settings.find("shared_preload_libraries") {
+                let mut extra_shared_preload_libraries = String::new();
+                if !libs.contains("pgaudit") {
+                    extra_shared_preload_libraries.push_str(",pgaudit");
+                }
+                if !libs.contains("pgauditlogtofile") {
+                    extra_shared_preload_libraries.push_str(",pgauditlogtofile");
+                }
+                writeln!(
+                    file,
+                    "shared_preload_libraries='{}{}'",
+                    libs, extra_shared_preload_libraries
+                )?;
+            } else {
+                // Typically, this should be unreacheable,
+                // because we always set at least some shared_preload_libraries in the spec
+                // but let's handle it explicitly anyway.
+                writeln!(
+                    file,
+                    "shared_preload_libraries='neon,pgaudit,pgauditlogtofile'"
+                )?;
+            }
            writeln!(
                file,
-                "shared_preload_libraries='neon,pgaudit,pgauditlogtofile'"
+                "# Managed by compute_ctl compliance audit settings: end"
            )?;
        }
-        writeln!(file, "# Managed by compute_ctl audit settings: end")?;
    }

    writeln!(file, "neon.extension_server_port={}", extension_server_port)?;
@@ -197,6 +252,12 @@ pub fn write_postgres_conf(
        writeln!(file, "neon.disable_logical_replication_subscribers=false")?;
    }

+    // We need Postgres to send logs to rsyslog so that we can forward them
+    // further to customers' log aggregation systems.
+    if spec.features.contains(&ComputeFeature::PostgresLogsExport) {
+        writeln!(file, "log_destination='stderr,syslog'")?;
+    }
+
    // This is essential to keep this line at the end of the file,
    // because it is intended to override any settings above.
    writeln!(file, "include_if_exists = 'compute_ctl_temp_override.conf'")?;
--- a/compute_tools/src/config_template/compute_audit_rsyslog_template.conf
+++ b/compute_tools/src/config_template/compute_audit_rsyslog_template.conf
@@ -4,7 +4,8 @@ module(load="imfile")
 # Input configuration for log files in the specified directory
 # Replace {log_directory} with the directory containing the log files
 input(type="imfile" File="{log_directory}/*.log" Tag="{tag}" Severity="info" Facility="local0")
-global(workDirectory="/var/log")
+# the directory to store rsyslog state files
+global(workDirectory="/var/log/rsyslog")

 # Forward logs to remote syslog server
-*.* @@{remote_endpoint}
+*.* @@{remote_endpoint}
--- a/compute_tools/src/config_template/compute_rsyslog_postgres_export_template.conf
+++ b/compute_tools/src/config_template/compute_rsyslog_postgres_export_template.conf
@@ -0,0 +1,10 @@
+# Program name comes from postgres' syslog_facility configuration: https://www.postgresql.org/docs/current/runtime-config-logging.html#GUC-SYSLOG-IDENT
+# Default value is 'postgres'.
+if $programname == 'postgres' then {{
+    # Forward Postgres logs to telemetry otel collector
+    action(type="omfwd" target="{logs_export_target}" port="{logs_export_port}" protocol="tcp"
+           template="RSYSLOG_SyslogProtocol23Format"
+           action.resumeRetryCount="3"
+           queue.type="linkedList" queue.size="1000")
+    stop
+}}
--- a/compute_tools/src/http/openapi_spec.yaml
+++ b/compute_tools/src/http/openapi_spec.yaml
@@ -306,6 +306,36 @@ paths:
              schema:
                $ref: "#/components/schemas/GenericError"

+  /configure_telemetry:
+    post:
+      tags:
+        - Configure
+      summary: Configure rsyslog
+      description: |
+        This API endpoint configures rsyslog to forward Postgres logs
+        to a specified otel collector.
+      operationId: configureTelemetry
+      requestBody:
+        required: true
+        content:
+          application/json:
+            schema:
+              type: object
+              properties:
+                logs_export_host:
+                  type: string
+                  description: |
+                    Hostname and the port of the otel collector. Leave empty to disable logs forwarding.
+                    Example: config-shy-breeze-123-collector-monitoring.neon-telemetry.svc.cluster.local:54526
+      responses:
+        204:
+          description: "Telemetry configured successfully"
+        500:
+          content:
+            application/json:
+              schema:
+                $ref: "#/components/schemas/GenericError"
+
 components:
  securitySchemes:
    JWT:
--- a/compute_tools/src/http/routes/configure.rs
+++ b/compute_tools/src/http/routes/configure.rs
@@ -1,9 +1,11 @@
 use std::sync::Arc;

+use axum::body::Body;
 use axum::extract::State;
 use axum::response::Response;
-use compute_api::requests::ConfigurationRequest;
+use compute_api::requests::{ConfigurationRequest, ConfigureTelemetryRequest};
 use compute_api::responses::{ComputeStatus, ComputeStatusResponse};
+use compute_api::spec::ComputeFeature;
 use http::StatusCode;
 use tokio::task;
 use tracing::info;
@@ -11,6 +13,7 @@ use tracing::info;
 use crate::compute::{ComputeNode, ParsedSpec};
 use crate::http::JsonResponse;
 use crate::http::extract::Json;
+use crate::rsyslog::{PostgresLogsRsyslogConfig, configure_postgres_logs_export};

 // Accept spec in JSON format and request compute configuration. If anything
 // goes wrong after we set the compute status to `ConfigurationPending` and
@@ -92,3 +95,25 @@ pub(in crate::http) async fn configure(

    JsonResponse::success(StatusCode::OK, body)
 }
+
+pub(in crate::http) async fn configure_telemetry(
+    State(compute): State<Arc<ComputeNode>>,
+    request: Json<ConfigureTelemetryRequest>,
+) -> Response {
+    if !compute.has_feature(ComputeFeature::PostgresLogsExport) {
+        return JsonResponse::error(
+            StatusCode::PRECONDITION_FAILED,
+            "Postgres logs export feature is not enabled".to_string(),
+        );
+    }
+
+    let conf = PostgresLogsRsyslogConfig::new(request.logs_export_host.as_deref());
+    if let Err(err) = configure_postgres_logs_export(conf) {
+        return JsonResponse::error(StatusCode::INTERNAL_SERVER_ERROR, err.to_string());
+    }
+
+    Response::builder()
+        .status(StatusCode::NO_CONTENT)
+        .body(Body::from(""))
+        .unwrap()
+}
--- a/compute_tools/src/http/server.rs
+++ b/compute_tools/src/http/server.rs
@@ -8,8 +8,8 @@ use axum::Router;
 use axum::middleware::{self};
 use axum::response::IntoResponse;
 use axum::routing::{get, post};
+use compute_api::responses::ComputeCtlConfig;
 use http::StatusCode;
-use jsonwebtoken::jwk::JwkSet;
 use tokio::net::TcpListener;
 use tower::ServiceBuilder;
 use tower_http::{
@@ -41,7 +41,7 @@ pub enum Server {
    },
    External {
        port: u16,
-        jwks: JwkSet,
+        config: ComputeCtlConfig,
        compute_id: String,
    },
 }
@@ -79,7 +79,7 @@ impl From<&Server> for Router<Arc<ComputeNode>> {
                router
            }
            Server::External {
-                jwks, compute_id, ..
+                config, compute_id, ..
            } => {
                let unauthenticated_router =
                    Router::<Arc<ComputeNode>>::new().route("/metrics", get(metrics::get_metrics));
@@ -87,6 +87,7 @@ impl From<&Server> for Router<Arc<ComputeNode>> {
                let authenticated_router = Router::<Arc<ComputeNode>>::new()
                    .route("/check_writability", post(check_writability::is_writable))
                    .route("/configure", post(configure::configure))
+                    .route("/configure_telemetry", post(configure::configure_telemetry))
                    .route("/database_schema", get(database_schema::get_schema_dump))
                    .route("/dbs_and_roles", get(dbs_and_roles::get_catalog_objects))
                    .route("/insights", get(insights::get_insights))
@@ -95,7 +96,7 @@ impl From<&Server> for Router<Arc<ComputeNode>> {
                    .route("/terminate", post(terminate::terminate))
                    .layer(AsyncRequireAuthorizationLayer::new(Authorize::new(
                        compute_id.clone(),
-                        jwks.clone(),
+                        config.jwks.clone(),
                    )));

                router
--- a/compute_tools/src/lib.rs
+++ b/compute_tools/src/lib.rs
@@ -26,3 +26,4 @@ pub mod spec;
 mod spec_apply;
 pub mod swap;
 pub mod sync_sk;
+pub mod tls;
--- a/compute_tools/src/metrics.rs
+++ b/compute_tools/src/metrics.rs
@@ -1,6 +1,8 @@
-use metrics::core::Collector;
+use metrics::core::{AtomicF64, Collector, GenericGauge};
 use metrics::proto::MetricFamily;
-use metrics::{IntCounterVec, UIntGaugeVec, register_int_counter_vec, register_uint_gauge_vec};
+use metrics::{
+    IntCounterVec, UIntGaugeVec, register_gauge, register_int_counter_vec, register_uint_gauge_vec,
+};
 use once_cell::sync::Lazy;

 pub(crate) static INSTALLED_EXTENSIONS: Lazy<UIntGaugeVec> = Lazy::new(|| {
@@ -59,10 +61,20 @@ pub(crate) static REMOTE_EXT_REQUESTS_TOTAL: Lazy<IntCounterVec> = Lazy::new(||
    .expect("failed to define a metric")
 });

+// Size of audit log directory in bytes
+pub(crate) static AUDIT_LOG_DIR_SIZE: Lazy<GenericGauge<AtomicF64>> = Lazy::new(|| {
+    register_gauge!(
+        "compute_audit_log_dir_size",
+        "Size of audit log directory in bytes",
+    )
+    .expect("failed to define a metric")
+});
+
 pub fn collect() -> Vec<MetricFamily> {
    let mut metrics = INSTALLED_EXTENSIONS.collect();
    metrics.extend(CPLANE_REQUESTS_TOTAL.collect());
    metrics.extend(REMOTE_EXT_REQUESTS_TOTAL.collect());
    metrics.extend(DB_MIGRATION_FAILED.collect());
+    metrics.extend(AUDIT_LOG_DIR_SIZE.collect());
    metrics
 }
--- a/compute_tools/src/pg_helpers.rs
+++ b/compute_tools/src/pg_helpers.rs
@@ -10,8 +10,10 @@ use std::str::FromStr;
 use std::time::{Duration, Instant};

 use anyhow::{Result, bail};
+use compute_api::responses::TlsConfig;
 use compute_api::spec::{Database, GenericOption, GenericOptions, PgIdent, Role};
 use futures::StreamExt;
+use indexmap::IndexMap;
 use ini::Ini;
 use notify::{RecursiveMode, Watcher};
 use postgres::config::Config;
@@ -206,8 +208,8 @@ impl Escaping for PgIdent {
    /// Here we somewhat mimic the logic of Postgres' `pg_get_functiondef()`,
    /// <https://github.com/postgres/postgres/blob/8b49392b270b4ac0b9f5c210e2a503546841e832/src/backend/utils/adt/ruleutils.c#L2924>
    fn pg_quote_dollar(&self) -> (String, String) {
-        let mut tag: String = "".to_string();
-        let mut outer_tag = "x".to_string();
+        let mut tag: String = "x".to_string();
+        let mut outer_tag = "xx".to_string();

        // Find the first suitable tag that is not present in the string.
        // Postgres' max role/DB name length is 63 bytes, so even in the
@@ -406,7 +408,7 @@ pub fn create_pgdata(pgdata: &str) -> Result<()> {

 /// Update pgbouncer.ini with provided options
 fn update_pgbouncer_ini(
-    pgbouncer_config: HashMap<String, String>,
+    pgbouncer_config: IndexMap<String, String>,
    pgbouncer_ini_path: &str,
 ) -> Result<()> {
    let mut conf = Ini::load_from_file(pgbouncer_ini_path)?;
@@ -427,7 +429,10 @@ fn update_pgbouncer_ini(
 /// Tune pgbouncer.
 /// 1. Apply new config using pgbouncer admin console
 /// 2. Add new values to pgbouncer.ini to preserve them after restart
-pub async fn tune_pgbouncer(pgbouncer_config: HashMap<String, String>) -> Result<()> {
+pub async fn tune_pgbouncer(
+    mut pgbouncer_config: IndexMap<String, String>,
+    tls_config: Option<TlsConfig>,
+) -> Result<()> {
    let pgbouncer_connstr = if std::env::var_os("AUTOSCALING").is_some() {
        // for VMs use pgbouncer specific way to connect to
        // pgbouncer admin console without password
@@ -473,19 +478,21 @@ pub async fn tune_pgbouncer(pgbouncer_config: HashMap<String, String>) -> Result
        }
    };

-    // Apply new config
-    for (option_name, value) in pgbouncer_config.iter() {
-        let query = format!("SET {}={}", option_name, value);
-        // keep this log line for debugging purposes
-        info!("Applying pgbouncer setting change: {}", query);
+    if let Some(tls_config) = tls_config {
+        // pgbouncer starts in a half-ok state if it cannot find these files.
+        // It will default to client_tls_sslmode=deny, which causes proxy to error.
+        // There is a small window at startup where these files don't yet exist in the VM.
+        // Best to wait until it exists.
+        loop {
+            if let Ok(true) = tokio::fs::try_exists(&tls_config.key_path).await {
+                break;
+            }
+            tokio::time::sleep(Duration::from_millis(500)).await
+        }

-        if let Err(err) = client.simple_query(&query).await {
-            // Don't fail on error, just print it into log
-            error!(
-                "Failed to apply pgbouncer setting change: {},  {}",
-                query, err
-            );
-        };
+        pgbouncer_config.insert("client_tls_cert_file".to_string(), tls_config.cert_path);
+        pgbouncer_config.insert("client_tls_key_file".to_string(), tls_config.key_path);
+        pgbouncer_config.insert("client_tls_sslmode".to_string(), "allow".to_string());
    }

    // save values to pgbouncer.ini
@@ -501,6 +508,13 @@ pub async fn tune_pgbouncer(pgbouncer_config: HashMap<String, String>) -> Result
    };
    update_pgbouncer_ini(pgbouncer_config, &pgbouncer_ini_path)?;

+    info!("Applying pgbouncer setting change");
+
+    if let Err(err) = client.simple_query("RELOAD").await {
+        // Don't fail on error, just print it into log
+        error!("Failed to apply pgbouncer setting change,  {err}",);
+    };
+
    Ok(())
 }

--- a/compute_tools/src/rsyslog.rs
+++ b/compute_tools/src/rsyslog.rs
@@ -1,8 +1,14 @@
+use std::fs;
+use std::io::ErrorKind;
+use std::path::Path;
 use std::process::Command;
+use std::time::Duration;
 use std::{fs::OpenOptions, io::Write};

-use anyhow::{Context, Result};
-use tracing::info;
+use anyhow::{Context, Result, anyhow};
+use tracing::{error, info, instrument, warn};
+
+const POSTGRES_LOGS_CONF_PATH: &str = "/etc/rsyslog.d/postgres_logs.conf";

 fn get_rsyslog_pid() -> Option<String> {
    let output = Command::new("pgrep")
@@ -43,7 +49,7 @@ fn restart_rsyslog() -> Result<()> {
 }

 pub fn configure_audit_rsyslog(
-    log_directory: &str,
+    log_directory: String,
    tag: &str,
    remote_endpoint: &str,
 ) -> Result<()> {
@@ -75,3 +81,196 @@ pub fn configure_audit_rsyslog(

    Ok(())
 }
+
+/// Configuration for enabling Postgres logs forwarding from rsyslogd
+pub struct PostgresLogsRsyslogConfig<'a> {
+    pub host: Option<&'a str>,
+}
+
+impl<'a> PostgresLogsRsyslogConfig<'a> {
+    pub fn new(host: Option<&'a str>) -> Self {
+        Self { host }
+    }
+
+    pub fn build(&self) -> Result<String> {
+        match self.host {
+            Some(host) => {
+                if let Some((target, port)) = host.split_once(":") {
+                    Ok(format!(
+                        include_str!(
+                            "config_template/compute_rsyslog_postgres_export_template.conf"
+                        ),
+                        logs_export_target = target,
+                        logs_export_port = port,
+                    ))
+                } else {
+                    Err(anyhow!("Invalid host format for Postgres logs export"))
+                }
+            }
+            None => Ok("".to_string()),
+        }
+    }
+
+    fn current_config() -> Result<String> {
+        let config_content = match std::fs::read_to_string(POSTGRES_LOGS_CONF_PATH) {
+            Ok(c) => c,
+            Err(err) if err.kind() == ErrorKind::NotFound => String::new(),
+            Err(err) => return Err(err.into()),
+        };
+        Ok(config_content)
+    }
+
+    /// Returns the default host for otel collector that receives Postgres logs
+    pub fn default_host(project_id: &str) -> String {
+        format!(
+            "config-{}-collector.neon-telemetry.svc.cluster.local:10514",
+            project_id
+        )
+    }
+}
+
+pub fn configure_postgres_logs_export(conf: PostgresLogsRsyslogConfig) -> Result<()> {
+    let new_config = conf.build()?;
+    let current_config = PostgresLogsRsyslogConfig::current_config()?;
+
+    if new_config == current_config {
+        info!("postgres logs rsyslog configuration is up-to-date");
+        return Ok(());
+    }
+
+    // When new config is empty we can simply remove the configuration file.
+    if new_config.is_empty() {
+        info!("removing rsyslog config file: {}", POSTGRES_LOGS_CONF_PATH);
+        match std::fs::remove_file(POSTGRES_LOGS_CONF_PATH) {
+            Ok(_) => {}
+            Err(err) if err.kind() == ErrorKind::NotFound => {}
+            Err(err) => return Err(err.into()),
+        }
+        restart_rsyslog()?;
+        return Ok(());
+    }
+
+    info!(
+        "configuring rsyslog for postgres logs export to: {:?}",
+        conf.host
+    );
+
+    let mut file = OpenOptions::new()
+        .create(true)
+        .write(true)
+        .truncate(true)
+        .open(POSTGRES_LOGS_CONF_PATH)?;
+    file.write_all(new_config.as_bytes())?;
+
+    info!(
+        "rsyslog configuration file {} added successfully. Starting rsyslogd",
+        POSTGRES_LOGS_CONF_PATH
+    );
+
+    restart_rsyslog()?;
+    Ok(())
+}
+
+#[instrument(skip_all)]
+async fn pgaudit_gc_main_loop(log_directory: String) -> Result<()> {
+    info!("running pgaudit GC main loop");
+    loop {
+        // Check log_directory for old pgaudit logs and delete them.
+        // New log files are checked every 5 minutes, as set in pgaudit.log_rotation_age
+        // Find files that were not modified in the last 15 minutes and delete them.
+        // This should be enough time for rsyslog to process the logs and for us to catch the alerts.
+        //
+        // In case of a very high load, we might need to adjust this value and pgaudit.log_rotation_age.
+        //
+        // TODO: add some smarter logic to delete the files that are fully streamed according to rsyslog
+        // imfile-state files, but for now just do a simple GC to avoid filling up the disk.
+        let _ = Command::new("find")
+            .arg(&log_directory)
+            .arg("-name")
+            .arg("audit*.log")
+            .arg("-mmin")
+            .arg("+15")
+            .arg("-delete")
+            .output()?;
+
+        // also collect the metric for the size of the log directory
+        async fn get_log_files_size(path: &Path) -> Result<u64> {
+            let mut total_size = 0;
+
+            for entry in fs::read_dir(path)? {
+                let entry = entry?;
+                let entry_path = entry.path();
+
+                if entry_path.is_file() && entry_path.to_string_lossy().ends_with("log") {
+                    total_size += entry.metadata()?.len();
+                }
+            }
+
+            Ok(total_size)
+        }
+
+        let log_directory_size = get_log_files_size(Path::new(&log_directory))
+            .await
+            .unwrap_or_else(|e| {
+                warn!("Failed to get log directory size: {}", e);
+                0
+            });
+        crate::metrics::AUDIT_LOG_DIR_SIZE.set(log_directory_size as f64);
+        tokio::time::sleep(Duration::from_secs(60)).await;
+    }
+}
+
+// launch pgaudit GC thread to clean up the old pgaudit logs stored in the log_directory
+pub fn launch_pgaudit_gc(log_directory: String) {
+    tokio::spawn(async move {
+        if let Err(e) = pgaudit_gc_main_loop(log_directory).await {
+            error!("pgaudit GC main loop failed: {}", e);
+        }
+    });
+}
+
+#[cfg(test)]
+mod tests {
+    use crate::rsyslog::PostgresLogsRsyslogConfig;
+
+    #[test]
+    fn test_postgres_logs_config() {
+        {
+            // Verify empty config
+            let conf = PostgresLogsRsyslogConfig::new(None);
+            let res = conf.build();
+            assert!(res.is_ok());
+            let conf_str = res.unwrap();
+            assert_eq!(&conf_str, "");
+        }
+
+        {
+            // Verify config
+            let conf = PostgresLogsRsyslogConfig::new(Some("collector.cvc.local:514"));
+            let res = conf.build();
+            assert!(res.is_ok());
+            let conf_str = res.unwrap();
+            assert!(conf_str.contains("omfwd"));
+            assert!(conf_str.contains(r#"target="collector.cvc.local""#));
+            assert!(conf_str.contains(r#"port="514""#));
+        }
+
+        {
+            // Verify invalid config
+            let conf = PostgresLogsRsyslogConfig::new(Some("invalid"));
+            let res = conf.build();
+            assert!(res.is_err());
+        }
+
+        {
+            // Verify config with default host
+            let host = PostgresLogsRsyslogConfig::default_host("shy-breeze-123");
+            let conf = PostgresLogsRsyslogConfig::new(Some(&host));
+            let res = conf.build();
+            assert!(res.is_ok());
+            let conf_str = res.unwrap();
+            assert!(conf_str.contains(r#"shy-breeze-123"#));
+            assert!(conf_str.contains(r#"port="10514""#));
+        }
+    }
+}
--- a/compute_tools/src/spec.rs
+++ b/compute_tools/src/spec.rs
@@ -8,13 +8,12 @@ use compute_api::responses::{
 use compute_api::spec::ComputeSpec;
 use reqwest::StatusCode;
 use tokio_postgres::Client;
-use tracing::{error, info, instrument, warn};
+use tracing::{error, info, instrument};

 use crate::config;
 use crate::metrics::{CPLANE_REQUESTS_TOTAL, CPlaneRequestRPC, UNKNOWN_HTTP_STATUS};
 use crate::migration::MigrationRunner;
 use crate::params::PG_HBA_ALL_MD5;
-use crate::pg_helpers::*;

 // Do control plane request and return response if any. In case of error it
 // returns a bool flag indicating whether it makes sense to retry the request
@@ -212,122 +211,3 @@ pub async fn handle_migrations(client: &mut Client) -> Result<()> {

    Ok(())
 }
-
-/// Connect to the database as superuser and pre-create anon extension
-/// if it is present in shared_preload_libraries
-#[instrument(skip_all)]
-pub async fn handle_extension_anon(
-    spec: &ComputeSpec,
-    db_owner: &str,
-    db_client: &mut Client,
-    grants_only: bool,
-) -> Result<()> {
-    info!("handle extension anon");
-
-    if let Some(libs) = spec.cluster.settings.find("shared_preload_libraries") {
-        if libs.contains("anon") {
-            if !grants_only {
-                // check if extension is already initialized using anon.is_initialized()
-                let query = "SELECT anon.is_initialized()";
-                match db_client.query(query, &[]).await {
-                    Ok(rows) => {
-                        if !rows.is_empty() {
-                            let is_initialized: bool = rows[0].get(0);
-                            if is_initialized {
-                                info!("anon extension is already initialized");
-                                return Ok(());
-                            }
-                        }
-                    }
-                    Err(e) => {
-                        warn!(
-                            "anon extension is_installed check failed with expected error: {}",
-                            e
-                        );
-                    }
-                };
-
-                // Create anon extension if this compute needs it
-                // Users cannot create it themselves, because superuser is required.
-                let mut query = "CREATE EXTENSION IF NOT EXISTS anon CASCADE";
-                info!("creating anon extension with query: {}", query);
-                match db_client.query(query, &[]).await {
-                    Ok(_) => {}
-                    Err(e) => {
-                        error!("anon extension creation failed with error: {}", e);
-                        return Ok(());
-                    }
-                }
-
-                // check that extension is installed
-                query = "SELECT extname FROM pg_extension WHERE extname = 'anon'";
-                let rows = db_client.query(query, &[]).await?;
-                if rows.is_empty() {
-                    error!("anon extension is not installed");
-                    return Ok(());
-                }
-
-                // Initialize anon extension
-                // This also requires superuser privileges, so users cannot do it themselves.
-                query = "SELECT anon.init()";
-                match db_client.query(query, &[]).await {
-                    Ok(_) => {}
-                    Err(e) => {
-                        error!("anon.init() failed with error: {}", e);
-                        return Ok(());
-                    }
-                }
-            }
-
-            // check that extension is installed, if not bail early
-            let query = "SELECT extname FROM pg_extension WHERE extname = 'anon'";
-            match db_client.query(query, &[]).await {
-                Ok(rows) => {
-                    if rows.is_empty() {
-                        error!("anon extension is not installed");
-                        return Ok(());
-                    }
-                }
-                Err(e) => {
-                    error!("anon extension check failed with error: {}", e);
-                    return Ok(());
-                }
-            };
-
-            let query = format!("GRANT ALL ON SCHEMA anon TO {}", db_owner);
-            info!("granting anon extension permissions with query: {}", query);
-            db_client.simple_query(&query).await?;
-
-            // Grant permissions to db_owner to use anon extension functions
-            let query = format!("GRANT ALL ON ALL FUNCTIONS IN SCHEMA anon TO {}", db_owner);
-            info!("granting anon extension permissions with query: {}", query);
-            db_client.simple_query(&query).await?;
-
-            // This is needed, because some functions are defined as SECURITY DEFINER.
-            // In Postgres SECURITY DEFINER functions are executed with the privileges
-            // of the owner.
-            // In anon extension this it is needed to access some GUCs, which are only accessible to
-            // superuser. But we've patched postgres to allow db_owner to access them as well.
-            // So we need to change owner of these functions to db_owner.
-            let query = format!("
-                SELECT 'ALTER FUNCTION '||nsp.nspname||'.'||p.proname||'('||pg_get_function_identity_arguments(p.oid)||') OWNER TO {};'
-                from pg_proc p
-                join pg_namespace nsp ON p.pronamespace = nsp.oid
-                where nsp.nspname = 'anon';", db_owner);
-
-            info!("change anon extension functions owner to db owner");
-            db_client.simple_query(&query).await?;
-
-            //  affects views as well
-            let query = format!("GRANT ALL ON ALL TABLES IN SCHEMA anon TO {}", db_owner);
-            info!("granting anon extension permissions with query: {}", query);
-            db_client.simple_query(&query).await?;
-
-            let query = format!("GRANT ALL ON ALL SEQUENCES IN SCHEMA anon TO {}", db_owner);
-            info!("granting anon extension permissions with query: {}", query);
-            db_client.simple_query(&query).await?;
-        }
-    }
-
-    Ok(())
-}
--- a/compute_tools/src/spec_apply.rs
+++ b/compute_tools/src/spec_apply.rs
@@ -6,7 +6,7 @@ use std::sync::Arc;

 use anyhow::{Context, Result};
 use compute_api::responses::ComputeStatus;
-use compute_api::spec::{ComputeAudit, ComputeFeature, ComputeSpec, Database, PgIdent, Role};
+use compute_api::spec::{ComputeAudit, ComputeSpec, Database, PgIdent, Role};
 use futures::future::join_all;
 use tokio::sync::RwLock;
 use tokio_postgres::Client;
@@ -26,7 +26,7 @@ use crate::spec_apply::ApplySpecPhase::{
    RunInEachDatabase,
 };
 use crate::spec_apply::PerDatabasePhase::{
-    ChangeSchemaPerms, DeleteDBRoleReferences, DropLogicalSubscriptions, HandleAnonExtension,
+    ChangeSchemaPerms, DeleteDBRoleReferences, DropLogicalSubscriptions,
 };

 impl ComputeNode {
@@ -238,7 +238,6 @@ impl ComputeNode {
                    let mut phases = vec![
                        DeleteDBRoleReferences,
                        ChangeSchemaPerms,
-                        HandleAnonExtension,
                    ];

                    if spec.drop_subscriptions_before_start && !drop_subscriptions_done {
@@ -287,7 +286,10 @@ impl ComputeNode {
                    phases.push(CreatePgauditlogtofileExtension);
                    phases.push(DisablePostgresDBPgAudit);
                }
-                ComputeAudit::Log => { /* not implemented yet */ }
+                ComputeAudit::Log => {
+                    phases.push(CreatePgauditExtension);
+                    phases.push(DisablePostgresDBPgAudit);
+                }
                ComputeAudit::Disabled => {}
            }

@@ -458,7 +460,6 @@ impl Debug for DB {
 pub enum PerDatabasePhase {
    DeleteDBRoleReferences,
    ChangeSchemaPerms,
-    HandleAnonExtension,
    /// This is a shared phase, used for both i) dropping dangling LR subscriptions
    /// before dropping the DB, and ii) dropping all subscriptions after creating
    /// a fresh branch.
@@ -1012,98 +1013,6 @@ async fn get_operations<'a>(
                    ]
                    .into_iter();

-                    Ok(Box::new(operations))
-                }
-                // TODO: remove this completely https://github.com/neondatabase/cloud/issues/22663
-                PerDatabasePhase::HandleAnonExtension => {
-                    // Only install Anon into user databases
-                    let db = match &db {
-                        DB::SystemDB => return Ok(Box::new(empty())),
-                        DB::UserDB(db) => db,
-                    };
-                    // Never install Anon when it's not enabled as feature
-                    if !spec.features.contains(&ComputeFeature::AnonExtension) {
-                        return Ok(Box::new(empty()));
-                    }
-
-                    // Only install Anon when it's added in preload libraries
-                    let opt_libs = spec.cluster.settings.find("shared_preload_libraries");
-
-                    let libs = match opt_libs {
-                        Some(libs) => libs,
-                        None => return Ok(Box::new(empty())),
-                    };
-
-                    if !libs.contains("anon") {
-                        return Ok(Box::new(empty()));
-                    }
-
-                    let db_owner = db.owner.pg_quote();
-
-                    let operations = vec![
-                        // Create anon extension if this compute needs it
-                        // Users cannot create it themselves, because superuser is required.
-                        Operation {
-                            query: String::from("CREATE EXTENSION IF NOT EXISTS anon CASCADE"),
-                            comment: Some(String::from("creating anon extension")),
-                        },
-                        // Initialize anon extension
-                        // This also requires superuser privileges, so users cannot do it themselves.
-                        Operation {
-                            query: String::from("SELECT anon.init()"),
-                            comment: Some(String::from("initializing anon extension data")),
-                        },
-                        Operation {
-                            query: format!("GRANT ALL ON SCHEMA anon TO {}", db_owner),
-                            comment: Some(String::from(
-                                "granting anon extension schema permissions",
-                            )),
-                        },
-                        Operation {
-                            query: format!(
-                                "GRANT ALL ON ALL FUNCTIONS IN SCHEMA anon TO {}",
-                                db_owner
-                            ),
-                            comment: Some(String::from(
-                                "granting anon extension schema functions permissions",
-                            )),
-                        },
-                        // We need this, because some functions are defined as SECURITY DEFINER.
-                        // In Postgres SECURITY DEFINER functions are executed with the privileges
-                        // of the owner.
-                        // In anon extension this it is needed to access some GUCs, which are only accessible to
-                        // superuser. But we've patched postgres to allow db_owner to access them as well.
-                        // So we need to change owner of these functions to db_owner.
-                        Operation {
-                            query: format!(
-                                include_str!("sql/anon_ext_fn_reassign.sql"),
-                                db_owner = db_owner,
-                            ),
-                            comment: Some(String::from(
-                                "change anon extension functions owner to database_owner",
-                            )),
-                        },
-                        Operation {
-                            query: format!(
-                                "GRANT ALL ON ALL TABLES IN SCHEMA anon TO {}",
-                                db_owner,
-                            ),
-                            comment: Some(String::from(
-                                "granting anon extension tables permissions",
-                            )),
-                        },
-                        Operation {
-                            query: format!(
-                                "GRANT ALL ON ALL SEQUENCES IN SCHEMA anon TO {}",
-                                db_owner,
-                            ),
-                            comment: Some(String::from(
-                                "granting anon extension sequences permissions",
-                            )),
-                        },
-                    ]
-                    .into_iter();
-
                    Ok(Box::new(operations))
                }
            }
--- a/compute_tools/src/tls.rs
+++ b/compute_tools/src/tls.rs
@@ -0,0 +1,117 @@
+use std::{io::Write, os::unix::fs::OpenOptionsExt, path::Path, time::Duration};
+
+use anyhow::{Context, Result, bail};
+use compute_api::responses::TlsConfig;
+use ring::digest;
+use spki::der::{Decode, PemReader};
+use x509_cert::Certificate;
+
+#[derive(Clone, Copy)]
+pub struct CertDigest(digest::Digest);
+
+pub async fn watch_cert_for_changes(cert_path: String) -> tokio::sync::watch::Receiver<CertDigest> {
+    let mut digest = compute_digest(&cert_path).await;
+    let (tx, rx) = tokio::sync::watch::channel(digest);
+    tokio::spawn(async move {
+        while !tx.is_closed() {
+            let new_digest = compute_digest(&cert_path).await;
+            if digest.0.as_ref() != new_digest.0.as_ref() {
+                digest = new_digest;
+                _ = tx.send(digest);
+            }
+
+            tokio::time::sleep(Duration::from_secs(60)).await
+        }
+    });
+    rx
+}
+
+async fn compute_digest(cert_path: &str) -> CertDigest {
+    loop {
+        match try_compute_digest(cert_path).await {
+            Ok(d) => break d,
+            Err(e) => {
+                tracing::error!("could not read cert file {e:?}");
+                tokio::time::sleep(Duration::from_secs(1)).await
+            }
+        }
+    }
+}
+
+async fn try_compute_digest(cert_path: &str) -> Result<CertDigest> {
+    let data = tokio::fs::read(cert_path).await?;
+    // sha256 is extremely collision resistent. can safely assume the digest to be unique
+    Ok(CertDigest(digest::digest(&digest::SHA256, &data)))
+}
+
+pub const SERVER_CRT: &str = "server.crt";
+pub const SERVER_KEY: &str = "server.key";
+
+pub fn update_key_path_blocking(pg_data: &Path, tls_config: &TlsConfig) {
+    loop {
+        match try_update_key_path_blocking(pg_data, tls_config) {
+            Ok(()) => break,
+            Err(e) => {
+                tracing::error!("could not create key file {e:?}");
+                std::thread::sleep(Duration::from_secs(1))
+            }
+        }
+    }
+}
+
+// Postgres requires the keypath be "secure". This means
+// 1. Owned by the postgres user.
+// 2. Have permission 600.
+fn try_update_key_path_blocking(pg_data: &Path, tls_config: &TlsConfig) -> Result<()> {
+    let key = std::fs::read_to_string(&tls_config.key_path)?;
+    let crt = std::fs::read_to_string(&tls_config.cert_path)?;
+
+    // to mitigate a race condition during renewal.
+    verify_key_cert(&key, &crt)?;
+
+    let mut key_file = std::fs::OpenOptions::new()
+        .write(true)
+        .create(true)
+        .truncate(true)
+        .mode(0o600)
+        .open(pg_data.join(SERVER_KEY))?;
+
+    let mut crt_file = std::fs::OpenOptions::new()
+        .write(true)
+        .create(true)
+        .truncate(true)
+        .mode(0o600)
+        .open(pg_data.join(SERVER_CRT))?;
+
+    key_file.write_all(key.as_bytes())?;
+    crt_file.write_all(crt.as_bytes())?;
+
+    Ok(())
+}
+
+fn verify_key_cert(key: &str, cert: &str) -> Result<()> {
+    use x509_cert::der::oid::db::rfc5912::ECDSA_WITH_SHA_256;
+
+    let cert = Certificate::decode(&mut PemReader::new(cert.as_bytes()).context("pem reader")?)
+        .context("decode cert")?;
+
+    match cert.signature_algorithm.oid {
+        ECDSA_WITH_SHA_256 => {
+            let key = p256::SecretKey::from_sec1_pem(key).context("parse key")?;
+
+            let a = key.public_key().to_sec1_bytes();
+            let b = cert
+                .tbs_certificate
+                .subject_public_key_info
+                .subject_public_key
+                .raw_bytes();
+
+            if *a != *b {
+                bail!("private key file does not match certificate")
+            }
+        }
+        _ => bail!("unknown TLS key type"),
+    }
+
+    Ok(())
+}
--- a/compute_tools/tests/pg_helpers_tests.rs
+++ b/compute_tools/tests/pg_helpers_tests.rs
@@ -64,7 +64,8 @@ test.escaping = 'here''s a backslash \\ and a quote '' and a double-quote " hoor
    #[test]
    fn ident_pg_quote_dollar() {
        let test_cases = vec![
-            ("name", ("$$name$$", "x")),
+            ("name", ("$x$name$x$", "xx")),
+            ("name$", ("$x$name$$x$", "xx")),
            ("name$$", ("$x$name$$$x$", "xx")),
            ("name$$$", ("$x$name$$$$x$", "xx")),
            ("name$$$$", ("$x$name$$$$$x$", "xx")),
--- a/control_plane/src/bin/neon_local.rs
+++ b/control_plane/src/bin/neon_local.rs
@@ -979,7 +979,7 @@ fn handle_init(args: &InitCmdArgs) -> anyhow::Result<LocalEnv> {
            neon_distrib_dir: None,
            default_tenant_id: TenantId::from_array(std::array::from_fn(|_| 0)),
            storage_controller: None,
-            control_plane_compute_hook_api: None,
+            control_plane_hooks_api: None,
            generate_local_ssl_certs: false,
        }
    };
--- a/control_plane/src/local_env.rs
+++ b/control_plane/src/local_env.rs
@@ -72,9 +72,9 @@ pub struct LocalEnv {
    // be propagated into each pageserver's configuration.
    pub control_plane_api: Url,

-    // Control plane upcall API for storage controller.  If set, this will be propagated into the
+    // Control plane upcall APIs for storage controller.  If set, this will be propagated into the
    // storage controller's configuration.
-    pub control_plane_compute_hook_api: Option<Url>,
+    pub control_plane_hooks_api: Option<Url>,

    /// Keep human-readable aliases in memory (and persist them to config), to hide ZId hex strings from the user.
    // A `HashMap<String, HashMap<TenantId, TimelineId>>` would be more appropriate here,
@@ -104,6 +104,7 @@ pub struct OnDiskConfig {
    pub pageservers: Vec<PageServerConf>,
    pub safekeepers: Vec<SafekeeperConf>,
    pub control_plane_api: Option<Url>,
+    pub control_plane_hooks_api: Option<Url>,
    pub control_plane_compute_hook_api: Option<Url>,
    branch_name_mappings: HashMap<String, Vec<(TenantId, TimelineId)>>,
    // Note: skip serializing because in compat tests old storage controller fails
@@ -136,7 +137,7 @@ pub struct NeonLocalInitConf {
    pub pageservers: Vec<NeonLocalInitPageserverConf>,
    pub safekeepers: Vec<SafekeeperConf>,
    pub control_plane_api: Option<Url>,
-    pub control_plane_compute_hook_api: Option<Option<Url>>,
+    pub control_plane_hooks_api: Option<Url>,
    pub generate_local_ssl_certs: bool,
 }

@@ -148,7 +149,7 @@ pub struct NeonBroker {
    pub listen_addr: SocketAddr,
 }

-/// Broker config for cluster internal communication.
+/// A part of storage controller's config the neon_local knows about.
 #[derive(Serialize, Deserialize, PartialEq, Eq, Clone, Debug)]
 #[serde(default)]
 pub struct NeonStorageControllerConf {
@@ -175,10 +176,11 @@ pub struct NeonStorageControllerConf {
    #[serde(with = "humantime_serde")]
    pub long_reconcile_threshold: Option<Duration>,

-    #[serde(default)]
    pub use_https_pageserver_api: bool,

    pub timelines_onto_safekeepers: bool,
+
+    pub use_https_safekeeper_api: bool,
 }

 impl NeonStorageControllerConf {
@@ -204,6 +206,7 @@ impl Default for NeonStorageControllerConf {
            long_reconcile_threshold: None,
            use_https_pageserver_api: false,
            timelines_onto_safekeepers: false,
+            use_https_safekeeper_api: false,
        }
    }
 }
@@ -301,6 +304,7 @@ pub struct SafekeeperConf {
    pub pg_port: u16,
    pub pg_tenant_only_port: Option<u16>,
    pub http_port: u16,
+    pub https_port: Option<u16>,
    pub sync: bool,
    pub remote_storage: Option<String>,
    pub backup_threads: Option<u32>,
@@ -315,6 +319,7 @@ impl Default for SafekeeperConf {
            pg_port: 0,
            pg_tenant_only_port: None,
            http_port: 0,
+            https_port: None,
            sync: true,
            remote_storage: None,
            backup_threads: None,
@@ -573,7 +578,8 @@ impl LocalEnv {
                pageservers,
                safekeepers,
                control_plane_api,
-                control_plane_compute_hook_api,
+                control_plane_hooks_api,
+                control_plane_compute_hook_api: _,
                branch_name_mappings,
                generate_local_ssl_certs,
            } = on_disk_config;
@@ -588,7 +594,7 @@ impl LocalEnv {
                pageservers,
                safekeepers,
                control_plane_api: control_plane_api.unwrap(),
-                control_plane_compute_hook_api,
+                control_plane_hooks_api,
                branch_name_mappings,
                generate_local_ssl_certs,
            }
@@ -695,7 +701,8 @@ impl LocalEnv {
                pageservers: vec![], // it's skip_serializing anyway
                safekeepers: self.safekeepers.clone(),
                control_plane_api: Some(self.control_plane_api.clone()),
-                control_plane_compute_hook_api: self.control_plane_compute_hook_api.clone(),
+                control_plane_hooks_api: self.control_plane_hooks_api.clone(),
+                control_plane_compute_hook_api: None,
                branch_name_mappings: self.branch_name_mappings.clone(),
                generate_local_ssl_certs: self.generate_local_ssl_certs,
            },
@@ -779,8 +786,8 @@ impl LocalEnv {
            pageservers,
            safekeepers,
            control_plane_api,
-            control_plane_compute_hook_api,
            generate_local_ssl_certs,
+            control_plane_hooks_api,
        } = conf;

        // Find postgres binaries.
@@ -827,7 +834,7 @@ impl LocalEnv {
            pageservers: pageservers.iter().map(Into::into).collect(),
            safekeepers,
            control_plane_api: control_plane_api.unwrap(),
-            control_plane_compute_hook_api: control_plane_compute_hook_api.unwrap_or_default(),
+            control_plane_hooks_api,
            branch_name_mappings: Default::default(),
            generate_local_ssl_certs,
        };
@@ -842,6 +849,9 @@ impl LocalEnv {
        // create safekeeper dirs
        for safekeeper in &env.safekeepers {
            fs::create_dir_all(SafekeeperNode::datadir_path_by_id(&env, safekeeper.id))?;
+            SafekeeperNode::from_env(&env, safekeeper)
+                .initialize()
+                .context("safekeeper init failed")?;
        }

        // initialize pageserver state
--- a/control_plane/src/safekeeper.rs
+++ b/control_plane/src/safekeeper.rs
@@ -111,6 +111,18 @@ impl SafekeeperNode {
            .expect("non-Unicode path")
    }

+    /// Initializes a safekeeper node by creating all necessary files,
+    /// e.g. SSL certificates.
+    pub fn initialize(&self) -> anyhow::Result<()> {
+        if self.env.generate_local_ssl_certs {
+            self.env.generate_ssl_cert(
+                &self.datadir_path().join("server.crt"),
+                &self.datadir_path().join("server.key"),
+            )?;
+        }
+        Ok(())
+    }
+
    pub async fn start(
        &self,
        extra_opts: &[String],
@@ -196,6 +208,16 @@ impl SafekeeperNode {
            ]);
        }

+        if let Some(https_port) = self.conf.https_port {
+            args.extend([
+                "--listen-https".to_owned(),
+                format!("{}:{}", self.listen_addr, https_port),
+            ]);
+        }
+        if let Some(ssl_ca_file) = self.env.ssl_ca_cert_path() {
+            args.push(format!("--ssl-ca-file={}", ssl_ca_file.to_str().unwrap()));
+        }
+
        args.extend_from_slice(extra_opts);

        background_process::start_process(
--- a/control_plane/src/storage_controller.rs
+++ b/control_plane/src/storage_controller.rs
@@ -538,6 +538,10 @@ impl StorageController {
            args.push("--use-https-pageserver-api".to_string());
        }

+        if self.config.use_https_safekeeper_api {
+            args.push("--use-https-safekeeper-api".to_string());
+        }
+
        if let Some(ssl_ca_file) = self.env.ssl_ca_cert_path() {
            args.push(format!("--ssl-ca-file={}", ssl_ca_file.to_str().unwrap()));
        }
@@ -558,10 +562,8 @@ impl StorageController {
            args.push(format!("--public-key=\"{public_key}\""));
        }

-        if let Some(control_plane_compute_hook_api) = &self.env.control_plane_compute_hook_api {
-            args.push(format!(
-                "--compute-hook-url={control_plane_compute_hook_api}"
-            ));
+        if let Some(control_plane_hooks_api) = &self.env.control_plane_hooks_api {
+            args.push(format!("--control-plane-url={control_plane_hooks_api}"));
        }

        if let Some(split_threshold) = self.config.split_threshold.as_ref() {
--- a/deny.toml
+++ b/deny.toml
@@ -31,10 +31,6 @@ reason = "the marvin attack only affects private key decryption, not public key
 id = "RUSTSEC-2024-0436"
 reason = "The paste crate is a build-only dependency with no runtime components. It is unlikely to have any security impact."

-[[advisories.ignore]]
-id = "RUSTSEC-2025-0014"
-reason = "The humantime is widely used and is not easy to replace right now. It is unmaintained, but it has no known vulnerabilities to care about. #11179"
-
 # This section is considered when running `cargo deny check licenses`
 # More documentation for the licenses section can be found here:
 # https://embarkstudios.github.io/cargo-deny/checks/licenses/cfg.html
--- a/docker-compose/compute_wrapper/Dockerfile
+++ b/docker-compose/compute_wrapper/Dockerfile
@@ -1,4 +1,4 @@
-ARG REPOSITORY=neondatabase
+ARG REPOSITORY=ghcr.io/neondatabase
 ARG COMPUTE_IMAGE=compute-node-v14
 ARG TAG=latest

--- a/docker-compose/docker-compose.yml
+++ b/docker-compose/docker-compose.yml
@@ -29,7 +29,7 @@ services:

  pageserver:
    restart: always
-    image: ${REPOSITORY:-neondatabase}/neon:${TAG:-latest}
+    image: ${REPOSITORY:-ghcr.io/neondatabase}/neon:${TAG:-latest}
    environment:
      - AWS_ACCESS_KEY_ID=minio
      - AWS_SECRET_ACCESS_KEY=password
@@ -45,7 +45,7 @@ services:

  safekeeper1:
    restart: always
-    image: ${REPOSITORY:-neondatabase}/neon:${TAG:-latest}
+    image: ${REPOSITORY:-ghcr.io/neondatabase}/neon:${TAG:-latest}
    environment:
      - SAFEKEEPER_ADVERTISE_URL=safekeeper1:5454
      - SAFEKEEPER_ID=1
@@ -75,7 +75,7 @@ services:

  safekeeper2:
    restart: always
-    image: ${REPOSITORY:-neondatabase}/neon:${TAG:-latest}
+    image: ${REPOSITORY:-ghcr.io/neondatabase}/neon:${TAG:-latest}
    environment:
      - SAFEKEEPER_ADVERTISE_URL=safekeeper2:5454
      - SAFEKEEPER_ID=2
@@ -105,7 +105,7 @@ services:

  safekeeper3:
    restart: always
-    image: ${REPOSITORY:-neondatabase}/neon:${TAG:-latest}
+    image: ${REPOSITORY:-ghcr.io/neondatabase}/neon:${TAG:-latest}
    environment:
      - SAFEKEEPER_ADVERTISE_URL=safekeeper3:5454
      - SAFEKEEPER_ID=3
@@ -135,7 +135,7 @@ services:

  storage_broker:
    restart: always
-    image: ${REPOSITORY:-neondatabase}/neon:${TAG:-latest}
+    image: ${REPOSITORY:-ghcr.io/neondatabase}/neon:${TAG:-latest}
    ports:
      - 50051:50051
    command:
@@ -147,7 +147,7 @@ services:
    build:
      context: ./compute_wrapper/
      args:
-        - REPOSITORY=${REPOSITORY:-neondatabase}
+        - REPOSITORY=${REPOSITORY:-ghcr.io/neondatabase}
        - COMPUTE_IMAGE=compute-node-v${PG_VERSION:-16}
        - TAG=${COMPUTE_TAG:-${TAG:-latest}}
        - http_proxy=${http_proxy:-}
@@ -186,7 +186,7 @@ services:

  neon-test-extensions:
    profiles: ["test-extensions"]
-    image: ${REPOSITORY:-neondatabase}/neon-test-extensions-v${PG_TEST_VERSION:-16}:${TEST_EXTENSIONS_TAG:-${TAG:-latest}}
+    image: ${REPOSITORY:-ghcr.io/neondatabase}/neon-test-extensions-v${PG_TEST_VERSION:-16}:${TEST_EXTENSIONS_TAG:-${TAG:-latest}}
    environment:
      - PGPASSWORD=cloud_admin
    entrypoint:
--- a/docker-compose/run-tests.sh
+++ b/docker-compose/run-tests.sh
@@ -20,4 +20,4 @@ for d in ${LIST}; do
 done
 [ -z "${FAILED}" ] && exit 0
 echo "${FAILED}"
-exit 1
+exit 1
--- a/docs/rfcs/001-cluster-size-limits.md
+++ b/docs/rfcs/001-cluster-size-limits.md
--- a/docs/rfcs/README.md
+++ b/docs/rfcs/README.md
@@ -1,3 +1,7 @@
+# Neon RFCs
+
+## Overview
+
 This directory contains Request for Comments documents, or RFCs, for
 features or concepts that have been proposed. Alternative names:
 technical design doc, ERD, one-pager
@@ -59,37 +63,10 @@ RFC lifecycle:

 ### RFC template

+Use template with `YYYY-MM-DD-copy-me.md` as a starting point. Timestamp prefix helps to avoid awkward 'id' collisions.
+
+```sh
+cp docs/rfcs/YYYY-MM-DD-copy-me.md docs/rfcs/$(date +"%Y-%m-%d")-<name>.md
+```
+
 Note, a lot of the sections are marked as ‘if relevant’. They are included into the template as a reminder and to help inspiration.
-
-```
-# Name
-Created on ..
-Implemented on ..
-
-## Summary
-
-## Motivation
-
-## Non Goals (if relevant)
-
-## Impacted components (e.g. pageserver, safekeeper, console, etc)
-
-## Proposed implementation
-
-### Reliability, failure modes and corner cases (if relevant)
-
-### Interaction/Sequence diagram (if relevant)
-
-### Scalability (if relevant)
-
-### Security implications (if relevant)
-
-### Unresolved questions (if relevant)
-
-## Alternative implementation (if relevant)
-
-## Pros/cons of proposed approaches (if relevant)
-
-## Definition of Done (if relevant)
-
-```
--- a/docs/rfcs/YYYY-MM-DD-copy-me.md
+++ b/docs/rfcs/YYYY-MM-DD-copy-me.md
@@ -0,0 +1,30 @@
+# Name
+
+Created on YYYY-MM-DD
+Implemented on _TBD_
+
+## Summary
+
+## Motivation
+
+## Non Goals (if relevant)
+
+## Impacted components (e.g. pageserver, safekeeper, console, etc)
+
+## Proposed implementation
+
+### Reliability, failure modes and corner cases (if relevant)
+
+### Interaction/Sequence diagram (if relevant)
+
+### Scalability (if relevant)
+
+### Security implications (if relevant)
+
+### Unresolved questions (if relevant)
+
+## Alternative implementation (if relevant)
+
+## Pros/cons of proposed approaches (if relevant)
+
+## Definition of Done (if relevant)
--- a/docs/storage_controller.md
+++ b/docs/storage_controller.md
@@ -101,15 +101,25 @@ changes such as a pageserver node becoming unavailable, or the tenant's shard co
 postgres clients to handle such changes, the storage controller calls an API hook when a tenant's pageserver
 location changes.

-The hook is configured using the storage controller's `--control-plane-url` CLI option. If the hook requires
-JWT auth, the token may be provided with `--control-plane-jwt-token`. The hook will be invoked with a `PUT` request.
+The hook is configured using the storage controller's `--control-plane-url` CLI option, from which the hook URL is computed.

-In the Neon cloud service, this hook is implemented by Neon's internal cloud control plane. In `neon_local` systems
+Currently, there is two hooks, each computed by appending the name to the provided control plane URL prefix:
+
+- `notify-attach`, called whenever attachment for pageservers changes
+- `notify-safekeepers`, called whenever attachment for safekeepers changes
+
+If the hooks require JWT auth, the token may be provided with `--control-plane-jwt-token`.
+The hooks will be invoked with a `PUT` request.
+
+In the Neon cloud service, these hooks are implemented by Neon's internal cloud control plane. In `neon_local` systems,
 the storage controller integrates directly with neon_local to reconfigure local postgres processes instead of calling
 the compute hook.

-When implementing an on-premise Neon deployment, you must implement a service that handles the compute hook. This is not complicated:
-the request body has format of the `ComputeHookNotifyRequest` structure, provided below for convenience.
+When implementing an on-premise Neon deployment, you must implement a service that handles the compute hooks. This is not complicated.
+
+### `notify-attach` body
+
+The `notify-attach` request body follows the format of the `ComputeHookNotifyRequest` structure, provided below for convenience.

 ```
 struct ComputeHookNotifyRequestShard {
@@ -128,15 +138,15 @@ When a notification is received:

 1. Modify postgres configuration for this tenant:

-   - set `neon.pageserver_connstr` to a comma-separated list of postgres connection strings to pageservers according to the `shards` list. The
+   - set `neon.pageserver_connstring` to a comma-separated list of postgres connection strings to pageservers according to the `shards` list. The
     shards identified by `NodeId` must be converted to the address+port of the node.
-   - if stripe_size is not None, set `neon.stripe_size` to this value
+   - if stripe_size is not None, set `neon.shard_stripe_size` to this value

 2. Send SIGHUP to postgres to reload configuration
 3. Respond with 200 to the notification request. Do not return success if postgres was not updated: if an error is returned, the controller
   will retry the notification until it succeeds..

-### Example notification body
+Example body:

 ```
 {
@@ -148,3 +158,34 @@ When a notification is received:
  ],
 }
 ```
+
+### `notify-safekeepers` body
+
+The `notify-safekeepers` request body forllows the format of the `SafekeepersNotifyRequest` structure, provided below for convenience.
+
+```
+pub struct SafekeeperInfo {
+    pub id: NodeId,
+    pub hostname: String,
+}
+
+pub struct SafekeepersNotifyRequest {
+    pub tenant_id: TenantId,
+    pub timeline_id: TimelineId,
+    pub generation: u32,
+    pub safekeepers: Vec<SafekeeperInfo>,
+}
+```
+
+When a notification is received:
+
+1. Modify postgres configuration for this tenant:
+
+   - set `neon.safekeeper_connstrings` to an array of postgres connection strings to safekeepers according to the `safekeepers` list. The
+     safekeepers identified by `NodeId` must be converted to the address+port of the respective safekeeper.
+     The hostname is provided for debugging purposes, so we reserve changes to how we pass it.
+   - set `neon.safekeepers_generation` to the provided `generation` value.
+
+2. Send SIGHUP to postgres to reload configuration
+3. Respond with 200 to the notification request. Do not return success if postgres was not updated: if an error is returned, the controller
+   will retry the notification until it succeeds..
--- a/libs/compute_api/Cargo.toml
+++ b/libs/compute_api/Cargo.toml
@@ -7,6 +7,7 @@ license.workspace = true
 [dependencies]
 anyhow.workspace = true
 chrono.workspace = true
+indexmap.workspace = true
 jsonwebtoken.workspace = true
 serde.workspace = true
 serde_json.workspace = true
--- a/libs/compute_api/src/requests.rs
+++ b/libs/compute_api/src/requests.rs
@@ -30,3 +30,9 @@ pub struct SetRoleGrantsRequest {
    pub privileges: Vec<Privilege>,
    pub role: PgIdent,
 }
+
+/// Request of the /configure_telemetry API
+#[derive(Debug, Deserialize, Serialize)]
+pub struct ConfigureTelemetryRequest {
+    pub logs_export_host: Option<String>,
+}
--- a/libs/compute_api/src/responses.rs
+++ b/libs/compute_api/src/responses.rs
@@ -139,6 +139,7 @@ pub struct ComputeCtlConfig {
    /// Set of JSON web keys that the compute can use to authenticate
    /// communication from the control plane.
    pub jwks: JwkSet,
+    pub tls: Option<TlsConfig>,
 }

 impl Default for ComputeCtlConfig {
@@ -147,10 +148,17 @@ impl Default for ComputeCtlConfig {
            jwks: JwkSet {
                keys: Vec::default(),
            },
+            tls: None,
        }
    }
 }

+#[derive(Clone, Debug, Deserialize, Serialize)]
+pub struct TlsConfig {
+    pub key_path: String,
+    pub cert_path: String,
+}
+
 /// Response of the `/computes/{compute_id}/spec` control-plane API.
 #[derive(Deserialize, Debug)]
 pub struct ControlPlaneSpecResponse {
--- a/libs/compute_api/src/spec.rs
+++ b/libs/compute_api/src/spec.rs
@@ -5,12 +5,15 @@
 //! and connect it to the storage nodes.
 use std::collections::HashMap;

+use indexmap::IndexMap;
 use regex::Regex;
 use remote_storage::RemotePath;
 use serde::{Deserialize, Serialize};
 use utils::id::{TenantId, TimelineId};
 use utils::lsn::Lsn;

+use crate::responses::TlsConfig;
+
 /// String type alias representing Postgres identifier and
 /// intended to be used for DB / role names.
 pub type PgIdent = String;
@@ -125,7 +128,7 @@ pub struct ComputeSpec {
    // information about available remote extensions
    pub remote_extensions: Option<RemoteExtSpec>,

-    pub pgbouncer_settings: Option<HashMap<String, String>>,
+    pub pgbouncer_settings: Option<IndexMap<String, String>>,

    // Stripe size for pageserver sharding, in pages
    #[serde(default)]
@@ -176,8 +179,8 @@ pub enum ComputeFeature {
    /// track short-lived connections as user activity.
    ActivityMonitorExperimental,

-    /// Pre-install and initialize anon extension for every database in the cluster
-    AnonExtension,
+    /// Allow to configure rsyslog for Postgres logs export
+    PostgresLogsExport,

    /// This is a special feature flag that is used to represent unknown feature flags.
    /// Basically all unknown to enum flags are represented as this one. See unit test
@@ -357,6 +360,9 @@ pub struct LocalProxySpec {
    #[serde(default)]
    #[serde(skip_serializing_if = "Option::is_none")]
    pub jwks: Option<Vec<JwksSettings>>,
+    #[serde(default)]
+    #[serde(skip_serializing_if = "Option::is_none")]
+    pub tls: Option<TlsConfig>,
 }

 #[derive(Clone, Debug, Deserialize, Serialize)]
--- a/libs/compute_api/tests/cluster_spec.json
+++ b/libs/compute_api/tests/cluster_spec.json
@@ -208,7 +208,6 @@
    ],
    "remote_extensions": {
        "library_index": {
-          "anon": "anon",
          "postgis-3": "postgis",
          "libpgrouting-3.4": "postgis",
          "postgis_raster-3": "postgis",
@@ -217,12 +216,6 @@
          "address_standardizer-3": "postgis"
        },
        "extension_data": {
-          "anon": {
-            "archive_path": "5834329303/v15/extensions/anon.tar.zst",
-            "control_data": {
-              "anon.control": "# PostgreSQL Anonymizer (anon) extension\ncomment = ''Data anonymization tools''\ndefault_version = ''1.1.0''\ndirectory=''extension/anon''\nrelocatable = false\nrequires = ''pgcrypto''\nsuperuser = false\nmodule_pathname = ''$libdir/anon''\ntrusted = true\n"
-            }
-          },
          "postgis": {
            "archive_path": "5834329303/v15/extensions/postgis.tar.zst",
            "control_data": {
@@ -238,7 +231,6 @@
          }
        },
        "custom_extensions": [
-          "anon"
        ],
        "public_extensions": [
          "postgis"
--- a/libs/http-utils/Cargo.toml
+++ b/libs/http-utils/Cargo.toml
@@ -7,6 +7,7 @@ license.workspace = true
 [dependencies]
 anyhow.workspace = true
 bytes.workspace = true
+camino.workspace = true
 fail.workspace = true
 futures.workspace = true
 hyper0.workspace = true
@@ -16,6 +17,7 @@ once_cell.workspace = true
 pprof.workspace = true
 regex.workspace = true
 routerify.workspace = true
+rustls-pemfile.workspace = true
 serde.workspace = true
 serde_json.workspace = true
 serde_path_to_error.workspace = true
--- a/libs/http-utils/src/lib.rs
+++ b/libs/http-utils/src/lib.rs
@@ -4,6 +4,7 @@ pub mod failpoints;
 pub mod json;
 pub mod request;
 pub mod server;
+pub mod tls_certs;

 extern crate hyper0 as hyper;

--- a/libs/http-utils/src/tls_certs.rs
+++ b/libs/http-utils/src/tls_certs.rs
@@ -0,0 +1,21 @@
+use camino::Utf8Path;
+use tokio_rustls::rustls::pki_types::{CertificateDer, PrivateKeyDer};
+
+pub fn load_cert_chain(filename: &Utf8Path) -> anyhow::Result<Vec<CertificateDer<'static>>> {
+    let file = std::fs::File::open(filename)?;
+    let mut reader = std::io::BufReader::new(file);
+
+    Ok(rustls_pemfile::certs(&mut reader).collect::<Result<Vec<_>, _>>()?)
+}
+
+pub fn load_private_key(filename: &Utf8Path) -> anyhow::Result<PrivateKeyDer<'static>> {
+    let file = std::fs::File::open(filename)?;
+    let mut reader = std::io::BufReader::new(file);
+
+    let key = rustls_pemfile::private_key(&mut reader)?;
+
+    key.ok_or(anyhow::anyhow!(
+        "no private key found in {}",
+        filename.as_str(),
+    ))
+}
--- a/libs/pageserver_api/src/config.rs
+++ b/libs/pageserver_api/src/config.rs
@@ -272,15 +272,16 @@ pub struct TenantConfigToml {
    /// size exceeds `compaction_upper_limit * checkpoint_distance`.
    pub compaction_upper_limit: usize,
    pub compaction_algorithm: crate::models::CompactionAlgorithmSettings,
-    /// If true, compact down L0 across all tenant timelines before doing regular compaction.
+    /// If true, compact down L0 across all tenant timelines before doing regular compaction. L0
+    /// compaction must be responsive to avoid read amp during heavy ingestion. Defaults to true.
    pub compaction_l0_first: bool,
    /// If true, use a separate semaphore (i.e. concurrency limit) for the L0 compaction pass. Only
-    /// has an effect if `compaction_l0_first` is `true`.
+    /// has an effect if `compaction_l0_first` is true. Defaults to true.
    pub compaction_l0_semaphore: bool,
-    /// Level0 delta layer threshold at which to delay layer flushes for compaction backpressure,
-    /// such that they take 2x as long, and start waiting for layer flushes during ephemeral layer
-    /// rolls. This helps compaction keep up with WAL ingestion, and avoids read amplification
-    /// blowing up. Should be >compaction_threshold. 0 to disable. Disabled by default.
+    /// Level0 delta layer threshold at which to delay layer flushes such that they take 2x as long,
+    /// and block on layer flushes during ephemeral layer rolls, for compaction backpressure. This
+    /// helps compaction keep up with WAL ingestion, and avoids read amplification blowing up.
+    /// Should be >compaction_threshold. 0 to disable. Defaults to 3x compaction_threshold.
    pub l0_flush_delay_threshold: Option<usize>,
    /// Level0 delta layer threshold at which to stall layer flushes. Must be >compaction_threshold
    /// to avoid deadlock. 0 to disable. Disabled by default.
@@ -288,6 +289,8 @@ pub struct TenantConfigToml {
    /// If true, Level0 delta layer flushes will wait for S3 upload before flushing the next
    /// layer. This is a temporary backpressure mechanism which should be removed once
    /// l0_flush_{delay,stall}_threshold is fully enabled.
+    ///
+    /// TODO: this is no longer enabled, remove it when the config option is no longer set.
    pub l0_flush_wait_upload: bool,
    // Determines how much history is retained, to allow
    // branching and read replicas at an older point in time.
@@ -567,13 +570,15 @@ pub mod tenant_conf_defaults {
    // be reduced later by optimizing L0 hole calculation to avoid loading all keys into memory). So
    // with this config, we can get a maximum peak compaction usage of 9 GB.
    pub const DEFAULT_COMPACTION_UPPER_LIMIT: usize = 20;
-    pub const DEFAULT_COMPACTION_L0_FIRST: bool = false;
+    // Enable L0 compaction pass and semaphore by default. L0 compaction must be responsive to avoid
+    // read amp.
+    pub const DEFAULT_COMPACTION_L0_FIRST: bool = true;
    pub const DEFAULT_COMPACTION_L0_SEMAPHORE: bool = true;

    pub const DEFAULT_COMPACTION_ALGORITHM: crate::models::CompactionAlgorithm =
        crate::models::CompactionAlgorithm::Legacy;

-    pub const DEFAULT_L0_FLUSH_WAIT_UPLOAD: bool = true;
+    pub const DEFAULT_L0_FLUSH_WAIT_UPLOAD: bool = false;

    pub const DEFAULT_GC_HORIZON: u64 = 64 * 1024 * 1024;

@@ -584,9 +589,8 @@ pub mod tenant_conf_defaults {
    pub const DEFAULT_GC_PERIOD: &str = "1 hr";
    pub const DEFAULT_IMAGE_CREATION_THRESHOLD: usize = 3;
    // If there are more than threshold * compaction_threshold (that is 3 * 10 in the default config) L0 layers, image
-    // layer creation will end immediately. Set to 0 to disable. The target default will be 3 once we
-    // want to enable this feature.
-    pub const DEFAULT_IMAGE_CREATION_PREEMPT_THRESHOLD: usize = 0;
+    // layer creation will end immediately. Set to 0 to disable.
+    pub const DEFAULT_IMAGE_CREATION_PREEMPT_THRESHOLD: usize = 3;
    pub const DEFAULT_PITR_INTERVAL: &str = "7 days";
    pub const DEFAULT_WALRECEIVER_CONNECT_TIMEOUT: &str = "10 seconds";
    pub const DEFAULT_WALRECEIVER_LAGGING_WAL_TIMEOUT: &str = "10 seconds";
--- a/libs/pageserver_api/src/models.rs
+++ b/libs/pageserver_api/src/models.rs
@@ -176,6 +176,39 @@ impl LsnLease {
    }
 }

+/// Controls the detach ancestor behavior.
+/// - When set to `NoAncestorAndReparent`, we will only detach a branch if its ancestor is a root branch. It will automatically reparent any children of the ancestor before and at the branch point.
+/// - When set to `MultiLevelAndNoReparent`, we will detach a branch from multiple levels of ancestors, and no reparenting will happen at all.
+#[derive(Debug, Clone, Copy, Default)]
+pub enum DetachBehavior {
+    #[default]
+    NoAncestorAndReparent,
+    MultiLevelAndNoReparent,
+}
+
+impl std::str::FromStr for DetachBehavior {
+    type Err = &'static str;
+
+    fn from_str(s: &str) -> Result<Self, Self::Err> {
+        match s {
+            "no_ancestor_and_reparent" => Ok(DetachBehavior::NoAncestorAndReparent),
+            "multi_level_and_no_reparent" => Ok(DetachBehavior::MultiLevelAndNoReparent),
+            "v1" => Ok(DetachBehavior::NoAncestorAndReparent),
+            "v2" => Ok(DetachBehavior::MultiLevelAndNoReparent),
+            _ => Err("cannot parse detach behavior"),
+        }
+    }
+}
+
+impl std::fmt::Display for DetachBehavior {
+    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
+        match self {
+            DetachBehavior::NoAncestorAndReparent => write!(f, "no_ancestor_and_reparent"),
+            DetachBehavior::MultiLevelAndNoReparent => write!(f, "multi_level_and_no_reparent"),
+        }
+    }
+}
+
 /// The only [`TenantState`] variants we could be `TenantState::Activating` from.
 ///
 /// XXX: We used to have more variants here, but now it's just one, which makes this rather
--- a/libs/safekeeper_api/src/models.rs
+++ b/libs/safekeeper_api/src/models.rs
@@ -23,6 +23,7 @@ pub struct TimelineCreateRequest {
    pub tenant_id: TenantId,
    pub timeline_id: TimelineId,
    pub mconf: Configuration,
+    /// In the PG_VERSION_NUM macro format, like 140017.
    pub pg_version: u32,
    pub system_id: Option<u64>,
    // By default WAL_SEGMENT_SIZE
@@ -221,6 +222,11 @@ pub struct TimelineMembershipSwitchResponse {
    pub current_conf: Configuration,
 }

+#[derive(Clone, Copy, Serialize, Deserialize)]
+pub struct TimelineDeleteResult {
+    pub dir_existed: bool,
+}
+
 fn lsn_invalid() -> Lsn {
    Lsn::INVALID
 }
--- a/libs/utils/benches/benchmarks.rs
+++ b/libs/utils/benches/benchmarks.rs
@@ -49,7 +49,13 @@ pub fn bench_log_slow(c: &mut Criterion) {
        // performance too. Use a simple noop future that yields once, to avoid any scheduler fast
        // paths for a ready future.
        if enabled {
-            b.iter(|| runtime.block_on(log_slow("ready", THRESHOLD, tokio::task::yield_now())));
+            b.iter(|| {
+                runtime.block_on(log_slow(
+                    "ready",
+                    THRESHOLD,
+                    std::pin::pin!(tokio::task::yield_now()),
+                ))
+            });
        } else {
            b.iter(|| runtime.block_on(tokio::task::yield_now()));
        }
--- a/libs/utils/src/logging.rs
+++ b/libs/utils/src/logging.rs
@@ -331,37 +331,90 @@ impl std::fmt::Debug for SecretString {
 ///
 /// TODO: consider upgrading this to a warning, but currently it fires too often.
 #[inline]
-pub async fn log_slow<O>(name: &str, threshold: Duration, f: impl Future<Output = O>) -> O {
-    // TODO: we unfortunately have to pin the future on the heap, since GetPage futures are huge and
-    // won't fit on the stack.
-    let mut f = Box::pin(f);
+pub async fn log_slow<F, O>(name: &str, threshold: Duration, f: std::pin::Pin<&mut F>) -> O
+where
+    F: Future<Output = O>,
+{
+    monitor_slow_future(
+        threshold,
+        threshold, // period = threshold
+        f,
+        |MonitorSlowFutureCallback {
+             ready,
+             is_slow,
+             elapsed_total,
+             elapsed_since_last_callback: _,
+         }| {
+            if !is_slow {
+                return;
+            }
+            if ready {
+                info!(
+                    "slow {name} completed after {:.3}s",
+                    elapsed_total.as_secs_f64()
+                );
+            } else {
+                info!(
+                    "slow {name} still running after {:.3}s",
+                    elapsed_total.as_secs_f64()
+                );
+            }
+        },
+    )
+    .await
+}

+/// Poll future `fut` to completion, invoking callback `cb` at the given `threshold` and every
+/// `period` afterwards, and also unconditionally when the future completes.
+#[inline]
+pub async fn monitor_slow_future<F, O>(
+    threshold: Duration,
+    period: Duration,
+    mut fut: std::pin::Pin<&mut F>,
+    mut cb: impl FnMut(MonitorSlowFutureCallback),
+) -> O
+where
+    F: Future<Output = O>,
+{
    let started = Instant::now();
    let mut attempt = 1;
-
+    let mut last_cb = started;
    loop {
        // NB: use timeout_at() instead of timeout() to avoid an extra clock reading in the common
        // case where the timeout doesn't fire.
-        let deadline = started + attempt * threshold;
-        if let Ok(output) = tokio::time::timeout_at(deadline, &mut f).await {
-            // NB: we check if we exceeded the threshold even if the timeout never fired, because
-            // scheduling or execution delays may cause the future to succeed even if it exceeds the
-            // timeout. This costs an extra unconditional clock reading, but seems worth it to avoid
-            // false negatives.
-            let elapsed = started.elapsed();
-            if elapsed >= threshold {
-                info!("slow {name} completed after {:.3}s", elapsed.as_secs_f64());
-            }
+        let deadline = started + threshold + (attempt - 1) * period;
+        // TODO: still call the callback if the future panics? Copy how we do it for the page_service flush_in_progress counter.
+        let res = tokio::time::timeout_at(deadline, &mut fut).await;
+        let now = Instant::now();
+        let elapsed_total = now - started;
+        cb(MonitorSlowFutureCallback {
+            ready: res.is_ok(),
+            is_slow: elapsed_total >= threshold,
+            elapsed_total,
+            elapsed_since_last_callback: now - last_cb,
+        });
+        last_cb = now;
+        if let Ok(output) = res {
            return output;
        }
-
-        let elapsed = started.elapsed().as_secs_f64();
-        info!("slow {name} still running after {elapsed:.3}s",);
-
        attempt += 1;
    }
 }

+/// See [`monitor_slow_future`].
+pub struct MonitorSlowFutureCallback {
+    /// Whether the future completed. If true, there will be no more callbacks.
+    pub ready: bool,
+    /// Whether the future is taking `>=` the specififed threshold duration to complete.
+    /// Monotonic: if true in one callback invocation, true in all subsequent onces.
+    pub is_slow: bool,
+    /// The time elapsed since the [`monitor_slow_future`] was first polled.
+    pub elapsed_total: Duration,
+    /// The time elapsed since the last callback invocation.
+    /// For the initial callback invocation, the time elapsed since the [`monitor_slow_future`] was first polled.
+    pub elapsed_since_last_callback: Duration,
+}
+
 #[cfg(test)]
 mod tests {
    use metrics::IntCounterVec;
--- a/libs/utils/src/sync/duplex/mpsc.rs
+++ b/libs/utils/src/sync/duplex/mpsc.rs
@@ -26,6 +26,10 @@ impl<S: Send, R: Send> Duplex<S, R> {
        self.tx.send(x).await
    }

+    pub fn try_send(&self, x: S) -> Result<(), mpsc::error::TrySendError<S>> {
+        self.tx.try_send(x)
+    }
+
    /// Receives the next value for this receiver.
    ///
    /// This method returns `None` if the channel has been closed and there are
--- a/pageserver/Cargo.toml
+++ b/pageserver/Cargo.toml
@@ -48,8 +48,6 @@ pprof.workspace = true
 rand.workspace = true
 range-set-blaze = { version = "0.1.16", features = ["alloc"] }
 regex.workspace = true
-rustls-pemfile.workspace = true
-rustls-pki-types.workspace = true
 rustls.workspace = true
 scopeguard.workspace = true
 send-future.workspace = true
--- a/pageserver/benches/bench_ingest.rs
+++ b/pageserver/benches/bench_ingest.rs
@@ -13,6 +13,7 @@ use pageserver::{page_cache, virtual_file};
 use pageserver_api::key::Key;
 use pageserver_api::shard::TenantShardId;
 use pageserver_api::value::Value;
+use tokio_util::sync::CancellationToken;
 use utils::bin_ser::BeSer;
 use utils::id::{TenantId, TimelineId};
 use wal_decoder::serialized_batch::SerializedValueBatch;
@@ -57,11 +58,22 @@ async fn ingest(

    tokio::fs::create_dir_all(conf.timeline_path(&tenant_shard_id, &timeline_id)).await?;

-    let ctx = RequestContext::new(TaskKind::DebugTool, DownloadBehavior::Error);
+    let ctx =
+        RequestContext::new(TaskKind::DebugTool, DownloadBehavior::Error).with_scope_debug_tools();

    let gate = utils::sync::gate::Gate::default();
+    let cancel = CancellationToken::new();

-    let layer = InMemoryLayer::create(conf, timeline_id, tenant_shard_id, lsn, &gate, &ctx).await?;
+    let layer = InMemoryLayer::create(
+        conf,
+        timeline_id,
+        tenant_shard_id,
+        lsn,
+        &gate,
+        &cancel,
+        &ctx,
+    )
+    .await?;

    let data = Value::Image(Bytes::from(vec![0u8; put_size]));
    let data_ser_size = data.serialized_size().unwrap() as usize;
--- a/pageserver/client/src/mgmt_api.rs
+++ b/pageserver/client/src/mgmt_api.rs
@@ -7,7 +7,7 @@ use http_utils::error::HttpErrorBody;
 use pageserver_api::models::*;
 use pageserver_api::shard::TenantShardId;
 pub use reqwest::Body as ReqwestBody;
-use reqwest::{Certificate, IntoUrl, Method, StatusCode};
+use reqwest::{Certificate, IntoUrl, Method, StatusCode, Url};
 use utils::id::{TenantId, TimelineId};
 use utils::lsn::Lsn;

@@ -458,13 +458,21 @@ impl Client {
        &self,
        tenant_shard_id: TenantShardId,
        timeline_id: TimelineId,
+        behavior: Option<DetachBehavior>,
    ) -> Result<AncestorDetached> {
        let uri = format!(
            "{}/v1/tenant/{tenant_shard_id}/timeline/{timeline_id}/detach_ancestor",
            self.mgmt_api_endpoint
        );
+        let mut uri = Url::parse(&uri)
+            .map_err(|e| Error::ApiError(StatusCode::INTERNAL_SERVER_ERROR, format!("{e}")))?;

-        self.request(Method::PUT, &uri, ())
+        if let Some(behavior) = behavior {
+            uri.query_pairs_mut()
+                .append_pair("detach_behavior", &behavior.to_string());
+        }
+
+        self.request(Method::PUT, uri, ())
            .await?
            .json()
            .await
--- a/pageserver/ctl/src/layer_map_analyzer.rs
+++ b/pageserver/ctl/src/layer_map_analyzer.rs
@@ -131,7 +131,8 @@ async fn get_holes(path: &Utf8Path, max_holes: usize, ctx: &RequestContext) -> R
 pub(crate) async fn main(cmd: &AnalyzeLayerMapCmd) -> Result<()> {
    let storage_path = &cmd.path;
    let max_holes = cmd.max_holes.unwrap_or(DEFAULT_MAX_HOLES);
-    let ctx = RequestContext::new(TaskKind::DebugTool, DownloadBehavior::Error);
+    let ctx =
+        RequestContext::new(TaskKind::DebugTool, DownloadBehavior::Error).with_scope_debug_tools();

    // Initialize virtual_file (file desriptor cache) and page cache which are needed to access layer persistent B-Tree.
    pageserver::virtual_file::init(
--- a/pageserver/ctl/src/layers.rs
+++ b/pageserver/ctl/src/layers.rs
@@ -76,7 +76,8 @@ async fn read_image_file(path: impl AsRef<Path>, ctx: &RequestContext) -> Result
 }

 pub(crate) async fn main(cmd: &LayerCmd) -> Result<()> {
-    let ctx = RequestContext::new(TaskKind::DebugTool, DownloadBehavior::Error);
+    let ctx =
+        RequestContext::new(TaskKind::DebugTool, DownloadBehavior::Error).with_scope_debug_tools();
    match cmd {
        LayerCmd::List { path } => {
            for tenant in fs::read_dir(path.join(TENANTS_SEGMENT_NAME))? {
@@ -176,7 +177,8 @@ pub(crate) async fn main(cmd: &LayerCmd) -> Result<()> {
            );
            pageserver::page_cache::init(100);

-            let ctx = RequestContext::new(TaskKind::DebugTool, DownloadBehavior::Error);
+            let ctx = RequestContext::new(TaskKind::DebugTool, DownloadBehavior::Error)
+                .with_scope_debug_tools();

            macro_rules! rewrite_closure {
                ($($summary_ty:tt)*) => {{
--- a/pageserver/ctl/src/main.rs
+++ b/pageserver/ctl/src/main.rs
@@ -208,7 +208,8 @@ async fn print_layerfile(path: &Utf8Path) -> anyhow::Result<()> {
        virtual_file::SyncMode::Sync,
    );
    page_cache::init(100);
-    let ctx = RequestContext::new(TaskKind::DebugTool, DownloadBehavior::Error);
+    let ctx =
+        RequestContext::new(TaskKind::DebugTool, DownloadBehavior::Error).with_scope_debug_tools();
    dump_layerfile_from_path(path, true, &ctx).await
 }

--- a/pageserver/src/bin/pageserver.rs
+++ b/pageserver/src/bin/pageserver.rs
@@ -30,7 +30,6 @@ use pageserver::{
 };
 use postgres_backend::AuthType;
 use remote_storage::GenericRemoteStorage;
-use rustls_pki_types::{CertificateDer, PrivateKeyDer};
 use tokio::signal::unix::SignalKind;
 use tokio::time::Instant;
 use tokio_util::sync::CancellationToken;
@@ -622,8 +621,8 @@ fn start_pageserver(

        let https_task = match https_listener {
            Some(https_listener) => {
-                let certs = load_certs(&conf.ssl_cert_file)?;
-                let key = load_private_key(&conf.ssl_key_file)?;
+                let certs = http_utils::tls_certs::load_cert_chain(&conf.ssl_cert_file)?;
+                let key = http_utils::tls_certs::load_private_key(&conf.ssl_key_file)?;

                let server_config = rustls::ServerConfig::builder()
                    .with_no_client_auth()
@@ -735,25 +734,6 @@ fn start_pageserver(
    })
 }

-fn load_certs(filename: &Utf8Path) -> std::io::Result<Vec<CertificateDer<'static>>> {
-    let file = std::fs::File::open(filename)?;
-    let mut reader = std::io::BufReader::new(file);
-
-    rustls_pemfile::certs(&mut reader).collect()
-}
-
-fn load_private_key(filename: &Utf8Path) -> anyhow::Result<PrivateKeyDer<'static>> {
-    let file = std::fs::File::open(filename)?;
-    let mut reader = std::io::BufReader::new(file);
-
-    let key = rustls_pemfile::private_key(&mut reader)?;
-
-    key.ok_or(anyhow::anyhow!(
-        "no private key found in {}",
-        filename.as_str(),
-    ))
-}
-
 async fn create_remote_storage_client(
    conf: &'static PageServerConf,
 ) -> anyhow::Result<GenericRemoteStorage> {
--- a/pageserver/src/context.rs
+++ b/pageserver/src/context.rs
@@ -134,6 +134,9 @@ pub(crate) enum Scope {
    UnitTest {
        io_size_metrics: &'static crate::metrics::StorageIoSizeMetrics,
    },
+    DebugTools {
+        io_size_metrics: &'static crate::metrics::StorageIoSizeMetrics,
+    },
 }

 static GLOBAL_IO_SIZE_METRICS: Lazy<crate::metrics::StorageIoSizeMetrics> =
@@ -195,6 +198,12 @@ impl Scope {
            io_size_metrics: &GLOBAL_IO_SIZE_METRICS,
        }
    }
+
+    pub(crate) fn new_debug_tools() -> Self {
+        Scope::DebugTools {
+            io_size_metrics: &GLOBAL_IO_SIZE_METRICS,
+        }
+    }
 }

 /// The kind of access to the page cache.
@@ -435,6 +444,12 @@ impl RequestContext {
            .build()
    }

+    pub fn with_scope_debug_tools(&self) -> Self {
+        RequestContextBuilder::new(TaskKind::DebugTool)
+            .scope(Scope::new_debug_tools())
+            .build()
+    }
+
    pub fn task_kind(&self) -> TaskKind {
        self.task_kind
    }
@@ -486,6 +501,7 @@ impl RequestContext {
            Scope::SecondaryTenant { io_size_metrics } => io_size_metrics,
            #[cfg(test)]
            Scope::UnitTest { io_size_metrics } => io_size_metrics,
+            Scope::DebugTools { io_size_metrics } => io_size_metrics,
        }
    }
 }
--- a/pageserver/src/http/routes.rs
+++ b/pageserver/src/http/routes.rs
@@ -28,9 +28,9 @@ use hyper::{Body, Request, Response, StatusCode, Uri, header};
 use metrics::launch_timestamp::LaunchTimestamp;
 use pageserver_api::models::virtual_file::IoMode;
 use pageserver_api::models::{
-    DownloadRemoteLayersTaskSpawnRequest, IngestAuxFilesRequest, ListAuxFilesRequest,
-    LocationConfig, LocationConfigListResponse, LocationConfigMode, LsnLease, LsnLeaseRequest,
-    OffloadedTimelineInfo, PageTraceEvent, ShardParameters, StatusResponse,
+    DetachBehavior, DownloadRemoteLayersTaskSpawnRequest, IngestAuxFilesRequest,
+    ListAuxFilesRequest, LocationConfig, LocationConfigListResponse, LocationConfigMode, LsnLease,
+    LsnLeaseRequest, OffloadedTimelineInfo, PageTraceEvent, ShardParameters, StatusResponse,
    TenantConfigPatchRequest, TenantConfigRequest, TenantDetails, TenantInfo,
    TenantLocationConfigRequest, TenantLocationConfigResponse, TenantScanRemoteStorageResponse,
    TenantScanRemoteStorageShard, TenantShardLocation, TenantShardSplitRequest,
@@ -72,7 +72,6 @@ use crate::tenant::remote_timeline_client::{
 use crate::tenant::secondary::SecondaryController;
 use crate::tenant::size::ModelInputs;
 use crate::tenant::storage_layer::{IoConcurrency, LayerAccessStatsReset, LayerName};
-use crate::tenant::timeline::detach_ancestor::DetachBehavior;
 use crate::tenant::timeline::offload::{OffloadError, offload_timeline};
 use crate::tenant::timeline::{
    CompactFlags, CompactOptions, CompactRequest, CompactionError, Timeline, WaitLsnTimeout,
@@ -2392,6 +2391,7 @@ async fn timeline_checkpoint_handler(
    let state = get_state(&request);

    let mut flags = EnumSet::empty();
+    flags |= CompactFlags::NoYield; // run compaction to completion
    if Some(true) == parse_query_param::<_, bool>(&request, "force_l0_compaction")? {
        flags |= CompactFlags::ForceL0Compaction;
    }
@@ -2507,6 +2507,7 @@ async fn timeline_detach_ancestor_handler(
    check_permission(&request, Some(tenant_shard_id.tenant_id))?;
    let timeline_id: TimelineId = parse_request_param(&request, "timeline_id")?;
    let behavior: Option<DetachBehavior> = parse_query_param(&request, "detach_behavior")?;
+
    let behavior = behavior.unwrap_or_default();

    let span = tracing::info_span!("detach_ancestor", tenant_id=%tenant_shard_id.tenant_id, shard_id=%tenant_shard_id.shard_slug(), %timeline_id);
--- a/pageserver/src/metrics.rs
+++ b/pageserver/src/metrics.rs
@@ -465,12 +465,40 @@ pub(crate) fn page_cache_errors_inc(error_kind: PageCacheErrorKind) {
 pub(crate) static WAIT_LSN_TIME: Lazy<Histogram> = Lazy::new(|| {
    register_histogram!(
        "pageserver_wait_lsn_seconds",
-        "Time spent waiting for WAL to arrive",
+        "Time spent waiting for WAL to arrive. Updated on completion of the wait_lsn operation.",
        CRITICAL_OP_BUCKETS.into(),
    )
    .expect("failed to define a metric")
 });

+pub(crate) static WAIT_LSN_START_FINISH_COUNTERPAIR: Lazy<IntCounterPairVec> = Lazy::new(|| {
+    register_int_counter_pair_vec!(
+        "pageserver_wait_lsn_started_count",
+        "Number of wait_lsn operations started.",
+        "pageserver_wait_lsn_finished_count",
+        "Number of wait_lsn operations finished.",
+        &["tenant_id", "shard_id", "timeline_id"],
+    )
+    .expect("failed to define a metric")
+});
+
+pub(crate) static WAIT_LSN_IN_PROGRESS_MICROS: Lazy<IntCounterVec> = Lazy::new(|| {
+    register_int_counter_vec!(
+        "pageserver_wait_lsn_in_progress_micros",
+        "Time spent waiting for WAL to arrive, by timeline_id. Updated periodically while waiting.",
+        &["tenant_id", "shard_id", "timeline_id"],
+    )
+    .expect("failed to define a metric")
+});
+
+pub(crate) static WAIT_LSN_IN_PROGRESS_GLOBAL_MICROS: Lazy<IntCounter> = Lazy::new(|| {
+    register_int_counter!(
+        "pageserver_wait_lsn_in_progress_micros_global",
+        "Time spent waiting for WAL to arrive, globally. Updated periodically while waiting."
+    )
+    .expect("failed to define a metric")
+});
+
 static FLUSH_WAIT_UPLOAD_TIME: Lazy<GaugeVec> = Lazy::new(|| {
    register_gauge_vec!(
        "pageserver_flush_wait_upload_seconds",
@@ -2830,7 +2858,6 @@ impl StorageTimeMetrics {
    }
 }

-#[derive(Debug)]
 pub(crate) struct TimelineMetrics {
    tenant_id: String,
    shard_id: String,
@@ -2863,6 +2890,8 @@ pub(crate) struct TimelineMetrics {
    pub valid_lsn_lease_count_gauge: UIntGauge,
    pub wal_records_received: IntCounter,
    pub storage_io_size: StorageIoSizeMetrics,
+    pub wait_lsn_in_progress_micros: GlobalAndPerTenantIntCounter,
+    pub wait_lsn_start_finish_counterpair: IntCounterPair,
    shutdown: std::sync::atomic::AtomicBool,
 }

@@ -3000,6 +3029,17 @@ impl TimelineMetrics {

        let storage_io_size = StorageIoSizeMetrics::new(&tenant_id, &shard_id, &timeline_id);

+        let wait_lsn_in_progress_micros = GlobalAndPerTenantIntCounter {
+            global: WAIT_LSN_IN_PROGRESS_GLOBAL_MICROS.clone(),
+            per_tenant: WAIT_LSN_IN_PROGRESS_MICROS
+                .get_metric_with_label_values(&[&tenant_id, &shard_id, &timeline_id])
+                .unwrap(),
+        };
+
+        let wait_lsn_start_finish_counterpair = WAIT_LSN_START_FINISH_COUNTERPAIR
+            .get_metric_with_label_values(&[&tenant_id, &shard_id, &timeline_id])
+            .unwrap();
+
        TimelineMetrics {
            tenant_id,
            shard_id,
@@ -3032,6 +3072,8 @@ impl TimelineMetrics {
            storage_io_size,
            valid_lsn_lease_count_gauge,
            wal_records_received,
+            wait_lsn_in_progress_micros,
+            wait_lsn_start_finish_counterpair,
            shutdown: std::sync::atomic::AtomicBool::default(),
        }
    }
@@ -3224,6 +3266,15 @@ impl TimelineMetrics {
            let _ = STORAGE_IO_SIZE.remove_label_values(&[op, tenant_id, shard_id, timeline_id]);
        }

+        let _ =
+            WAIT_LSN_IN_PROGRESS_MICROS.remove_label_values(&[tenant_id, shard_id, timeline_id]);
+
+        {
+            let mut res = [Ok(()), Ok(())];
+            WAIT_LSN_START_FINISH_COUNTERPAIR
+                .remove_label_values(&mut res, &[tenant_id, shard_id, timeline_id]);
+        }
+
        let _ = SMGR_QUERY_STARTED_PER_TENANT_TIMELINE.remove_label_values(&[
            SmgrQueryType::GetPageAtLsn.into(),
            tenant_id,
@@ -3836,27 +3887,29 @@ pub mod tokio_epoll_uring {
    });
 }

+pub(crate) struct GlobalAndPerTenantIntCounter {
+    global: IntCounter,
+    per_tenant: IntCounter,
+}
+
+impl GlobalAndPerTenantIntCounter {
+    #[inline(always)]
+    pub(crate) fn inc(&self) {
+        self.inc_by(1)
+    }
+    #[inline(always)]
+    pub(crate) fn inc_by(&self, n: u64) {
+        self.global.inc_by(n);
+        self.per_tenant.inc_by(n);
+    }
+}
+
 pub(crate) mod tenant_throttling {
-    use metrics::{IntCounter, register_int_counter_vec};
+    use metrics::register_int_counter_vec;
    use once_cell::sync::Lazy;
    use utils::shard::TenantShardId;

-    pub(crate) struct GlobalAndPerTenantIntCounter {
-        global: IntCounter,
-        per_tenant: IntCounter,
-    }
-
-    impl GlobalAndPerTenantIntCounter {
-        #[inline(always)]
-        pub(crate) fn inc(&self) {
-            self.inc_by(1)
-        }
-        #[inline(always)]
-        pub(crate) fn inc_by(&self, n: u64) {
-            self.global.inc_by(n);
-            self.per_tenant.inc_by(n);
-        }
-    }
+    use super::GlobalAndPerTenantIntCounter;

    pub(crate) struct Metrics<const KIND: usize> {
        pub(super) count_accounted_start: GlobalAndPerTenantIntCounter,
@@ -4102,6 +4155,7 @@ pub fn preinitialize_metrics(conf: &'static PageServerConf) {
        &CIRCUIT_BREAKERS_BROKEN,
        &CIRCUIT_BREAKERS_UNBROKEN,
        &PAGE_SERVICE_SMGR_FLUSH_INPROGRESS_MICROS_GLOBAL,
+        &WAIT_LSN_IN_PROGRESS_GLOBAL_MICROS,
    ]
    .into_iter()
    .for_each(|c| {
--- a/pageserver/src/page_service.rs
+++ b/pageserver/src/page_service.rs
@@ -1106,12 +1106,19 @@ impl PageServerHandler {
        };

        // Dispatch the batch to the appropriate request handler.
-        let (mut handler_results, span) = log_slow(
-            batch.as_static_str(),
-            LOG_SLOW_GETPAGE_THRESHOLD,
-            self.pagestream_dispatch_batched_message(batch, io_concurrency, ctx),
-        )
-        .await?;
+        let log_slow_name = batch.as_static_str();
+        let (mut handler_results, span) = {
+            // TODO: we unfortunately have to pin the future on the heap, since GetPage futures are huge and
+            // won't fit on the stack.
+            let mut boxpinned =
+                Box::pin(self.pagestream_dispatch_batched_message(batch, io_concurrency, ctx));
+            log_slow(
+                log_slow_name,
+                LOG_SLOW_GETPAGE_THRESHOLD,
+                boxpinned.as_mut(),
+            )
+            .await?
+        };

        // We purposefully don't count flush time into the smgr operation timer.
        //
--- a/pageserver/src/tenant.rs
+++ b/pageserver/src/tenant.rs
@@ -6559,7 +6559,11 @@ mod tests {

        tline.freeze_and_flush().await?;
        tline
-            .compact(&CancellationToken::new(), EnumSet::empty(), &ctx)
+            .compact(
+                &CancellationToken::new(),
+                CompactFlags::NoYield.into(),
+                &ctx,
+            )
            .await?;

        let mut writer = tline.writer().await;
@@ -6576,7 +6580,11 @@ mod tests {

        tline.freeze_and_flush().await?;
        tline
-            .compact(&CancellationToken::new(), EnumSet::empty(), &ctx)
+            .compact(
+                &CancellationToken::new(),
+                CompactFlags::NoYield.into(),
+                &ctx,
+            )
            .await?;

        let mut writer = tline.writer().await;
@@ -6593,7 +6601,11 @@ mod tests {

        tline.freeze_and_flush().await?;
        tline
-            .compact(&CancellationToken::new(), EnumSet::empty(), &ctx)
+            .compact(
+                &CancellationToken::new(),
+                CompactFlags::NoYield.into(),
+                &ctx,
+            )
            .await?;

        let mut writer = tline.writer().await;
@@ -6610,7 +6622,11 @@ mod tests {

        tline.freeze_and_flush().await?;
        tline
-            .compact(&CancellationToken::new(), EnumSet::empty(), &ctx)
+            .compact(
+                &CancellationToken::new(),
+                CompactFlags::NoYield.into(),
+                &ctx,
+            )
            .await?;

        assert_eq!(
@@ -6693,7 +6709,9 @@ mod tests {
            timeline.freeze_and_flush().await?;
            if compact {
                // this requires timeline to be &Arc<Timeline>
-                timeline.compact(&cancel, EnumSet::empty(), ctx).await?;
+                timeline
+                    .compact(&cancel, CompactFlags::NoYield.into(), ctx)
+                    .await?;
            }

            // this doesn't really need to use the timeline_id target, but it is closer to what it
@@ -7020,6 +7038,7 @@ mod tests {
        child_timeline.freeze_and_flush().await?;
        let mut flags = EnumSet::new();
        flags.insert(CompactFlags::ForceRepartition);
+        flags.insert(CompactFlags::NoYield);
        child_timeline
            .compact(&CancellationToken::new(), flags, &ctx)
            .await?;
@@ -7398,7 +7417,9 @@ mod tests {

            // Perform a cycle of flush, compact, and GC
            tline.freeze_and_flush().await?;
-            tline.compact(&cancel, EnumSet::empty(), &ctx).await?;
+            tline
+                .compact(&cancel, CompactFlags::NoYield.into(), &ctx)
+                .await?;
            tenant
                .gc_iteration(Some(tline.timeline_id), 0, Duration::ZERO, &cancel, &ctx)
                .await?;
@@ -7727,6 +7748,7 @@ mod tests {
                            let mut flags = EnumSet::new();
                            flags.insert(CompactFlags::ForceImageLayerCreation);
                            flags.insert(CompactFlags::ForceRepartition);
+                            flags.insert(CompactFlags::NoYield);
                            flags
                        } else {
                            EnumSet::empty()
@@ -7777,7 +7799,9 @@ mod tests {
        let before_num_l0_delta_files =
            tline.layers.read().await.layer_map()?.level0_deltas().len();

-        tline.compact(&cancel, EnumSet::empty(), &ctx).await?;
+        tline
+            .compact(&cancel, CompactFlags::NoYield.into(), &ctx)
+            .await?;

        let after_num_l0_delta_files = tline.layers.read().await.layer_map()?.level0_deltas().len();

@@ -7893,7 +7917,6 @@ mod tests {
            Ok((res, reconstruct_state.get_delta_layers_visited() as usize))
        }

-        #[allow(clippy::needless_range_loop)]
        for blknum in 0..NUM_KEYS {
            lsn = Lsn(lsn.0 + 0x10);
            test_key.field6 = (blknum * STEP) as u32;
@@ -7943,6 +7966,7 @@ mod tests {
                            let mut flags = EnumSet::new();
                            flags.insert(CompactFlags::ForceImageLayerCreation);
                            flags.insert(CompactFlags::ForceRepartition);
+                            flags.insert(CompactFlags::NoYield);
                            flags
                        },
                        &ctx,
@@ -8405,6 +8429,7 @@ mod tests {
                    let mut flags = EnumSet::new();
                    flags.insert(CompactFlags::ForceImageLayerCreation);
                    flags.insert(CompactFlags::ForceRepartition);
+                    flags.insert(CompactFlags::NoYield);
                    flags
                },
                &ctx,
@@ -8472,6 +8497,7 @@ mod tests {
                    let mut flags = EnumSet::new();
                    flags.insert(CompactFlags::ForceImageLayerCreation);
                    flags.insert(CompactFlags::ForceRepartition);
+                    flags.insert(CompactFlags::NoYield);
                    flags
                },
                &ctx,
--- a/pageserver/src/tenant/ephemeral_file.rs
+++ b/pageserver/src/tenant/ephemeral_file.rs
@@ -9,6 +9,7 @@ use camino::Utf8PathBuf;
 use num_traits::Num;
 use pageserver_api::shard::TenantShardId;
 use tokio_epoll_uring::{BoundedBuf, Slice};
+use tokio_util::sync::CancellationToken;
 use tracing::{error, info_span};
 use utils::id::TimelineId;

@@ -19,7 +20,7 @@ use crate::page_cache;
 use crate::tenant::storage_layer::inmemory_layer::vectored_dio_read::File;
 use crate::virtual_file::owned_buffers_io::io_buf_aligned::IoBufAlignedMut;
 use crate::virtual_file::owned_buffers_io::slice::SliceMutExt;
-use crate::virtual_file::owned_buffers_io::write::Buffer;
+use crate::virtual_file::owned_buffers_io::write::{Buffer, FlushTaskError};
 use crate::virtual_file::{self, IoBufferMut, VirtualFile, owned_buffers_io};

 pub struct EphemeralFile {
@@ -40,6 +41,7 @@ impl EphemeralFile {
        tenant_shard_id: TenantShardId,
        timeline_id: TimelineId,
        gate: &utils::sync::gate::Gate,
+        cancel: &CancellationToken,
        ctx: &RequestContext,
    ) -> anyhow::Result<EphemeralFile> {
        static NEXT_FILENAME: AtomicU64 = AtomicU64::new(1);
@@ -75,6 +77,7 @@ impl EphemeralFile {
                file,
                || IoBufferMut::with_capacity(TAIL_SZ),
                gate.enter()?,
+                cancel.child_token(),
                ctx,
                info_span!(parent: None, "ephemeral_file_buffered_writer", tenant_id=%tenant_shard_id.tenant_id, shard_id=%tenant_shard_id.shard_slug(), timeline_id=%timeline_id, path = %filename),
            ),
@@ -101,6 +104,14 @@ impl Drop for EphemeralFile {
    }
 }

+#[derive(Debug, thiserror::Error)]
+pub(crate) enum EphemeralFileWriteError {
+    #[error("{0}")]
+    TooLong(String),
+    #[error("cancelled")]
+    Cancelled,
+}
+
 impl EphemeralFile {
    pub(crate) fn len(&self) -> u64 {
        self.bytes_written
@@ -133,7 +144,7 @@ impl EphemeralFile {
        &mut self,
        srcbuf: &[u8],
        ctx: &RequestContext,
-    ) -> std::io::Result<u64> {
+    ) -> Result<u64, EphemeralFileWriteError> {
        let (pos, control) = self.write_raw_controlled(srcbuf, ctx).await?;
        if let Some(control) = control {
            control.release().await;
@@ -145,24 +156,24 @@ impl EphemeralFile {
        &mut self,
        srcbuf: &[u8],
        ctx: &RequestContext,
-    ) -> std::io::Result<(u64, Option<owned_buffers_io::write::FlushControl>)> {
+    ) -> Result<(u64, Option<owned_buffers_io::write::FlushControl>), EphemeralFileWriteError> {
        let pos = self.bytes_written;

        let new_bytes_written = pos.checked_add(srcbuf.len().into_u64()).ok_or_else(|| {
-            std::io::Error::new(
-                std::io::ErrorKind::Other,
-                format!(
-                    "write would grow EphemeralFile beyond u64::MAX: len={pos} writen={srcbuf_len}",
-                    srcbuf_len = srcbuf.len(),
-                ),
-            )
+            EphemeralFileWriteError::TooLong(format!(
+                "write would grow EphemeralFile beyond u64::MAX: len={pos} writen={srcbuf_len}",
+                srcbuf_len = srcbuf.len(),
+            ))
        })?;

        // Write the payload
        let (nwritten, control) = self
            .buffered_writer
            .write_buffered_borrowed_controlled(srcbuf, ctx)
-            .await?;
+            .await
+            .map_err(|e| match e {
+                FlushTaskError::Cancelled => EphemeralFileWriteError::Cancelled,
+            })?;
        assert_eq!(
            nwritten,
            srcbuf.len(),
@@ -184,8 +195,14 @@ impl super::storage_layer::inmemory_layer::vectored_dio_read::File for Ephemeral
    ) -> std::io::Result<(tokio_epoll_uring::Slice<B>, usize)> {
        let submitted_offset = self.buffered_writer.bytes_submitted();

-        let mutable = self.buffered_writer.inspect_mutable();
-        let mutable = &mutable[0..mutable.pending()];
+        let mutable = match self.buffered_writer.inspect_mutable() {
+            Some(mutable) => &mutable[0..mutable.pending()],
+            None => {
+                // Timeline::cancel and hence buffered writer flush was cancelled.
+                // Remain read-available while timeline is shutting down.
+                &[]
+            }
+        };

        let maybe_flushed = self.buffered_writer.inspect_maybe_flushed();

@@ -216,7 +233,6 @@ impl super::storage_layer::inmemory_layer::vectored_dio_read::File for Ephemeral
        let (written_range, maybe_flushed_range) = {
            if maybe_flushed.is_some() {
                // [       written       ][ maybe_flushed ][    mutable    ]
-                //                        <-   TAIL_SZ   -><-   TAIL_SZ   ->
                //                                         ^
                //                                 `submitted_offset`
                // <++++++ on disk +++++++????????????????>
@@ -232,7 +248,6 @@ impl super::storage_layer::inmemory_layer::vectored_dio_read::File for Ephemeral
                )
            } else {
                // [       written                        ][    mutable    ]
-                //                                         <-   TAIL_SZ   ->
                //                                         ^
                //                                 `submitted_offset`
                // <++++++ on disk +++++++++++++++++++++++>
@@ -366,8 +381,9 @@ mod tests {
            harness("ephemeral_file_holds_gate_open").unwrap();

        let gate = utils::sync::gate::Gate::default();
+        let cancel = CancellationToken::new();

-        let file = EphemeralFile::create(conf, tenant_id, timeline_id, &gate, &ctx)
+        let file = EphemeralFile::create(conf, tenant_id, timeline_id, &gate, &cancel, &ctx)
            .await
            .unwrap();

@@ -397,12 +413,13 @@ mod tests {
        let (conf, tenant_id, timeline_id, ctx) = harness("test_ephemeral_file_basics").unwrap();

        let gate = utils::sync::gate::Gate::default();
+        let cancel = CancellationToken::new();

-        let mut file = EphemeralFile::create(conf, tenant_id, timeline_id, &gate, &ctx)
+        let mut file = EphemeralFile::create(conf, tenant_id, timeline_id, &gate, &cancel, &ctx)
            .await
            .unwrap();

-        let mutable = file.buffered_writer.inspect_mutable();
+        let mutable = file.buffered_writer.mutable();
        let cap = mutable.capacity();
        let align = mutable.align();

@@ -445,7 +462,7 @@ mod tests {
        let maybe_flushed_buffer_contents = file.buffered_writer.inspect_maybe_flushed().unwrap();
        assert_eq!(&maybe_flushed_buffer_contents[..], &content[cap..cap * 2]);

-        let mutable_buffer_contents = file.buffered_writer.inspect_mutable();
+        let mutable_buffer_contents = file.buffered_writer.mutable();
        assert_eq!(mutable_buffer_contents, &content[cap * 2..write_nbytes]);
    }

@@ -454,13 +471,13 @@ mod tests {
        let (conf, tenant_id, timeline_id, ctx) = harness("test_flushes_do_happen").unwrap();

        let gate = utils::sync::gate::Gate::default();
-
-        let mut file = EphemeralFile::create(conf, tenant_id, timeline_id, &gate, &ctx)
+        let cancel = CancellationToken::new();
+        let mut file = EphemeralFile::create(conf, tenant_id, timeline_id, &gate, &cancel, &ctx)
            .await
            .unwrap();

        // mutable buffer and maybe_flushed buffer each has `cap` bytes.
-        let cap = file.buffered_writer.inspect_mutable().capacity();
+        let cap = file.buffered_writer.mutable().capacity();

        let content: Vec<u8> = rand::thread_rng()
            .sample_iter(rand::distributions::Standard)
@@ -470,10 +487,8 @@ mod tests {
        file.write_raw(&content, &ctx).await.unwrap();

        // assert the state is as this test expects it to be
-        assert_eq!(
-            &file.load_to_io_buf(&ctx).await.unwrap(),
-            &content[0..cap * 2 + cap / 2]
-        );
+        let load_io_buf_res = file.load_to_io_buf(&ctx).await.unwrap();
+        assert_eq!(&load_io_buf_res[..], &content[0..cap * 2 + cap / 2]);
        let md = file.buffered_writer.as_inner().path().metadata().unwrap();
        assert_eq!(
            md.len(),
@@ -485,7 +500,7 @@ mod tests {
            &content[cap..cap * 2]
        );
        assert_eq!(
-            &file.buffered_writer.inspect_mutable()[0..cap / 2],
+            &file.buffered_writer.mutable()[0..cap / 2],
            &content[cap * 2..cap * 2 + cap / 2]
        );
    }
@@ -501,12 +516,13 @@ mod tests {
            harness("test_read_split_across_file_and_buffer").unwrap();

        let gate = utils::sync::gate::Gate::default();
+        let cancel = CancellationToken::new();

-        let mut file = EphemeralFile::create(conf, tenant_id, timeline_id, &gate, &ctx)
+        let mut file = EphemeralFile::create(conf, tenant_id, timeline_id, &gate, &cancel, &ctx)
            .await
            .unwrap();

-        let mutable = file.buffered_writer.inspect_mutable();
+        let mutable = file.buffered_writer.mutable();
        let cap = mutable.capacity();
        let align = mutable.align();
        let content: Vec<u8> = rand::thread_rng()
--- a/pageserver/src/tenant/layer_map.rs
+++ b/pageserver/src/tenant/layer_map.rs
@@ -1112,6 +1112,7 @@ mod tests {
    };
    use pageserver_api::key::DBDIR_KEY;
    use pageserver_api::keyspace::{KeySpace, KeySpaceRandomAccum};
+    use tokio_util::sync::CancellationToken;
    use utils::id::{TenantId, TimelineId};
    use utils::shard::TenantShardId;

@@ -1193,6 +1194,7 @@ mod tests {
    async fn ranged_search() {
        let harness = TenantHarness::create("ranged_search").await.unwrap();
        let (tenant, ctx) = harness.load().await;
+        let cancel = CancellationToken::new();
        let timeline_id = TimelineId::generate();
        // Create the timeline such that the in-memory layers can be written
        // to the timeline directory.
@@ -1209,6 +1211,7 @@ mod tests {
                harness.tenant_shard_id,
                lsn_range.start,
                &gate,
+                &cancel,
                &ctx,
            )
            .await
--- a/pageserver/src/tenant/mgr.rs
+++ b/pageserver/src/tenant/mgr.rs
@@ -14,7 +14,7 @@ use futures::StreamExt;
 use itertools::Itertools;
 use once_cell::sync::Lazy;
 use pageserver_api::key::Key;
-use pageserver_api::models::LocationConfigMode;
+use pageserver_api::models::{DetachBehavior, LocationConfigMode};
 use pageserver_api::shard::{
    ShardCount, ShardIdentity, ShardIndex, ShardNumber, ShardStripeSize, TenantShardId,
 };
@@ -1914,7 +1914,7 @@ impl TenantManager {
        tenant_shard_id: TenantShardId,
        timeline_id: TimelineId,
        prepared: PreparedTimelineDetach,
-        behavior: detach_ancestor::DetachBehavior,
+        behavior: DetachBehavior,
        mut attempt: detach_ancestor::Attempt,
        ctx: &RequestContext,
    ) -> Result<HashSet<TimelineId>, detach_ancestor::Error> {
--- a/pageserver/src/tenant/remote_timeline_client/download.rs
+++ b/pageserver/src/tenant/remote_timeline_client/download.rs
@@ -205,6 +205,7 @@ async fn download_object(
        }
        #[cfg(target_os = "linux")]
        crate::virtual_file::io_engine::IoEngine::TokioEpollUring => {
+            use crate::virtual_file::owned_buffers_io::write::FlushTaskError;
            use std::sync::Arc;

            use crate::virtual_file::{IoBufferMut, owned_buffers_io};
@@ -228,6 +229,7 @@ async fn download_object(
                    destination_file,
                    || IoBufferMut::with_capacity(super::BUFFER_SIZE),
                    gate.enter().map_err(|_| DownloadError::Cancelled)?,
+                    cancel.child_token(),
                    ctx,
                    tracing::info_span!(parent: None, "download_object_buffered_writer", %dst_path),
                );
@@ -240,11 +242,21 @@ async fn download_object(
                    {
                        let chunk = match res {
                            Ok(chunk) => chunk,
-                            Err(e) => return Err(e),
+                            Err(e) => return Err(DownloadError::from(e)),
                        };
-                        buffered.write_buffered_borrowed(&chunk, ctx).await?;
+                        buffered
+                            .write_buffered_borrowed(&chunk, ctx)
+                            .await
+                            .map_err(|e| match e {
+                                FlushTaskError::Cancelled => DownloadError::Cancelled,
+                            })?;
                    }
-                    let inner = buffered.flush_and_into_inner(ctx).await?;
+                    let inner = buffered
+                        .flush_and_into_inner(ctx)
+                        .await
+                        .map_err(|e| match e {
+                            FlushTaskError::Cancelled => DownloadError::Cancelled,
+                        })?;
                    Ok(inner)
                }
                .await?;
--- a/pageserver/src/tenant/storage_layer/inmemory_layer.rs
+++ b/pageserver/src/tenant/storage_layer/inmemory_layer.rs
@@ -19,6 +19,7 @@ use pageserver_api::keyspace::KeySpace;
 use pageserver_api::models::InMemoryLayerInfo;
 use pageserver_api::shard::TenantShardId;
 use tokio::sync::RwLock;
+use tokio_util::sync::CancellationToken;
 use tracing::*;
 use utils::id::TimelineId;
 use utils::lsn::Lsn;
@@ -552,13 +553,15 @@ impl InMemoryLayer {
        tenant_shard_id: TenantShardId,
        start_lsn: Lsn,
        gate: &utils::sync::gate::Gate,
+        cancel: &CancellationToken,
        ctx: &RequestContext,
    ) -> Result<InMemoryLayer> {
        trace!(
            "initializing new empty InMemoryLayer for writing on timeline {timeline_id} at {start_lsn}"
        );

-        let file = EphemeralFile::create(conf, tenant_shard_id, timeline_id, gate, ctx).await?;
+        let file =
+            EphemeralFile::create(conf, tenant_shard_id, timeline_id, gate, cancel, ctx).await?;
        let key = InMemoryLayerFileId(file.page_cache_file_id());

        Ok(InMemoryLayer {
--- a/pageserver/src/tenant/tasks.rs
+++ b/pageserver/src/tenant/tasks.rs
@@ -268,7 +268,7 @@ async fn compaction_loop(tenant: Arc<Tenant>, cancel: CancellationToken) {
                error_run += 1;
                let backoff =
                    exponential_backoff_duration(error_run, BASE_BACKOFF_SECS, MAX_BACKOFF_SECS);
-                log_compaction_error(&err, error_run, backoff, cancel.is_cancelled());
+                log_compaction_error(&err, Some((error_run, backoff)), cancel.is_cancelled());
                continue;
            }
        }
@@ -281,10 +281,9 @@ async fn compaction_loop(tenant: Arc<Tenant>, cancel: CancellationToken) {
    }
 }

-fn log_compaction_error(
+pub(crate) fn log_compaction_error(
    err: &CompactionError,
-    error_count: u32,
-    sleep_duration: Duration,
+    retry_info: Option<(u32, Duration)>,
    task_cancelled: bool,
 ) {
    use CompactionError::*;
@@ -318,14 +317,26 @@ fn log_compaction_error(
        }
    };

-    match level {
-        Level::ERROR => {
-            error!("Compaction failed {error_count} times, retrying in {sleep_duration:?}: {err:#}")
+    if let Some((error_count, sleep_duration)) = retry_info {
+        match level {
+            Level::ERROR => {
+                error!(
+                    "Compaction failed {error_count} times, retrying in {sleep_duration:?}: {err:#}"
+                )
+            }
+            Level::INFO => {
+                info!(
+                    "Compaction failed {error_count} times, retrying in {sleep_duration:?}: {err:#}"
+                )
+            }
+            level => unimplemented!("unexpected level {level:?}"),
        }
-        Level::INFO => {
-            info!("Compaction failed {error_count} times, retrying in {sleep_duration:?}: {err:#}")
+    } else {
+        match level {
+            Level::ERROR => error!("Compaction failed: {err:#}"),
+            Level::INFO => info!("Compaction failed: {err:#}"),
+            level => unimplemented!("unexpected level {level:?}"),
        }
-        level => unimplemented!("unexpected level {level:?}"),
    }
 }

--- a/pageserver/src/tenant/timeline.rs
+++ b/pageserver/src/tenant/timeline.rs
@@ -45,8 +45,9 @@ use pageserver_api::key::{
 use pageserver_api::keyspace::{KeySpaceAccum, KeySpaceRandomAccum, SparseKeyPartitioning};
 use pageserver_api::models::{
    CompactKeyRange, CompactLsnRange, CompactionAlgorithm, CompactionAlgorithmSettings,
-    DownloadRemoteLayersTaskInfo, DownloadRemoteLayersTaskSpawnRequest, EvictionPolicy,
-    InMemoryLayerInfo, LayerMapInfo, LsnLease, PageTraceEvent, RelSizeMigration, TimelineState,
+    DetachBehavior, DownloadRemoteLayersTaskInfo, DownloadRemoteLayersTaskSpawnRequest,
+    EvictionPolicy, InMemoryLayerInfo, LayerMapInfo, LsnLease, PageTraceEvent, RelSizeMigration,
+    TimelineState,
 };
 use pageserver_api::reltag::{BlockNumber, RelTag};
 use pageserver_api::shard::{ShardIdentity, ShardIndex, ShardNumber, TenantShardId};
@@ -67,6 +68,7 @@ use tracing::*;
 use utils::generation::Generation;
 use utils::guard_arc_swap::GuardArcSwap;
 use utils::id::TimelineId;
+use utils::logging::{MonitorSlowFutureCallback, monitor_slow_future};
 use utils::lsn::{AtomicLsn, Lsn, RecordLsn};
 use utils::postgres_client::PostgresClientProtocol;
 use utils::rate_limit::RateLimit;
@@ -87,6 +89,7 @@ use super::remote_timeline_client::index::{GcCompactionState, IndexPart};
 use super::remote_timeline_client::{RemoteTimelineClient, WaitCompletionError};
 use super::secondary::heatmap::HeatMapLayer;
 use super::storage_layer::{LayerFringe, LayerVisibilityHint, ReadableLayer};
+use super::tasks::log_compaction_error;
 use super::upload_queue::NotInitialized;
 use super::{
    AttachedTenantConf, GcError, HeatMapTimeline, MaybeOffloaded,
@@ -439,6 +442,8 @@ pub struct Timeline {
    heatmap_layers_downloader: Mutex<Option<heatmap_layers_downloader::HeatmapLayersDownloader>>,

    pub(crate) rel_size_v2_status: ArcSwapOption<RelSizeMigration>,
+
+    wait_lsn_log_slow: tokio::sync::Semaphore,
 }

 pub(crate) enum PreviousHeatmap {
@@ -1479,17 +1484,67 @@ impl Timeline {
            WaitLsnTimeout::Default => self.conf.wait_lsn_timeout,
        };

-        let _timer = crate::metrics::WAIT_LSN_TIME.start_timer();
+        let timer = crate::metrics::WAIT_LSN_TIME.start_timer();
+        let start_finish_counterpair_guard = self.metrics.wait_lsn_start_finish_counterpair.guard();

-        match self.last_record_lsn.wait_for_timeout(lsn, timeout).await {
+        let wait_for_timeout = self.last_record_lsn.wait_for_timeout(lsn, timeout);
+        let wait_for_timeout = std::pin::pin!(wait_for_timeout);
+        // Use threshold of 1 because even 1 second of wait for ingest is very much abnormal.
+        let log_slow_threshold = Duration::from_secs(1);
+        // Use period of 10 to avoid flooding logs during an outage that affects all timelines.
+        let log_slow_period = Duration::from_secs(10);
+        let mut logging_permit = None;
+        let wait_for_timeout = monitor_slow_future(
+            log_slow_threshold,
+            log_slow_period,
+            wait_for_timeout,
+            |MonitorSlowFutureCallback {
+                 ready,
+                 is_slow,
+                 elapsed_total,
+                 elapsed_since_last_callback,
+             }| {
+                self.metrics
+                    .wait_lsn_in_progress_micros
+                    .inc_by(u64::try_from(elapsed_since_last_callback.as_micros()).unwrap());
+                if !is_slow {
+                    return;
+                }
+                // It's slow, see if we should log it.
+                // (We limit the logging to one per invocation per timeline to avoid excessive
+                // logging during an extended broker / networking outage that affects all timelines.)
+                if logging_permit.is_none() {
+                    logging_permit = self.wait_lsn_log_slow.try_acquire().ok();
+                }
+                if logging_permit.is_none() {
+                    return;
+                }
+                // We log it.
+                if ready {
+                    info!(
+                        "slow wait_lsn completed after {:.3}s",
+                        elapsed_total.as_secs_f64()
+                    );
+                } else {
+                    info!(
+                        "slow wait_lsn still running for {:.3}s",
+                        elapsed_total.as_secs_f64()
+                    );
+                }
+            },
+        );
+        let res = wait_for_timeout.await;
+        // don't count the time spent waiting for lock below, and also in walreceiver.status(), towards the wait_lsn_time_histo
+        drop(logging_permit);
+        drop(start_finish_counterpair_guard);
+        drop(timer);
+        match res {
            Ok(()) => Ok(()),
            Err(e) => {
                use utils::seqwait::SeqWaitError::*;
                match e {
                    Shutdown => Err(WaitLsnError::Shutdown),
                    Timeout => {
-                        // don't count the time spent waiting for lock below, and also in walreceiver.status(), towards the wait_lsn_time_histo
-                        drop(_timer);
                        let walreceiver_status = self.walreceiver_status();
                        Err(WaitLsnError::Timeout(format!(
                            "Timed out while waiting for WAL record at LSN {} to arrive, last_record_lsn {} disk consistent LSN={}, WalReceiver status: {}",
@@ -1802,18 +1857,23 @@ impl Timeline {
        flags: EnumSet<CompactFlags>,
        ctx: &RequestContext,
    ) -> Result<CompactionOutcome, CompactionError> {
-        self.compact_with_options(
-            cancel,
-            CompactOptions {
-                flags,
-                compact_key_range: None,
-                compact_lsn_range: None,
-                sub_compaction: false,
-                sub_compaction_max_job_size_mb: None,
-            },
-            ctx,
-        )
-        .await
+        let res = self
+            .compact_with_options(
+                cancel,
+                CompactOptions {
+                    flags,
+                    compact_key_range: None,
+                    compact_lsn_range: None,
+                    sub_compaction: false,
+                    sub_compaction_max_job_size_mb: None,
+                },
+                ctx,
+            )
+            .await;
+        if let Err(err) = &res {
+            log_compaction_error(err, None, cancel.is_cancelled());
+        }
+        res
    }

    /// Outermost timeline compaction operation; downloads needed layers.
@@ -2423,8 +2483,9 @@ impl Timeline {
    }

    fn get_l0_flush_delay_threshold(&self) -> Option<usize> {
-        // Disable L0 flushes by default. This and compaction needs further tuning.
-        const DEFAULT_L0_FLUSH_DELAY_FACTOR: usize = 0; // TODO: default to e.g. 3
+        // By default, delay L0 flushes at 3x the compaction threshold. The compaction threshold
+        // defaults to 10, and L0 compaction is generally able to keep L0 counts below 30.
+        const DEFAULT_L0_FLUSH_DELAY_FACTOR: usize = 3;

        // If compaction is disabled, don't delay.
        if self.get_compaction_period() == Duration::ZERO {
@@ -2452,8 +2513,9 @@ impl Timeline {
    }

    fn get_l0_flush_stall_threshold(&self) -> Option<usize> {
-        // Disable L0 stalls by default. In ingest benchmarks, we see image compaction take >10
-        // minutes, blocking L0 compaction, and we can't stall L0 flushes for that long.
+        // Disable L0 stalls by default. Stalling can cause unavailability if L0 compaction isn't
+        // responsive, and it can e.g. block on other compaction via the compaction semaphore or
+        // sibling timelines. We need more confidence before enabling this.
        const DEFAULT_L0_FLUSH_STALL_FACTOR: usize = 0; // TODO: default to e.g. 5

        // If compaction is disabled, don't stall.
@@ -2821,6 +2883,8 @@ impl Timeline {
                heatmap_layers_downloader: Mutex::new(None),

                rel_size_v2_status: ArcSwapOption::from_pointee(rel_size_v2_status),
+
+                wait_lsn_log_slow: tokio::sync::Semaphore::new(1),
            };

            result.repartition_threshold =
@@ -4122,6 +4186,7 @@ impl Timeline {
                self.timeline_id,
                self.tenant_shard_id,
                &self.gate,
+                &self.cancel,
                ctx,
            )
            .await?;
@@ -5388,7 +5453,7 @@ impl Timeline {
        self: &Arc<Timeline>,
        tenant: &crate::tenant::Tenant,
        options: detach_ancestor::Options,
-        behavior: detach_ancestor::DetachBehavior,
+        behavior: DetachBehavior,
        ctx: &RequestContext,
    ) -> Result<detach_ancestor::Progress, detach_ancestor::Error> {
        detach_ancestor::prepare(self, tenant, behavior, options, ctx).await
@@ -5409,7 +5474,7 @@ impl Timeline {
        prepared: detach_ancestor::PreparedTimelineDetach,
        ancestor_timeline_id: TimelineId,
        ancestor_lsn: Lsn,
-        behavior: detach_ancestor::DetachBehavior,
+        behavior: DetachBehavior,
        ctx: &RequestContext,
    ) -> Result<detach_ancestor::DetachingAndReparenting, detach_ancestor::Error> {
        detach_ancestor::detach_and_reparent(
@@ -6678,6 +6743,8 @@ impl Timeline {
            self.tenant_shard_id,
            in_memory.lsn_range.start,
            &self.gate,
+            // TODO: if we ever use this function in production code, we need to pass the real cancellation token
+            &CancellationToken::new(),
            ctx,
        )
        .await
--- a/pageserver/src/tenant/timeline/compaction.rs
+++ b/pageserver/src/tenant/timeline/compaction.rs
@@ -56,6 +56,7 @@ use crate::tenant::storage_layer::merge_iterator::MergeIterator;
 use crate::tenant::storage_layer::{
    AsLayerDesc, PersistentLayerDesc, PersistentLayerKey, ValueReconstructState,
 };
+use crate::tenant::tasks::log_compaction_error;
 use crate::tenant::timeline::{
    DeltaLayerWriter, ImageLayerCreationOutcome, ImageLayerWriter, IoConcurrency, Layer,
    ResidentLayer, drop_rlock,
@@ -440,6 +441,20 @@ impl GcCompactionQueue {
        ctx: &RequestContext,
        gc_block: &GcBlock,
        timeline: &Arc<Timeline>,
+    ) -> Result<CompactionOutcome, CompactionError> {
+        let res = self.iteration_inner(cancel, ctx, gc_block, timeline).await;
+        if let Err(err) = &res {
+            log_compaction_error(err, None, cancel.is_cancelled());
+        }
+        res
+    }
+
+    async fn iteration_inner(
+        &self,
+        cancel: &CancellationToken,
+        ctx: &RequestContext,
+        gc_block: &GcBlock,
+        timeline: &Arc<Timeline>,
    ) -> Result<CompactionOutcome, CompactionError> {
        let Ok(_one_op_at_a_time_guard) = self.consumer_lock.try_lock() else {
            return Err(CompactionError::AlreadyRunning(
@@ -3189,7 +3204,11 @@ impl Timeline {
        }

        // TODO: move the below part to the loop body
-        let last_key = last_key.expect("no keys produced during compaction");
+        let Some(last_key) = last_key else {
+            return Err(CompactionError::Other(anyhow!(
+                "no keys produced during compaction"
+            )));
+        };
        stat.on_unique_key_visited();

        let retention = self
--- a/pageserver/src/tenant/timeline/detach_ancestor.rs
+++ b/pageserver/src/tenant/timeline/detach_ancestor.rs
@@ -3,6 +3,7 @@ use std::sync::Arc;

 use anyhow::Context;
 use http_utils::error::ApiError;
+use pageserver_api::models::DetachBehavior;
 use pageserver_api::models::detach_ancestor::AncestorDetached;
 use pageserver_api::shard::ShardIdentity;
 use tokio::sync::Semaphore;
@@ -139,30 +140,6 @@ pub(crate) struct Options {
    pub(crate) copy_concurrency: std::num::NonZeroUsize,
 }

-/// Controls the detach ancestor behavior.
-/// - When set to `NoAncestorAndReparent`, we will only detach a branch if its ancestor is a root branch. It will automatically reparent any children of the ancestor before and at the branch point.
-/// - When set to `MultiLevelAndNoReparent`, we will detach a branch from multiple levels of ancestors, and no reparenting will happen at all.
-#[derive(Debug, Clone, Copy, Default)]
-pub enum DetachBehavior {
-    #[default]
-    NoAncestorAndReparent,
-    MultiLevelAndNoReparent,
-}
-
-impl std::str::FromStr for DetachBehavior {
-    type Err = &'static str;
-
-    fn from_str(s: &str) -> Result<Self, Self::Err> {
-        match s {
-            "no_ancestor_and_reparent" => Ok(DetachBehavior::NoAncestorAndReparent),
-            "multi_level_and_no_reparent" => Ok(DetachBehavior::MultiLevelAndNoReparent),
-            "v1" => Ok(DetachBehavior::NoAncestorAndReparent),
-            "v2" => Ok(DetachBehavior::MultiLevelAndNoReparent),
-            _ => Err("cannot parse detach behavior"),
-        }
-    }
-}
-
 impl Default for Options {
    fn default() -> Self {
        Self {
--- a/pageserver/src/tenant/timeline/layer_manager.rs
+++ b/pageserver/src/tenant/timeline/layer_manager.rs
@@ -4,6 +4,7 @@ use std::sync::Arc;
 use anyhow::{Context, bail, ensure};
 use itertools::Itertools;
 use pageserver_api::shard::TenantShardId;
+use tokio_util::sync::CancellationToken;
 use tracing::trace;
 use utils::id::TimelineId;
 use utils::lsn::{AtomicLsn, Lsn};
@@ -193,6 +194,7 @@ impl OpenLayerManager {

    /// Open a new writable layer to append data if there is no open layer, otherwise return the
    /// current open layer, called within `get_layer_for_write`.
+    #[allow(clippy::too_many_arguments)]
    pub(crate) async fn get_layer_for_write(
        &mut self,
        lsn: Lsn,
@@ -200,6 +202,7 @@ impl OpenLayerManager {
        timeline_id: TimelineId,
        tenant_shard_id: TenantShardId,
        gate: &utils::sync::gate::Gate,
+        cancel: &CancellationToken,
        ctx: &RequestContext,
    ) -> anyhow::Result<Arc<InMemoryLayer>> {
        ensure!(lsn.is_aligned());
@@ -227,9 +230,16 @@ impl OpenLayerManager {
                timeline_id, start_lsn, lsn
            );

-            let new_layer =
-                InMemoryLayer::create(conf, timeline_id, tenant_shard_id, start_lsn, gate, ctx)
-                    .await?;
+            let new_layer = InMemoryLayer::create(
+                conf,
+                timeline_id,
+                tenant_shard_id,
+                start_lsn,
+                gate,
+                cancel,
+                ctx,
+            )
+            .await?;
            let layer = Arc::new(new_layer);

            self.layer_map.open_layer = Some(layer.clone());
--- a/pageserver/src/virtual_file/owned_buffers_io/write.rs
+++ b/pageserver/src/virtual_file/owned_buffers_io/write.rs
@@ -3,7 +3,9 @@ use std::sync::Arc;

 pub(crate) use flush::FlushControl;
 use flush::FlushHandle;
+pub(crate) use flush::FlushTaskError;
 use tokio_epoll_uring::IoBuf;
+use tokio_util::sync::CancellationToken;

 use super::io_buf_aligned::IoBufAligned;
 use super::io_buf_ext::{FullSlice, IoBufExt};
@@ -40,11 +42,19 @@ pub trait OwnedAsyncWriter {
 // since we would avoid copying majority of the data into the internal buffer.
 pub struct BufferedWriter<B: Buffer, W> {
    writer: Arc<W>,
-    /// invariant: always remains Some(buf) except
-    /// - while IO is ongoing => goes back to Some() once the IO completed successfully
-    /// - after an IO error => stays `None` forever
-    ///
-    /// In these exceptional cases, it's `None`.
+    /// Clone of the buffer that was last submitted to the flush loop.
+    /// `None` if no flush request has been submitted, Some forever after.
+    pub(super) maybe_flushed: Option<FullSlice<B::IoBuf>>,
+    /// New writes are accumulated here.
+    /// `None` only during submission while we wait for flush loop to accept
+    /// the full dirty buffer in exchange for a clean buffer.
+    /// If that exchange fails with an [`FlushTaskError`], the write path
+    /// bails and leaves this as `None`.
+    /// Subsequent writes will panic if attempted.
+    /// The read path continues to work without error because [`Self::maybe_flushed`]
+    /// and [`Self::bytes_submitted`] are advanced before the flush loop exchange starts,
+    /// so, they will never try to read from [`Self::mutable`] anyway, because it's past
+    /// the [`Self::maybe_flushed`] point.
    mutable: Option<B>,
    /// A handle to the background flush task for writting data to disk.
    flush_handle: FlushHandle<B::IoBuf, W>,
@@ -65,16 +75,19 @@ where
        writer: Arc<W>,
        buf_new: impl Fn() -> B,
        gate_guard: utils::sync::gate::GateGuard,
+        cancel: CancellationToken,
        ctx: &RequestContext,
        flush_task_span: tracing::Span,
    ) -> Self {
        Self {
            writer: writer.clone(),
            mutable: Some(buf_new()),
+            maybe_flushed: None,
            flush_handle: FlushHandle::spawn_new(
                writer,
                buf_new(),
                gate_guard,
+                cancel,
                ctx.attached_child(),
                flush_task_span,
            ),
@@ -92,25 +105,26 @@ where
    }

    /// Panics if used after any of the write paths returned an error
-    pub fn inspect_mutable(&self) -> &B {
-        self.mutable()
+    pub fn inspect_mutable(&self) -> Option<&B> {
+        self.mutable.as_ref()
    }

    /// Gets a reference to the maybe flushed read-only buffer.
    /// Returns `None` if the writer has not submitted any flush request.
    pub fn inspect_maybe_flushed(&self) -> Option<&FullSlice<Buf>> {
-        self.flush_handle.maybe_flushed.as_ref()
+        self.maybe_flushed.as_ref()
    }

    #[cfg_attr(target_os = "macos", allow(dead_code))]
    pub async fn flush_and_into_inner(
        mut self,
        ctx: &RequestContext,
-    ) -> std::io::Result<(u64, Arc<W>)> {
+    ) -> Result<(u64, Arc<W>), FlushTaskError> {
        self.flush(ctx).await?;

        let Self {
            mutable: buf,
+            maybe_flushed: _,
            writer,
            mut flush_handle,
            bytes_submitted: bytes_amount,
@@ -120,12 +134,9 @@ where
        Ok((bytes_amount, writer))
    }

-    /// Gets a reference to the mutable in-memory buffer.
-    #[inline(always)]
-    fn mutable(&self) -> &B {
-        self.mutable
-            .as_ref()
-            .expect("must not use after we returned an error")
+    #[cfg(test)]
+    pub(crate) fn mutable(&self) -> &B {
+        self.mutable.as_ref().expect("must not use after an error")
    }

    #[cfg_attr(target_os = "macos", allow(dead_code))]
@@ -133,7 +144,7 @@ where
        &mut self,
        chunk: &[u8],
        ctx: &RequestContext,
-    ) -> std::io::Result<usize> {
+    ) -> Result<usize, FlushTaskError> {
        let (len, control) = self.write_buffered_borrowed_controlled(chunk, ctx).await?;
        if let Some(control) = control {
            control.release().await;
@@ -146,7 +157,7 @@ where
        &mut self,
        mut chunk: &[u8],
        ctx: &RequestContext,
-    ) -> std::io::Result<(usize, Option<FlushControl>)> {
+    ) -> Result<(usize, Option<FlushControl>), FlushTaskError> {
        let chunk_len = chunk.len();
        let mut control: Option<FlushControl> = None;
        while !chunk.is_empty() {
@@ -167,17 +178,47 @@ where
        Ok((chunk_len, control))
    }

+    /// This function can only error if the flush task got cancelled.
+    /// In that case, we leave [`Self::mutable`] intentionally as `None`.
+    ///
+    /// The read path continues to function correctly; it can read up to the
+    /// point where it could read before, i.e., including what was in [`Self::mutable`]
+    /// before the call to this function, because that's now stored in [`Self::maybe_flushed`].
+    ///
+    /// The write path becomes unavailable and will panic if used.
+    /// The only correct solution to retry writes is to discard the entire [`BufferedWriter`],
+    /// which upper layers of pageserver write path currently do not support.
+    /// It is in fact quite hard to reason about what exactly happens in today's code.
+    /// Best case we accumulate junk in the EphemeralFile, worst case is data corruption.
    #[must_use = "caller must explcitly check the flush control"]
-    async fn flush(&mut self, _ctx: &RequestContext) -> std::io::Result<Option<FlushControl>> {
+    async fn flush(
+        &mut self,
+        _ctx: &RequestContext,
+    ) -> Result<Option<FlushControl>, FlushTaskError> {
        let buf = self.mutable.take().expect("must not use after an error");
        let buf_len = buf.pending();
        if buf_len == 0 {
            self.mutable = Some(buf);
            return Ok(None);
        }
-        let (recycled, flush_control) = self.flush_handle.flush(buf, self.bytes_submitted).await?;
+        // Prepare the buffer for read while flushing.
+        let slice = buf.flush();
+        // NB: this assignment also drops thereference to the old buffer, allowing us to re-own & make it mutable below.
+        self.maybe_flushed = Some(slice.cheap_clone());
+        let offset = self.bytes_submitted;
        self.bytes_submitted += u64::try_from(buf_len).unwrap();
+
+        // If we return/panic here or later, we'll leave mutable = None, breaking further
+        // writers, but the read path should still work.
+        let (recycled, flush_control) = self.flush_handle.flush(slice, offset).await?;
+
+        // The only other place that could hold a reference to the recycled buffer
+        // is in `Self::maybe_flushed`, but we have already replace it with the new buffer.
+        let recycled = Buffer::reuse_after_flush(recycled.into_raw_slice().into_inner());
+
+        // We got back some recycled buffer, can open up for more writes again.
        self.mutable = Some(recycled);
+
        Ok(Some(flush_control))
    }
 }
@@ -290,10 +331,12 @@ mod tests {
        let ctx = &ctx;
        let recorder = Arc::new(RecorderWriter::default());
        let gate = utils::sync::gate::Gate::default();
+        let cancel = CancellationToken::new();
        let mut writer = BufferedWriter::<_, RecorderWriter>::new(
            recorder,
            || IoBufferMut::with_capacity(2),
            gate.enter()?,
+            cancel,
            ctx,
            tracing::Span::none(),
        );
--- a/pageserver/src/virtual_file/owned_buffers_io/write/flush.rs
+++ b/pageserver/src/virtual_file/owned_buffers_io/write/flush.rs
@@ -1,7 +1,6 @@
 use std::ops::ControlFlow;
 use std::sync::Arc;

-use once_cell::sync::Lazy;
 use tokio_util::sync::CancellationToken;
 use tracing::{Instrument, info, info_span, warn};
 use utils::sync::duplex;
@@ -15,9 +14,6 @@ use crate::virtual_file::owned_buffers_io::io_buf_ext::FullSlice;
 /// A handle to the flush task.
 pub struct FlushHandle<Buf, W> {
    inner: Option<FlushHandleInner<Buf, W>>,
-    /// Immutable buffer for serving tail reads.
-    /// `None` if no flush request has been submitted.
-    pub(super) maybe_flushed: Option<FullSlice<Buf>>,
 }

 pub struct FlushHandleInner<Buf, W> {
@@ -25,7 +21,7 @@ pub struct FlushHandleInner<Buf, W> {
    /// and receives recyled buffer.
    channel: duplex::mpsc::Duplex<FlushRequest<Buf>, FullSlice<Buf>>,
    /// Join handle for the background flush task.
-    join_handle: tokio::task::JoinHandle<std::io::Result<Arc<W>>>,
+    join_handle: tokio::task::JoinHandle<Result<Arc<W>, FlushTaskError>>,
 }

 struct FlushRequest<Buf> {
@@ -117,27 +113,31 @@ where
 {
    /// Spawns a new background flush task and obtains a handle.
    ///
-    /// Note: The background task so we do not need to explicitly maintain a queue of buffers.
+    /// Handle and background task are connected through a duplex channel.
+    /// Dirty buffers are sent to the background task for flushing.
+    /// Clean buffers are sent back to the handle for reuse.
+    ///
+    /// The queue depth is 1, and the passed-in `buf` seeds the queue depth.
+    /// I.e., the passed-in buf is immediately available to the handle as a recycled buffer.
    pub fn spawn_new<B>(
        file: Arc<W>,
        buf: B,
        gate_guard: utils::sync::gate::GateGuard,
+        cancel: CancellationToken,
        ctx: RequestContext,
        span: tracing::Span,
    ) -> Self
    where
        B: Buffer<IoBuf = Buf> + Send + 'static,
    {
-        // It is fine to buffer up to only 1 message. We only 1 message in-flight at a time.
        let (front, back) = duplex::mpsc::channel(1);
+        back.try_send(buf.flush())
+            .expect("we just created it with capacity 1");

        let join_handle = tokio::spawn(
-            async move {
-                FlushBackgroundTask::new(back, file, gate_guard, ctx)
-                    .run(buf.flush())
-                    .await
-            }
-            .instrument(span),
+            FlushBackgroundTask::new(back, file, gate_guard, cancel, ctx)
+                .run()
+                .instrument(span),
        );

        FlushHandle {
@@ -145,7 +145,6 @@ where
                channel: front,
                join_handle,
            }),
-            maybe_flushed: None,
        }
    }

@@ -153,15 +152,11 @@ where
    /// Returns a buffer that completed flushing for re-use, length reset to 0, capacity unchanged.
    /// If `save_buf_for_read` is true, then we save the buffer in `Self::maybe_flushed`, otherwise
    /// clear `maybe_flushed`.
-    pub async fn flush<B>(&mut self, buf: B, offset: u64) -> std::io::Result<(B, FlushControl)>
-    where
-        B: Buffer<IoBuf = Buf> + Send + 'static,
-    {
-        let slice = buf.flush();
-
-        // Saves a buffer for read while flushing. This also removes reference to the old buffer.
-        self.maybe_flushed = Some(slice.cheap_clone());
-
+    pub async fn flush(
+        &mut self,
+        slice: FullSlice<Buf>,
+        offset: u64,
+    ) -> Result<(FullSlice<Buf>, FlushControl), FlushTaskError> {
        let (request, flush_control) = new_flush_op(slice, offset);

        // Submits the buffer to the background task.
@@ -177,13 +172,10 @@ where
            return self.handle_error().await;
        };

-        // The only other place that could hold a reference to the recycled buffer
-        // is in `Self::maybe_flushed`, but we have already replace it with the new buffer.
-        let recycled = Buffer::reuse_after_flush(recycled.into_raw_slice().into_inner());
        Ok((recycled, flush_control))
    }

-    async fn handle_error<T>(&mut self) -> std::io::Result<T> {
+    async fn handle_error<T>(&mut self) -> Result<T, FlushTaskError> {
        Err(self
            .shutdown()
            .await
@@ -191,7 +183,7 @@ where
    }

    /// Cleans up the channel, join the flush task.
-    pub async fn shutdown(&mut self) -> std::io::Result<Arc<W>> {
+    pub async fn shutdown(&mut self) -> Result<Arc<W>, FlushTaskError> {
        let handle = self
            .inner
            .take()
@@ -217,10 +209,17 @@ pub struct FlushBackgroundTask<Buf, W> {
    /// A writter for persisting data to disk.
    writer: Arc<W>,
    ctx: RequestContext,
+    cancel: CancellationToken,
    /// Prevent timeline from shuting down until the flush background task finishes flushing all remaining buffers to disk.
    _gate_guard: utils::sync::gate::GateGuard,
 }

+#[derive(Debug, thiserror::Error)]
+pub enum FlushTaskError {
+    #[error("flush task cancelled")]
+    Cancelled,
+}
+
 impl<Buf, W> FlushBackgroundTask<Buf, W>
 where
    Buf: IoBufAligned + Send + Sync,
@@ -231,25 +230,20 @@ where
        channel: duplex::mpsc::Duplex<FullSlice<Buf>, FlushRequest<Buf>>,
        file: Arc<W>,
        gate_guard: utils::sync::gate::GateGuard,
+        cancel: CancellationToken,
        ctx: RequestContext,
    ) -> Self {
        FlushBackgroundTask {
            channel,
            writer: file,
            _gate_guard: gate_guard,
+            cancel,
            ctx,
        }
    }

    /// Runs the background flush task.
-    /// The passed in slice is immediately sent back to the flush handle through the duplex channel.
-    async fn run(mut self, slice: FullSlice<Buf>) -> std::io::Result<Arc<W>> {
-        // Sends the extra buffer back to the handle.
-        // TODO: can this ever await and or fail? I think not.
-        self.channel.send(slice).await.map_err(|_| {
-            std::io::Error::new(std::io::ErrorKind::BrokenPipe, "flush handle closed early")
-        })?;
-
+    async fn run(mut self) -> Result<Arc<W>, FlushTaskError> {
        //  Exit condition: channel is closed and there is no remaining buffer to be flushed
        while let Some(request) = self.channel.recv().await {
            #[cfg(test)]
@@ -267,15 +261,13 @@ where
            // (The upper layers of the Pageserver write path are not equipped to retry write errors
            //  becasuse they often deallocate the buffers that were already written).
            //
-            // TODO: cancellation sensitiity.
-            // Without it, if we hit a bug where retrying is never successful,
-            // then we can't shut down the timeline/tenant/pageserver cleanly because
-            // layers of the Pageserver write path are holding the gate open for EphemeralFile.
-            //
            // TODO: use utils::backoff::retry once async closures are actually usable
            //
            let mut slice_storage = Some(request.slice);
            for attempt in 1.. {
+                if self.cancel.is_cancelled() {
+                    return Err(FlushTaskError::Cancelled);
+                }
                let result = async {
                    if attempt > 1 {
                        info!("retrying flush");
@@ -283,6 +275,11 @@ where
                    let slice = slice_storage.take().expect(
                        "likely previous invocation of this future didn't get polled to completion",
                    );
+                    // Don't cancel this write by doing tokio::select with self.cancel.cancelled().
+                    // The underlying tokio-epoll-uring slot / kernel operation is still ongoing and occupies resources.
+                    // If we retry indefinitely, we'll deplete those resources.
+                    // Future: teach tokio-epoll-uring io_uring operation cancellation, but still,
+                    // wait for cancelled ops to complete and discard their error.
                    let (slice, res) = self.writer.write_all_at(slice, request.offset, &self.ctx).await;
                    slice_storage = Some(slice);
                    let res = res.maybe_fatal_err("owned_buffers_io flush");
@@ -290,8 +287,7 @@ where
                        return ControlFlow::Break(());
                    };
                    warn!(%err, "error flushing buffered writer buffer to disk, retrying after backoff");
-                    static NO_CANCELLATION: Lazy<CancellationToken> = Lazy::new(CancellationToken::new);
-                    utils::backoff::exponential_backoff(attempt, 1.0, 10.0, &NO_CANCELLATION).await;
+                    utils::backoff::exponential_backoff(attempt, 1.0, 10.0, &self.cancel).await;
                    ControlFlow::Continue(())
                }
                .instrument(info_span!("flush_attempt", %attempt))
--- a/pgxn/neon/pagestore_smgr.c
+++ b/pgxn/neon/pagestore_smgr.c
@@ -2778,6 +2778,9 @@ neon_extend(SMgrRelation reln, ForkNumber forkNum, BlockNumber blkno,
 		case RELPERSISTENCE_TEMP:
 		case RELPERSISTENCE_UNLOGGED:
 			mdextend(reln, forkNum, blkno, buffer, skipFsync);
+			/* Update LFC in case of unlogged index build */
+			if (reln == unlogged_build_rel && unlogged_build_phase == UNLOGGED_BUILD_PHASE_2)
+				lfc_write(InfoFromSMgrRel(reln), forkNum, blkno, buffer);
 			return;

 		default:
@@ -2866,6 +2869,14 @@ neon_zeroextend(SMgrRelation reln, ForkNumber forkNum, BlockNumber blocknum,
 		case RELPERSISTENCE_TEMP:
 		case RELPERSISTENCE_UNLOGGED:
 			mdzeroextend(reln, forkNum, blocknum, nblocks, skipFsync);
+			/* Update LFC in case of unlogged index build */
+			if (reln == unlogged_build_rel && unlogged_build_phase == UNLOGGED_BUILD_PHASE_2)
+			{
+				for (int i = 0; i < nblocks; i++)
+				{
+					lfc_write(InfoFromSMgrRel(reln), forkNum, blocknum + i, buffer.data);
+				}
+			}
 			return;

 		default:
@@ -2898,6 +2909,11 @@ neon_zeroextend(SMgrRelation reln, ForkNumber forkNum, BlockNumber blocknum,
 						relpath(reln->smgr_rlocator, forkNum),
 						InvalidBlockNumber)));

+#ifdef DEBUG_COMPARE_LOCAL
+	if (IS_LOCAL_REL(reln))
+		mdzeroextend(reln, forkNum, blocknum, nblocks, skipFsync);
+#endif
+
 	/* Don't log any pages if we're not allowed to do so. */
 	if (!XLogInsertAllowed())
 		return;
@@ -3709,6 +3725,9 @@ neon_write(SMgrRelation reln, ForkNumber forknum, BlockNumber blocknum, const vo
 			#else
 			mdwrite(reln, forknum, blocknum, buffer, skipFsync);
 			#endif
+			/* Update LFC in case of unlogged index build */
+			if (reln == unlogged_build_rel && unlogged_build_phase == UNLOGGED_BUILD_PHASE_2)
+				lfc_write(InfoFromSMgrRel(reln), forknum, blocknum, buffer);
 			return;
 		default:
 			neon_log(ERROR, "unknown relpersistence '%c'", reln->smgr_relpersistence);
@@ -3772,6 +3791,9 @@ neon_writev(SMgrRelation reln, ForkNumber forknum, BlockNumber blkno,
 		case RELPERSISTENCE_TEMP:
 		case RELPERSISTENCE_UNLOGGED:
 			mdwritev(reln, forknum, blkno, buffers, nblocks, skipFsync);
+			/* Update LFC in case of unlogged index build */
+			if (reln == unlogged_build_rel && unlogged_build_phase == UNLOGGED_BUILD_PHASE_2)
+				lfc_writev(InfoFromSMgrRel(reln), forknum, blkno, buffers, nblocks);
 			return;
 		default:
 			neon_log(ERROR, "unknown relpersistence '%c'", reln->smgr_relpersistence);
@@ -4171,8 +4193,10 @@ neon_start_unlogged_build(SMgrRelation reln)
 	 * FIXME: should we pass isRedo true to create the tablespace dir if it
 	 * doesn't exist? Is it needed?
 	 */
-	if (!IsParallelWorker())
+#ifndef DEBUG_COMPARE_LOCAL
+ 	if (!IsParallelWorker())
 		mdcreate(reln, MAIN_FORKNUM, false);
+#endif
 }

 /*
@@ -4247,8 +4271,10 @@ neon_end_unlogged_build(SMgrRelation reln)

 			forget_cached_relsize(InfoFromNInfoB(rinfob), forknum);
 			mdclose(reln, forknum);
+#ifndef DEBUG_COMPARE_LOCAL
 			/* use isRedo == true, so that we drop it immediately */
 			mdunlink(rinfob, forknum, true);
+#endif
 		}
 	}

--- a/poetry.lock
+++ b/poetry.lock
@@ -1491,14 +1491,38 @@ files = [

 [[package]]
 name = "jsonnet"
-version = "0.20.0"
-description = "Python bindings for Jsonnet - The data templating language"
+version = "0.21.0rc2"
+description = "Python bindings for Jsonnet - The data templating language "
 optional = false
 python-versions = "*"
 groups = ["main"]
-markers = "python_version < \"3.13\""
 files = [
-    {file = "jsonnet-0.20.0.tar.gz", hash = "sha256:7e770c7bf3a366b97b650a39430450f77612e74406731eb75c5bd59f3f104d4f"},
+    {file = "jsonnet-0.21.0rc2-cp310-cp310-macosx_10_9_x86_64.whl", hash = "sha256:8779ac6820fee44ef736df2baedc3ae93e8cd5d672ee105015c2a47fe627a727"},
+    {file = "jsonnet-0.21.0rc2-cp310-cp310-macosx_11_0_arm64.whl", hash = "sha256:99affe8c71e2551465064a8039bb3d1cba27a0b73b2b9ff1b652e06f17d4ea8b"},
+    {file = "jsonnet-0.21.0rc2-cp310-cp310-manylinux_2_17_x86_64.manylinux2014_x86_64.whl", hash = "sha256:4a9dffb9aa01013d100ddfb7230d1eeb80f2a8eef712b1825a60cad57106d8bd"},
+    {file = "jsonnet-0.21.0rc2-cp310-cp310-musllinux_1_2_x86_64.whl", hash = "sha256:cca6c95f2879dcab52650b7aa09a4e82a139b084931b1f6f8c840f834fecc08a"},
+    {file = "jsonnet-0.21.0rc2-cp310-cp310-win_amd64.whl", hash = "sha256:016d6afdb302a6d00bf3bce6a0c3d9c093b992e33f9bc67c64a868035892258e"},
+    {file = "jsonnet-0.21.0rc2-cp311-cp311-macosx_10_9_x86_64.whl", hash = "sha256:e893ab2c9bf10d8ec9e9b0cee8961879c88d0619cc6d8f75ea284a78e06ae32b"},
+    {file = "jsonnet-0.21.0rc2-cp311-cp311-macosx_11_0_arm64.whl", hash = "sha256:c06b353cd3daa2781e6cd308e05f2f116396376994bcb5f59aaadbc6a752c7f2"},
+    {file = "jsonnet-0.21.0rc2-cp311-cp311-manylinux_2_17_x86_64.manylinux2014_x86_64.whl", hash = "sha256:9eb2bc8e62b73101329072da322f7e2a1bdb3ac530b94669128d1b480e311e55"},
+    {file = "jsonnet-0.21.0rc2-cp311-cp311-musllinux_1_2_x86_64.whl", hash = "sha256:113766fd0c25620807bcf04d4c739f461c971a4f0e4aece9ba62b4e762de9598"},
+    {file = "jsonnet-0.21.0rc2-cp311-cp311-win_amd64.whl", hash = "sha256:8dab208c2c2760be60f87d1ceb8b28c86b51ed0e31129a7d90cd5fe890b41225"},
+    {file = "jsonnet-0.21.0rc2-cp312-cp312-macosx_10_13_x86_64.whl", hash = "sha256:95f5b9dd26a41d6f258d1baa8d22e557051beeed8c52a6202584f1becca9dcb5"},
+    {file = "jsonnet-0.21.0rc2-cp312-cp312-macosx_11_0_arm64.whl", hash = "sha256:cecc6d76e2b377260fae0a060097c113e6ac361b8f739903ea7f3f5f64cdebdf"},
+    {file = "jsonnet-0.21.0rc2-cp312-cp312-manylinux_2_17_x86_64.manylinux2014_x86_64.whl", hash = "sha256:aaa2d18224af7e63872ef4a101e93962505456cf5f5439c3cfc25dad6845f8b1"},
+    {file = "jsonnet-0.21.0rc2-cp312-cp312-musllinux_1_2_x86_64.whl", hash = "sha256:2a9063f811554487ed552445e964aeec969cafb266b965029c8d6b091ce47950"},
+    {file = "jsonnet-0.21.0rc2-cp312-cp312-win_amd64.whl", hash = "sha256:80d171182c169761f744ba50068a4ad35d48e52b91d25bf4c7bb9a72f0a04f71"},
+    {file = "jsonnet-0.21.0rc2-cp313-cp313-macosx_10_13_x86_64.whl", hash = "sha256:f3657938f87cb6bc6da20ca631d437b5faf469ca060a7c7def9c8fd2f25a5e06"},
+    {file = "jsonnet-0.21.0rc2-cp313-cp313-macosx_11_0_arm64.whl", hash = "sha256:3dcebc30cb991b58bc416ee05e9387004d04716d5c0b89714ff042bd069af5c8"},
+    {file = "jsonnet-0.21.0rc2-cp313-cp313-manylinux_2_17_x86_64.manylinux2014_x86_64.whl", hash = "sha256:3ac52c95482df3ed93c908468ca2f40d4825b6baba284b395ddc47bd663b8c3a"},
+    {file = "jsonnet-0.21.0rc2-cp313-cp313-musllinux_1_2_x86_64.whl", hash = "sha256:8b34450823a7a1861de892fef9f29de1b4c19e1a79e27d81ffe7e57646cc89d6"},
+    {file = "jsonnet-0.21.0rc2-cp313-cp313-win_amd64.whl", hash = "sha256:573fd2580e46f4875ec505f1732f9e804b7063cba790342ed6fdafe9a6b30556"},
+    {file = "jsonnet-0.21.0rc2-cp39-cp39-macosx_10_9_x86_64.whl", hash = "sha256:871ca1411de3626499bda60b330d37f85a592918f99ba4809089bbb8d4f5bfe4"},
+    {file = "jsonnet-0.21.0rc2-cp39-cp39-macosx_11_0_arm64.whl", hash = "sha256:5d33b25a9c5bf9099100b9b16cb385a2876d891fbe639ee9d476fc75c861903a"},
+    {file = "jsonnet-0.21.0rc2-cp39-cp39-manylinux_2_17_x86_64.manylinux2014_x86_64.whl", hash = "sha256:b2bac374565c7f89a4675f19fd2b624ed1376519267f4e444f49b6fc0368f6e5"},
+    {file = "jsonnet-0.21.0rc2-cp39-cp39-musllinux_1_2_x86_64.whl", hash = "sha256:fab7bbd88f9159f88a7350701a97bda24de9e3b9eef14c2501ba8b9224160d60"},
+    {file = "jsonnet-0.21.0rc2-cp39-cp39-win_amd64.whl", hash = "sha256:ed71ffba0fd233a1bca7b0f7be79730792c5383e562a9dc7da152478d9ee1612"},
+    {file = "jsonnet-0.21.0rc2.tar.gz", hash = "sha256:2b83ec4b5a771c3732e0972be23a71f042ad2940db6918d3a52aade69bc394fb"},
 ]

 [[package]]
@@ -3820,4 +3844,4 @@ cffi = ["cffi (>=1.11)"]
 [metadata]
 lock-version = "2.1"
 python-versions = "^3.11"
-content-hash = "010ffce959bb256880ab5a267048c182e4612b3151f9a94e3bf5d3a7807962fe"
+content-hash = "715fc8c896dcfa1b15054deeddcdec557ef93af91b26e1c8e4688fe4dbef5296"
--- a/proxy/Cargo.toml
+++ b/proxy/Cargo.toml
@@ -70,8 +70,9 @@ reqwest-middleware = { workspace = true, features = ["json"] }
 reqwest-retry.workspace = true
 reqwest-tracing.workspace = true
 rustc-hash.workspace = true
-rustls-pemfile.workspace = true
 rustls.workspace = true
+rustls-native-certs.workspace = true
+rustls-pemfile.workspace = true
 scopeguard.workspace = true
 serde.workspace = true
 serde_json.workspace = true
@@ -99,8 +100,7 @@ url.workspace = true
 urlencoding.workspace = true
 utils.workspace = true
 uuid.workspace = true
-rustls-native-certs.workspace = true
-x509-parser.workspace = true
+x509-cert.workspace = true
 redis.workspace = true
 zerocopy.workspace = true

--- a/proxy/src/binary/local_proxy.rs
+++ b/proxy/src/binary/local_proxy.rs
@@ -5,6 +5,7 @@ use std::sync::Arc;
 use std::time::Duration;

 use anyhow::{Context, bail, ensure};
+use arc_swap::ArcSwapOption;
 use camino::{Utf8Path, Utf8PathBuf};
 use clap::Parser;
 use compute_api::spec::LocalProxySpec;
@@ -27,6 +28,7 @@ use crate::config::{
 };
 use crate::control_plane::locks::ApiLocks;
 use crate::control_plane::messages::{EndpointJwksResponse, JwksSettings};
+use crate::ext::TaskExt;
 use crate::http::health_server::AppMetrics;
 use crate::intern::RoleNameInt;
 use crate::metrics::{Metrics, ThreadPoolMetrics};
@@ -190,7 +192,11 @@ pub async fn run() -> anyhow::Result<()> {
    // 2. The config file is written but the signal hook is not yet received
    // 3. local_proxy completes startup but has no config loaded, despite there being a registerd config.
    refresh_config_notify.notify_one();
-    tokio::spawn(refresh_config_loop(args.config_path, refresh_config_notify));
+    tokio::spawn(refresh_config_loop(
+        config,
+        args.config_path,
+        refresh_config_notify,
+    ));

    maintenance_tasks.spawn(crate::http::health_server::task_main(
        metrics_listener,
@@ -269,7 +275,7 @@ fn build_config(args: &LocalProxyCliArgs) -> anyhow::Result<&'static ProxyConfig
    };

    Ok(Box::leak(Box::new(ProxyConfig {
-        tls_config: None,
+        tls_config: ArcSwapOption::from(None),
        metric_collection: None,
        http_config,
        authentication_config: AuthenticationConfig {
@@ -311,14 +317,16 @@ enum RefreshConfigError {
    Parse(#[from] serde_json::Error),
    #[error(transparent)]
    Validate(anyhow::Error),
+    #[error(transparent)]
+    Tls(anyhow::Error),
 }

-async fn refresh_config_loop(path: Utf8PathBuf, rx: Arc<Notify>) {
+async fn refresh_config_loop(config: &ProxyConfig, path: Utf8PathBuf, rx: Arc<Notify>) {
    let mut init = true;
    loop {
        rx.notified().await;

-        match refresh_config_inner(&path).await {
+        match refresh_config_inner(config, &path).await {
            Ok(()) => {}
            // don't log for file not found errors if this is the first time we are checking
            // for computes that don't use local_proxy, this is not an error.
@@ -327,6 +335,9 @@ async fn refresh_config_loop(path: Utf8PathBuf, rx: Arc<Notify>) {
            {
                debug!(error=?e, ?path, "could not read config file");
            }
+            Err(RefreshConfigError::Tls(e)) => {
+                error!(error=?e, ?path, "could not read TLS certificates");
+            }
            Err(e) => {
                error!(error=?e, ?path, "could not read config file");
            }
@@ -336,7 +347,10 @@ async fn refresh_config_loop(path: Utf8PathBuf, rx: Arc<Notify>) {
    }
 }

-async fn refresh_config_inner(path: &Utf8Path) -> Result<(), RefreshConfigError> {
+async fn refresh_config_inner(
+    config: &ProxyConfig,
+    path: &Utf8Path,
+) -> Result<(), RefreshConfigError> {
    let bytes = tokio::fs::read(&path).await?;
    let data: LocalProxySpec = serde_json::from_slice(&bytes)?;

@@ -406,5 +420,20 @@ async fn refresh_config_inner(path: &Utf8Path) -> Result<(), RefreshConfigError>
    info!("successfully loaded new config");
    JWKS_ROLE_MAP.store(Some(Arc::new(EndpointJwksResponse { jwks: jwks_set })));

+    if let Some(tls_config) = data.tls {
+        let tls_config = tokio::task::spawn_blocking(move || {
+            crate::tls::server_config::configure_tls(
+                &tls_config.key_path,
+                &tls_config.cert_path,
+                None,
+                false,
+            )
+        })
+        .await
+        .propagate_task_panic()
+        .map_err(RefreshConfigError::Tls)?;
+        config.tls_config.store(Some(Arc::new(tls_config)));
+    }
+
    Ok(())
 }
--- a/proxy/src/binary/proxy.rs
+++ b/proxy/src/binary/proxy.rs
@@ -4,6 +4,7 @@ use std::sync::Arc;
 use std::time::Duration;

 use anyhow::bail;
+use arc_swap::ArcSwapOption;
 use futures::future::Either;
 use remote_storage::RemoteStorageConfig;
 use tokio::net::TcpListener;
@@ -563,6 +564,7 @@ fn build_config(args: &ProxyCliArgs) -> anyhow::Result<&'static ProxyConfig> {
        (None, None) => None,
        _ => bail!("either both or neither tls-key and tls-cert must be specified"),
    };
+    let tls_config = ArcSwapOption::from(tls_config.map(Arc::new));

    let backup_metric_collection_config = config::MetricBackupCollectionConfig {
        remote_storage_config: args.metric_backup_collection_remote_storage.clone(),
--- a/proxy/src/config.rs
+++ b/proxy/src/config.rs
@@ -3,6 +3,7 @@ use std::sync::Arc;
 use std::time::Duration;

 use anyhow::{Context, Ok, bail, ensure};
+use arc_swap::ArcSwapOption;
 use clap::ValueEnum;
 use remote_storage::RemoteStorageConfig;

@@ -17,7 +18,7 @@ pub use crate::tls::server_config::{TlsConfig, configure_tls};
 use crate::types::Host;

 pub struct ProxyConfig {
-    pub tls_config: Option<TlsConfig>,
+    pub tls_config: ArcSwapOption<TlsConfig>,
    pub metric_collection: Option<MetricCollectionConfig>,
    pub http_config: HttpConfig,
    pub authentication_config: AuthenticationConfig,
--- a/proxy/src/console_redirect_proxy.rs
+++ b/proxy/src/console_redirect_proxy.rs
@@ -177,7 +177,8 @@ pub(crate) async fn handle_client<S: AsyncRead + AsyncWrite + Unpin>(
    let proto = ctx.protocol();
    let request_gauge = metrics.connection_requests.guard(proto);

-    let tls = config.tls_config.as_ref();
+    let tls = config.tls_config.load();
+    let tls = tls.as_deref();

    let record_handshake_error = !ctx.has_private_peer_addr();
    let pause = ctx.latency_timer_pause(crate::metrics::Waiting::Client);
--- a/Show More
+++ b/Show More