compute: add non-skippable apply config operations.

apply_config() step of compute start is controlled by skip_pg_catalog_updates flag, this is a performance optimization to decrease compute startup time, but it introduces extra dependency on cplane. Introduce small subset of operations that we run always, independent from this flag.
Revert "feat(compute_ctl): allow to change audit_log_level for existing (#11308 )"
2026-01-23 21:30:36 +00:00 · 2025-03-21 15:27:35 +00:00 · 2025-03-21 15:21:22 +00:00
442 changed files with 8557 additions and 14954 deletions
--- a/.dockerignore
+++ b/.dockerignore
@@ -19,7 +19,6 @@
 !pageserver/
 !pgxn/
 !proxy/
-!object_storage/
 !storage_scrubber/
 !safekeeper/
 !storage_broker/
--- a/.github/actionlint.yml
+++ b/.github/actionlint.yml
@@ -6,10 +6,8 @@ self-hosted-runner:
    - small
    - small-metal
    - small-arm64
-    - unit-perf
    - us-east-2
 config-variables:
-  - AWS_ECR_REGION
  - AZURE_DEV_CLIENT_ID
  - AZURE_DEV_REGISTRY_NAME
  - AZURE_DEV_SUBSCRIPTION_ID
@@ -17,25 +15,23 @@ config-variables:
  - AZURE_PROD_REGISTRY_NAME
  - AZURE_PROD_SUBSCRIPTION_ID
  - AZURE_TENANT_ID
-  - BENCHMARK_INGEST_TARGET_PROJECTID
-  - BENCHMARK_LARGE_OLTP_PROJECTID
  - BENCHMARK_PROJECT_ID_PUB
  - BENCHMARK_PROJECT_ID_SUB
-  - DEV_AWS_OIDC_ROLE_ARN
-  - DEV_AWS_OIDC_ROLE_MANAGE_BENCHMARK_EC2_VMS_ARN
-  - HETZNER_CACHE_BUCKET
-  - HETZNER_CACHE_ENDPOINT
-  - HETZNER_CACHE_REGION
-  - NEON_DEV_AWS_ACCOUNT_ID
-  - NEON_PROD_AWS_ACCOUNT_ID
-  - PGREGRESS_PG16_PROJECT_ID
-  - PGREGRESS_PG17_PROJECT_ID
  - REMOTE_STORAGE_AZURE_CONTAINER
  - REMOTE_STORAGE_AZURE_REGION
-  - SLACK_CICD_CHANNEL_ID
-  - SLACK_ON_CALL_DEVPROD_STREAM
-  - SLACK_ON_CALL_QA_STAGING_STREAM
-  - SLACK_ON_CALL_STORAGE_STAGING_STREAM
-  - SLACK_RUST_CHANNEL_ID
-  - SLACK_STORAGE_CHANNEL_ID
  - SLACK_UPCOMING_RELEASE_CHANNEL_ID
+  - DEV_AWS_OIDC_ROLE_ARN
+  - BENCHMARK_INGEST_TARGET_PROJECTID
+  - PGREGRESS_PG16_PROJECT_ID
+  - PGREGRESS_PG17_PROJECT_ID
+  - SLACK_ON_CALL_QA_STAGING_STREAM
+  - DEV_AWS_OIDC_ROLE_MANAGE_BENCHMARK_EC2_VMS_ARN
+  - SLACK_ON_CALL_STORAGE_STAGING_STREAM
+  - SLACK_CICD_CHANNEL_ID
+  - SLACK_STORAGE_CHANNEL_ID
+  - NEON_DEV_AWS_ACCOUNT_ID
+  - NEON_PROD_AWS_ACCOUNT_ID
+  - AWS_ECR_REGION
+  - BENCHMARK_LARGE_OLTP_PROJECTID
+  - SLACK_ON_CALL_DEVPROD_STREAM
+  - SLACK_RUST_CHANNEL_ID
--- a/.github/actions/allure-report-generate/action.yml
+++ b/.github/actions/allure-report-generate/action.yml
@@ -70,7 +70,6 @@ runs:

    - name: Install Allure
      shell: bash -euxo pipefail {0}
-      working-directory: /tmp
      run: |
        if ! which allure; then
          ALLURE_ZIP=allure-${ALLURE_VERSION}.zip
--- a/.github/scripts/generate_image_maps.py
+++ b/.github/scripts/generate_image_maps.py
@@ -39,18 +39,12 @@ registries = {
    ],
 }

-release_branches = ["release", "release-proxy", "release-compute"]
-
 outputs: dict[str, dict[str, list[str]]] = {}

-target_tags = (
-    [target_tag, "latest"]
-    if branch == "main"
-    else [target_tag, "released"]
-    if branch in release_branches
-    else [target_tag]
+target_tags = [target_tag, "latest"] if branch == "main" else [target_tag]
+target_stages = (
+    ["dev", "prod"] if branch in ["release", "release-proxy", "release-compute"] else ["dev"]
 )
-target_stages = ["dev", "prod"] if branch in release_branches else ["dev"]

 for component_name, component_images in components.items():
    for stage in target_stages:
--- a/.github/scripts/push_with_image_map.py
+++ b/.github/scripts/push_with_image_map.py
@@ -2,9 +2,6 @@ import json
 import os
 import subprocess

-RED = "\033[91m"
-RESET = "\033[0m"
-
 image_map = os.getenv("IMAGE_MAP")
 if not image_map:
    raise ValueError("IMAGE_MAP environment variable is not set")
@@ -14,32 +11,12 @@ try:
 except json.JSONDecodeError as e:
    raise ValueError("Failed to parse IMAGE_MAP as JSON") from e

-failures = []
+for source, targets in parsed_image_map.items():
+    for target in targets:
+        cmd = ["docker", "buildx", "imagetools", "create", "-t", target, source]
+        print(f"Running: {' '.join(cmd)}")
+        result = subprocess.run(cmd, text=True, stdout=subprocess.PIPE, stderr=subprocess.STDOUT)

-pending = [(source, target) for source, targets in parsed_image_map.items() for target in targets]
-
-while len(pending) > 0:
-    if len(failures) > 10:
-        print("Error: more than 10 failures!")
-        for failure in failures:
-            print(f'"{failure[0]}" failed with the following output:')
-            print(failure[1])
-        raise RuntimeError("Retry limit reached.")
-
-    source, target = pending.pop(0)
-    cmd = ["docker", "buildx", "imagetools", "create", "-t", target, source]
-    print(f"Running: {' '.join(cmd)}")
-    result = subprocess.run(cmd, text=True, stdout=subprocess.PIPE, stderr=subprocess.STDOUT)
-
-    if result.returncode != 0:
-        failures.append((" ".join(cmd), result.stdout, target))
-        pending.append((source, target))
-        print(
-            f"{RED}[RETRY]{RESET} Push failed for {target}. Retrying... (failure count: {len(failures)})"
-        )
-        print(result.stdout)
-
-if len(failures) > 0 and (github_output := os.getenv("GITHUB_OUTPUT")):
-    failed_targets = [target for _, _, target in failures]
-    with open(github_output, "a") as f:
-        f.write(f"push_failures={json.dumps(failed_targets)}\n")
+        if result.returncode != 0:
+            print(f"Error: {result.stdout}")
+            raise RuntimeError(f"Command failed: {' '.join(cmd)}")
--- a/.github/workflows/_build-and-test-locally.yml
+++ b/.github/workflows/_build-and-test-locally.yml
@@ -128,49 +128,29 @@ jobs:

      - name: Cache postgres v14 build
        id: cache_pg_14
-        uses: tespkg/actions-cache@b7bf5fcc2f98a52ac6080eb0fd282c2f752074b1  # v1.8.0
+        uses: actions/cache@d4323d4df104b026a6aa633fdb11d772146be0bf # v4.2.2
        with:
-          endpoint: ${{ vars.HETZNER_CACHE_REGION }}.${{ vars.HETZNER_CACHE_ENDPOINT }}
-          bucket: ${{ vars.HETZNER_CACHE_BUCKET }}
-          accessKey: ${{ secrets.HETZNER_CACHE_ACCESS_KEY }}
-          secretKey: ${{ secrets.HETZNER_CACHE_SECRET_KEY }}
-          use-fallback: false
          path: pg_install/v14
          key: v1-${{ runner.os }}-${{ runner.arch }}-${{ inputs.build-type }}-pg-${{ steps.pg_v14_rev.outputs.pg_rev }}-bookworm-${{ hashFiles('Makefile', 'build-tools.Dockerfile') }}

      - name: Cache postgres v15 build
        id: cache_pg_15
-        uses: tespkg/actions-cache@b7bf5fcc2f98a52ac6080eb0fd282c2f752074b1  # v1.8.0
+        uses: actions/cache@d4323d4df104b026a6aa633fdb11d772146be0bf # v4.2.2
        with:
-          endpoint: ${{ vars.HETZNER_CACHE_REGION }}.${{ vars.HETZNER_CACHE_ENDPOINT }}
-          bucket: ${{ vars.HETZNER_CACHE_BUCKET }}
-          accessKey: ${{ secrets.HETZNER_CACHE_ACCESS_KEY }}
-          secretKey: ${{ secrets.HETZNER_CACHE_SECRET_KEY }}
-          use-fallback: false
          path: pg_install/v15
          key: v1-${{ runner.os }}-${{ runner.arch }}-${{ inputs.build-type }}-pg-${{ steps.pg_v15_rev.outputs.pg_rev }}-bookworm-${{ hashFiles('Makefile', 'build-tools.Dockerfile') }}

      - name: Cache postgres v16 build
        id: cache_pg_16
-        uses: tespkg/actions-cache@b7bf5fcc2f98a52ac6080eb0fd282c2f752074b1  # v1.8.0
+        uses: actions/cache@d4323d4df104b026a6aa633fdb11d772146be0bf # v4.2.2
        with:
-          endpoint: ${{ vars.HETZNER_CACHE_REGION }}.${{ vars.HETZNER_CACHE_ENDPOINT }}
-          bucket: ${{ vars.HETZNER_CACHE_BUCKET }}
-          accessKey: ${{ secrets.HETZNER_CACHE_ACCESS_KEY }}
-          secretKey: ${{ secrets.HETZNER_CACHE_SECRET_KEY }}
-          use-fallback: false
          path: pg_install/v16
          key: v1-${{ runner.os }}-${{ runner.arch }}-${{ inputs.build-type }}-pg-${{ steps.pg_v16_rev.outputs.pg_rev }}-bookworm-${{ hashFiles('Makefile', 'build-tools.Dockerfile') }}

      - name: Cache postgres v17 build
        id: cache_pg_17
-        uses: tespkg/actions-cache@b7bf5fcc2f98a52ac6080eb0fd282c2f752074b1  # v1.8.0
+        uses: actions/cache@d4323d4df104b026a6aa633fdb11d772146be0bf # v4.2.2
        with:
-          endpoint: ${{ vars.HETZNER_CACHE_REGION }}.${{ vars.HETZNER_CACHE_ENDPOINT }}
-          bucket: ${{ vars.HETZNER_CACHE_BUCKET }}
-          accessKey: ${{ secrets.HETZNER_CACHE_ACCESS_KEY }}
-          secretKey: ${{ secrets.HETZNER_CACHE_SECRET_KEY }}
-          use-fallback: false
          path: pg_install/v17
          key: v1-${{ runner.os }}-${{ runner.arch }}-${{ inputs.build-type }}-pg-${{ steps.pg_v17_rev.outputs.pg_rev }}-bookworm-${{ hashFiles('Makefile', 'build-tools.Dockerfile') }}

--- a/.github/workflows/_check-codestyle-python.yml
+++ b/.github/workflows/_check-codestyle-python.yml
@@ -37,14 +37,8 @@ jobs:

      - uses: actions/checkout@11bd71901bbe5b1630ceea73d27597364c9af683 # v4.2.2

-      - name: Cache poetry deps
-        uses: tespkg/actions-cache@b7bf5fcc2f98a52ac6080eb0fd282c2f752074b1  # v1.8.0
+      - uses: actions/cache@d4323d4df104b026a6aa633fdb11d772146be0bf # v4.2.2
        with:
-          endpoint: ${{ vars.HETZNER_CACHE_REGION }}.${{ vars.HETZNER_CACHE_ENDPOINT }}
-          bucket: ${{ vars.HETZNER_CACHE_BUCKET }}
-          accessKey: ${{ secrets.HETZNER_CACHE_ACCESS_KEY }}
-          secretKey: ${{ secrets.HETZNER_CACHE_SECRET_KEY }}
-          use-fallback: false
          path: ~/.cache/pypoetry/virtualenvs
          key: v2-${{ runner.os }}-${{ runner.arch }}-python-deps-bookworm-${{ hashFiles('poetry.lock') }}

--- a/.github/workflows/_check-codestyle-rust.yml
+++ b/.github/workflows/_check-codestyle-rust.yml
@@ -48,13 +48,8 @@ jobs:
          submodules: true

      - name: Cache cargo deps
-        uses: tespkg/actions-cache@b7bf5fcc2f98a52ac6080eb0fd282c2f752074b1  # v1.8.0
+        uses: actions/cache@d4323d4df104b026a6aa633fdb11d772146be0bf # v4.2.2
        with:
-          endpoint: ${{ vars.HETZNER_CACHE_REGION }}.${{ vars.HETZNER_CACHE_ENDPOINT }}
-          bucket: ${{ vars.HETZNER_CACHE_BUCKET }}
-          accessKey: ${{ secrets.HETZNER_CACHE_ACCESS_KEY }}
-          secretKey: ${{ secrets.HETZNER_CACHE_SECRET_KEY }}
-          use-fallback: false
          path: |
            ~/.cargo/registry
            !~/.cargo/registry/src
--- a/.github/workflows/_meta.yml
+++ b/.github/workflows/_meta.yml
@@ -5,9 +5,6 @@ on:
      github-event-name:
        type: string
        required: true
-      github-event-json:
-        type: string
-        required: true
    outputs:
      build-tag:
        description: "Tag for the current workflow run"
@@ -30,9 +27,6 @@ on:
      release-pr-run-id:
        description: "Only available if `run-kind in [storage-release, proxy-release, compute-release]`. Contains the run ID of the `Build and Test` workflow, assuming one with the current commit can be found."
        value: ${{ jobs.tags.outputs.release-pr-run-id }}
-      sha:
-        description: "github.event.pull_request.head.sha on release PRs, github.sha otherwise"
-        value: ${{ jobs.tags.outputs.sha }}

 permissions: {}

@@ -51,7 +45,6 @@ jobs:
      storage: ${{ steps.previous-releases.outputs.storage }}
      run-kind: ${{ steps.run-kind.outputs.run-kind }}
      release-pr-run-id: ${{ steps.release-pr-run-id.outputs.release-pr-run-id }}
-      sha: ${{ steps.sha.outputs.sha }}
    permissions:
      contents: read
    steps:
@@ -61,6 +54,10 @@ jobs:
        with:
          egress-policy: audit

+      - uses: actions/checkout@11bd71901bbe5b1630ceea73d27597364c9af683 # v4.2.2
+        with:
+          fetch-depth: 0
+
      - name: Get run kind
        id: run-kind
        env:
@@ -81,23 +78,6 @@ jobs:
        run: |
          echo "run-kind=$RUN_KIND" | tee -a $GITHUB_OUTPUT

-      - name: Get the right SHA
-        id: sha
-        env:
-          SHA: >
-            ${{
-              contains(fromJSON('["storage-rc-pr", "proxy-rc-pr", "compute-rc-pr"]'), steps.run-kind.outputs.run-kind)
-              && fromJSON(inputs.github-event-json).pull_request.head.sha
-              || github.sha
-            }}
-        run: |
-          echo "sha=$SHA" | tee -a $GITHUB_OUTPUT
-
-      - uses: actions/checkout@11bd71901bbe5b1630ceea73d27597364c9af683 # v4.2.2
-        with:
-          fetch-depth: 0
-          ref: ${{ steps.sha.outputs.sha }}
-
      - name: Get build tag
        id: build-tag
        env:
@@ -163,7 +143,7 @@ jobs:
        if: ${{ contains(fromJSON('["storage-release", "compute-release", "proxy-release"]'), steps.run-kind.outputs.run-kind) }}
        env:
          GH_TOKEN: ${{ secrets.GITHUB_TOKEN }}
-          CURRENT_SHA: ${{ github.sha }}
+          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 // ("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
@@ -104,25 +104,6 @@ jobs:
          password: ${{ secrets.NEON_DOCKERHUB_PASSWORD }}

      - name: Copy docker images to target registries
-        id: push
        run: python3 .github/scripts/push_with_image_map.py
        env:
          IMAGE_MAP: ${{ inputs.image-map }}
-
-      - name: Notify Slack if container image pushing fails
-        if: steps.push.outputs.push_failures || failure()
-        uses: slackapi/slack-github-action@485a9d42d3a73031f12ec201c457e2162c45d02d # v2.0.0
-        with:
-          method: chat.postMessage
-          token: ${{ secrets.SLACK_BOT_TOKEN }}
-          payload: |
-            channel: ${{ vars.SLACK_ON_CALL_DEVPROD_STREAM }}
-            text: >
-              *Container image pushing ${{
-                steps.push.outcome == 'failure' && 'failed completely' || 'succeeded with some retries'
-              }}* in
-              <${{ github.server_url }}/${{ github.repository }}/actions/runs/${{ github.run_id }}|GitHub Run>
-
-              ${{ steps.push.outputs.push_failures && format(
-                '*Failed targets:*\n• {0}', join(fromJson(steps.push.outputs.push_failures), '\n• ')
-              ) || '' }}
--- a/.github/workflows/build-macos.yml
+++ b/.github/workflows/build-macos.yml
@@ -63,13 +63,8 @@ jobs:

      - name: Cache postgres ${{ matrix.postgres-version }} build
        id: cache_pg
-        uses: tespkg/actions-cache@b7bf5fcc2f98a52ac6080eb0fd282c2f752074b1  # v1.8.0
+        uses: actions/cache@d4323d4df104b026a6aa633fdb11d772146be0bf # v4.2.2
        with:
-          endpoint: ${{ vars.HETZNER_CACHE_REGION }}.${{ vars.HETZNER_CACHE_ENDPOINT }}
-          bucket: ${{ vars.HETZNER_CACHE_BUCKET }}
-          accessKey: ${{ secrets.HETZNER_CACHE_ACCESS_KEY }}
-          secretKey: ${{ secrets.HETZNER_CACHE_SECRET_KEY }}
-          use-fallback: false
          path: pg_install/${{ matrix.postgres-version }}
          key: v1-${{ runner.os }}-${{ runner.arch }}-${{ env.BUILD_TYPE }}-pg-${{ matrix.postgres-version }}-${{ steps.pg_rev.outputs.pg_rev }}-${{ hashFiles('Makefile') }}

@@ -134,25 +129,15 @@ jobs:

      - name: Cache postgres v17 build
        id: cache_pg
-        uses: tespkg/actions-cache@b7bf5fcc2f98a52ac6080eb0fd282c2f752074b1  # v1.8.0
+        uses: actions/cache@d4323d4df104b026a6aa633fdb11d772146be0bf # v4.2.2
        with:
-          endpoint: ${{ vars.HETZNER_CACHE_REGION }}.${{ vars.HETZNER_CACHE_ENDPOINT }}
-          bucket: ${{ vars.HETZNER_CACHE_BUCKET }}
-          accessKey: ${{ secrets.HETZNER_CACHE_ACCESS_KEY }}
-          secretKey: ${{ secrets.HETZNER_CACHE_SECRET_KEY }}
-          use-fallback: false
          path: pg_install/v17
          key: v1-${{ runner.os }}-${{ runner.arch }}-${{ env.BUILD_TYPE }}-pg-v17-${{ steps.pg_rev.outputs.pg_rev }}-${{ hashFiles('Makefile') }}

      - name: Cache walproposer-lib
        id: cache_walproposer_lib
-        uses: tespkg/actions-cache@b7bf5fcc2f98a52ac6080eb0fd282c2f752074b1  # v1.8.0
+        uses: actions/cache@d4323d4df104b026a6aa633fdb11d772146be0bf # v4.2.2
        with:
-          endpoint: ${{ vars.HETZNER_CACHE_REGION }}.${{ vars.HETZNER_CACHE_ENDPOINT }}
-          bucket: ${{ vars.HETZNER_CACHE_BUCKET }}
-          accessKey: ${{ secrets.HETZNER_CACHE_ACCESS_KEY }}
-          secretKey: ${{ secrets.HETZNER_CACHE_SECRET_KEY }}
-          use-fallback: false
          path: pg_install/build/walproposer-lib
          key: v1-${{ runner.os }}-${{ runner.arch }}-${{ env.BUILD_TYPE }}-walproposer_lib-v17-${{ steps.pg_rev.outputs.pg_rev }}-${{ hashFiles('Makefile') }}

@@ -218,57 +203,32 @@ jobs:

      - name: Cache postgres v14 build
        id: cache_pg
-        uses: tespkg/actions-cache@b7bf5fcc2f98a52ac6080eb0fd282c2f752074b1  # v1.8.0
+        uses: actions/cache@d4323d4df104b026a6aa633fdb11d772146be0bf # v4.2.2
        with:
-          endpoint: ${{ vars.HETZNER_CACHE_REGION }}.${{ vars.HETZNER_CACHE_ENDPOINT }}
-          bucket: ${{ vars.HETZNER_CACHE_BUCKET }}
-          accessKey: ${{ secrets.HETZNER_CACHE_ACCESS_KEY }}
-          secretKey: ${{ secrets.HETZNER_CACHE_SECRET_KEY }}
-          use-fallback: false
          path: pg_install/v14
          key: v1-${{ runner.os }}-${{ runner.arch }}-${{ env.BUILD_TYPE }}-pg-v14-${{ steps.pg_rev_v14.outputs.pg_rev }}-${{ hashFiles('Makefile') }}
      - name: Cache postgres v15 build
        id: cache_pg_v15
-        uses: tespkg/actions-cache@b7bf5fcc2f98a52ac6080eb0fd282c2f752074b1  # v1.8.0
+        uses: actions/cache@d4323d4df104b026a6aa633fdb11d772146be0bf # v4.2.2
        with:
-          endpoint: ${{ vars.HETZNER_CACHE_REGION }}.${{ vars.HETZNER_CACHE_ENDPOINT }}
-          bucket: ${{ vars.HETZNER_CACHE_BUCKET }}
-          accessKey: ${{ secrets.HETZNER_CACHE_ACCESS_KEY }}
-          secretKey: ${{ secrets.HETZNER_CACHE_SECRET_KEY }}
-          use-fallback: false
          path: pg_install/v15
          key: v1-${{ runner.os }}-${{ runner.arch }}-${{ env.BUILD_TYPE }}-pg-v15-${{ steps.pg_rev_v15.outputs.pg_rev }}-${{ hashFiles('Makefile') }}
      - name: Cache postgres v16 build
        id: cache_pg_v16
-        uses: tespkg/actions-cache@b7bf5fcc2f98a52ac6080eb0fd282c2f752074b1  # v1.8.0
+        uses: actions/cache@d4323d4df104b026a6aa633fdb11d772146be0bf # v4.2.2
        with:
-          endpoint: ${{ vars.HETZNER_CACHE_REGION }}.${{ vars.HETZNER_CACHE_ENDPOINT }}
-          bucket: ${{ vars.HETZNER_CACHE_BUCKET }}
-          accessKey: ${{ secrets.HETZNER_CACHE_ACCESS_KEY }}
-          secretKey: ${{ secrets.HETZNER_CACHE_SECRET_KEY }}
-          use-fallback: false
          path: pg_install/v16
          key: v1-${{ runner.os }}-${{ runner.arch }}-${{ env.BUILD_TYPE }}-pg-v16-${{ steps.pg_rev_v16.outputs.pg_rev }}-${{ hashFiles('Makefile') }}
      - name: Cache postgres v17 build
        id: cache_pg_v17
-        uses: tespkg/actions-cache@b7bf5fcc2f98a52ac6080eb0fd282c2f752074b1  # v1.8.0
+        uses: actions/cache@d4323d4df104b026a6aa633fdb11d772146be0bf # v4.2.2
        with:
-          endpoint: ${{ vars.HETZNER_CACHE_REGION }}.${{ vars.HETZNER_CACHE_ENDPOINT }}
-          bucket: ${{ vars.HETZNER_CACHE_BUCKET }}
-          accessKey: ${{ secrets.HETZNER_CACHE_ACCESS_KEY }}
-          secretKey: ${{ secrets.HETZNER_CACHE_SECRET_KEY }}
-          use-fallback: false
          path: pg_install/v17
          key: v1-${{ runner.os }}-${{ runner.arch }}-${{ env.BUILD_TYPE }}-pg-v17-${{ steps.pg_rev_v17.outputs.pg_rev }}-${{ hashFiles('Makefile') }}

      - name: Cache cargo deps (only for v17)
-        uses: tespkg/actions-cache@b7bf5fcc2f98a52ac6080eb0fd282c2f752074b1  # v1.8.0
+        uses: actions/cache@d4323d4df104b026a6aa633fdb11d772146be0bf # v4.2.2
        with:
-          endpoint: ${{ vars.HETZNER_CACHE_REGION }}.${{ vars.HETZNER_CACHE_ENDPOINT }}
-          bucket: ${{ vars.HETZNER_CACHE_BUCKET }}
-          accessKey: ${{ secrets.HETZNER_CACHE_ACCESS_KEY }}
-          secretKey: ${{ secrets.HETZNER_CACHE_SECRET_KEY }}
-          use-fallback: false
          path: |
            ~/.cargo/registry
            !~/.cargo/registry/src
@@ -278,13 +238,8 @@ jobs:

      - name: Cache walproposer-lib
        id: cache_walproposer_lib
-        uses: tespkg/actions-cache@b7bf5fcc2f98a52ac6080eb0fd282c2f752074b1  # v1.8.0
+        uses: actions/cache@d4323d4df104b026a6aa633fdb11d772146be0bf # v4.2.2
        with:
-          endpoint: ${{ vars.HETZNER_CACHE_REGION }}.${{ vars.HETZNER_CACHE_ENDPOINT }}
-          bucket: ${{ vars.HETZNER_CACHE_BUCKET }}
-          accessKey: ${{ secrets.HETZNER_CACHE_ACCESS_KEY }}
-          secretKey: ${{ secrets.HETZNER_CACHE_SECRET_KEY }}
-          use-fallback: false
          path: pg_install/build/walproposer-lib
          key: v1-${{ runner.os }}-${{ runner.arch }}-${{ env.BUILD_TYPE }}-walproposer_lib-v17-${{ steps.pg_rev_v17.outputs.pg_rev }}-${{ hashFiles('Makefile') }}

--- a/.github/workflows/build_and_test.yml
+++ b/.github/workflows/build_and_test.yml
@@ -80,7 +80,6 @@ jobs:
    uses: ./.github/workflows/_meta.yml
    with:
      github-event-name: ${{ github.event_name }}
-      github-event-json: ${{ toJSON(github.event) }}

  build-build-tools-image:
    needs: [ check-permissions ]
@@ -89,8 +88,8 @@ jobs:

  check-codestyle-python:
    needs: [ meta, check-permissions, build-build-tools-image ]
-    # No need to run on `main` because we this in the merge queue. We do need to run this in `.*-rc-pr` because of hotfixes.
-    if: ${{ contains(fromJSON('["pr", "storage-rc-pr", "proxy-rc-pr", "compute-rc-pr"]'), needs.meta.outputs.run-kind) }}
+    # 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
@@ -98,8 +97,7 @@ jobs:

  check-codestyle-jsonnet:
    needs: [ meta, check-permissions, build-build-tools-image ]
-    # We do need to run this in `.*-rc-pr` because of hotfixes.
-    if: ${{ contains(fromJSON('["pr", "push-main", "storage-rc-pr", "proxy-rc-pr", "compute-rc-pr"]'), needs.meta.outputs.run-kind) }}
+    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 }}
@@ -182,8 +180,8 @@ jobs:

  check-codestyle-rust:
    needs: [ meta, check-permissions, build-build-tools-image ]
-    # No need to run on `main` because we this in the merge queue. We do need to run this in `.*-rc-pr` because of hotfixes.
-    if: ${{ contains(fromJSON('["pr", "storage-rc-pr", "proxy-rc-pr", "compute-rc-pr"]'), needs.meta.outputs.run-kind) }}
+    # 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
@@ -192,8 +190,7 @@ jobs:

  check-dependencies-rust:
    needs: [ meta, files-changed, build-build-tools-image ]
-    # No need to run on `main` because we this in the merge queue. We do need to run this in `.*-rc-pr` because of hotfixes.
-    if: ${{ needs.files-changed.outputs.check-rust-dependencies == 'true' && contains(fromJSON('["pr", "storage-rc-pr", "proxy-rc-pr", "compute-rc-pr"]'), needs.meta.outputs.run-kind) }}
+    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
@@ -201,8 +198,7 @@ jobs:

  build-and-test-locally:
    needs: [ meta, build-build-tools-image ]
-    # We do need to run this in `.*-rc-pr` because of hotfixes.
-    if: ${{ contains(fromJSON('["pr", "push-main", "storage-rc-pr", "proxy-rc-pr", "compute-rc-pr"]'), needs.meta.outputs.run-kind) }}
+    if: ${{ contains(fromJSON('["pr", "push-main"]'), needs.meta.outputs.run-kind) }}
    strategy:
      fail-fast: false
      matrix:
@@ -252,13 +248,8 @@ jobs:
        uses: actions/checkout@11bd71901bbe5b1630ceea73d27597364c9af683 # v4.2.2

      - name: Cache poetry deps
-        uses: tespkg/actions-cache@b7bf5fcc2f98a52ac6080eb0fd282c2f752074b1  # v1.8.0
+        uses: actions/cache@d4323d4df104b026a6aa633fdb11d772146be0bf # v4.2.2
        with:
-          endpoint: ${{ vars.HETZNER_CACHE_REGION }}.${{ vars.HETZNER_CACHE_ENDPOINT }}
-          bucket: ${{ vars.HETZNER_CACHE_BUCKET }}
-          accessKey: ${{ secrets.HETZNER_CACHE_ACCESS_KEY }}
-          secretKey: ${{ secrets.HETZNER_CACHE_SECRET_KEY }}
-          use-fallback: false
          path: ~/.cache/pypoetry/virtualenvs
          key: v2-${{ runner.os }}-${{ runner.arch }}-python-deps-bookworm-${{ hashFiles('poetry.lock') }}

@@ -284,7 +275,7 @@ jobs:
      statuses: write
      contents: write
      pull-requests: write
-    runs-on: [ self-hosted, unit-perf ]
+    runs-on: [ self-hosted, small-metal ]
    container:
      image: ${{ needs.build-build-tools-image.outputs.image }}-bookworm
      credentials:
@@ -549,7 +540,6 @@ jobs:
    uses: ./.github/workflows/trigger-e2e-tests.yml
    with:
      github-event-name: ${{ github.event_name }}
-      github-event-json: ${{ toJSON(github.event) }}
    secrets: inherit

  neon-image-arch:
@@ -573,7 +563,6 @@ jobs:
      - uses: actions/checkout@11bd71901bbe5b1630ceea73d27597364c9af683 # v4.2.2
        with:
          submodules: true
-          ref: ${{ needs.meta.outputs.sha }}

      - uses: neondatabase/dev-actions/set-docker-config-dir@6094485bf440001c94a94a3f9e221e81ff6b6193
      - uses: docker/setup-buildx-action@b5ca514318bd6ebac0fb2aedd5d36ec1b5c232a2 # v3.10.0
@@ -683,7 +672,6 @@ jobs:
      - uses: actions/checkout@11bd71901bbe5b1630ceea73d27597364c9af683 # v4.2.2
        with:
          submodules: true
-          ref: ${{ needs.meta.outputs.sha }}

      - uses: neondatabase/dev-actions/set-docker-config-dir@6094485bf440001c94a94a3f9e221e81ff6b6193
      - uses: docker/setup-buildx-action@b5ca514318bd6ebac0fb2aedd5d36ec1b5c232a2 # v3.10.0
@@ -980,7 +968,7 @@ jobs:
          TEST_EXTENSIONS_TAG: >-
            ${{
              contains(fromJSON('["storage-rc-pr", "proxy-rc-pr"]'), needs.meta.outputs.run-kind)
-              && needs.meta.outputs.previous-compute-release
+              && 'latest'
              || needs.meta.outputs.build-tag
            }}
          TEST_VERSION_ONLY: ${{ matrix.pg_version }}
@@ -1271,7 +1259,7 @@ jobs:
          exit 1

  deploy:
-    needs: [ check-permissions, push-neon-image-dev, push-compute-image-dev, push-neon-image-prod, push-compute-image-prod, meta, trigger-custom-extensions-build-and-wait ]
+    needs: [ check-permissions, push-neon-image-dev, push-compute-image-dev, push-neon-image-prod, push-compute-image-prod, meta, build-and-test-locally, trigger-custom-extensions-build-and-wait ]
    # `!failure() && !cancelled()` is required because the workflow depends on the job that can be skipped: `push-neon-image-prod` and `push-compute-image-prod`
    if: ${{ contains(fromJSON('["push-main", "storage-release", "proxy-release", "compute-release"]'), needs.meta.outputs.run-kind) && !failure() && !cancelled() }}
    permissions:
@@ -1568,10 +1556,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' && contains(fromJSON('["pr", "storage-rc-pr", "proxy-rc-pr", "compute-rc-pr"]'), needs.meta.outputs.run-kind))
-          || (needs.build-and-test-locally.result == 'skipped' && contains(fromJSON('["pr", "push-main", "storage-rc-pr", "proxy-rc-pr", "compute-rc-pr"]'), needs.meta.outputs.run-kind))
-          || (needs.check-codestyle-python.result == 'skipped' && contains(fromJSON('["pr", "storage-rc-pr", "proxy-rc-pr", "compute-rc-pr"]'), needs.meta.outputs.run-kind))
-          || (needs.check-codestyle-rust.result == 'skipped' && contains(fromJSON('["pr", "storage-rc-pr", "proxy-rc-pr", "compute-rc-pr"]'), needs.meta.outputs.run-kind))
+          || (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/force-test-extensions-upgrade.yml
+++ b/.github/workflows/force-test-extensions-upgrade.yml
@@ -55,7 +55,7 @@ jobs:
          echo tag=${tag} >> ${GITHUB_OUTPUT}

      - name: Test extension upgrade
-        timeout-minutes: 60
+        timeout-minutes: 20
        env:
          NEW_COMPUTE_TAG: latest
          OLD_COMPUTE_TAG: ${{ steps.get-last-compute-release-tag.outputs.tag }}
--- a/.github/workflows/report-workflow-stats-batch.yml
+++ b/.github/workflows/report-workflow-stats-batch.yml
@@ -23,7 +23,7 @@ jobs:
        egress-policy: audit

    - name: Export Workflow Run for the past 2 hours
-      uses: neondatabase/gh-workflow-stats-action@701b1f202666d0b82e67b4d387e909af2b920127 # v0.2.2
+      uses: neondatabase/gh-workflow-stats-action@4c998b25ab5cc6588b52a610b749531f6a566b6b # v0.2.1
      with:
        db_uri: ${{ secrets.GH_REPORT_STATS_DB_RW_CONNSTR }}
        db_table: "gh_workflow_stats_neon"
@@ -43,7 +43,7 @@ jobs:
        egress-policy: audit

    - name: Export Workflow Run for the past 48 hours
-      uses: neondatabase/gh-workflow-stats-action@701b1f202666d0b82e67b4d387e909af2b920127 # v0.2.2
+      uses: neondatabase/gh-workflow-stats-action@4c998b25ab5cc6588b52a610b749531f6a566b6b # v0.2.1
      with:
        db_uri: ${{ secrets.GH_REPORT_STATS_DB_RW_CONNSTR }}
        db_table: "gh_workflow_stats_neon"
@@ -63,7 +63,7 @@ jobs:
        egress-policy: audit

    - name: Export Workflow Run for the past 30 days
-      uses: neondatabase/gh-workflow-stats-action@701b1f202666d0b82e67b4d387e909af2b920127 # v0.2.2
+      uses: neondatabase/gh-workflow-stats-action@4c998b25ab5cc6588b52a610b749531f6a566b6b # v0.2.1
      with:
        db_uri: ${{ secrets.GH_REPORT_STATS_DB_RW_CONNSTR }}
        db_table: "gh_workflow_stats_neon"
--- a/.github/workflows/trigger-e2e-tests.yml
+++ b/.github/workflows/trigger-e2e-tests.yml
@@ -9,9 +9,6 @@ on:
      github-event-name:
        type: string
        required: true
-      github-event-json:
-        type: string
-        required: true

 defaults:
  run:
@@ -51,7 +48,6 @@ jobs:
    uses: ./.github/workflows/_meta.yml
    with:
      github-event-name: ${{ inputs.github-event-name || github.event_name }}
-      github-event-json: ${{ inputs.github-event-json || toJSON(github.event) }}

  trigger-e2e-tests:
    needs: [ meta ]
--- a/.gitignore
+++ b/.gitignore
@@ -1,4 +1,3 @@
-/artifact_cache
 /pg_install
 /target
 /tmp_check
--- a/Cargo.lock
+++ b/Cargo.lock
@@ -148,9 +148,9 @@ dependencies = [

 [[package]]
 name = "arc-swap"
-version = "1.7.1"
+version = "1.6.0"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "69f7f8c3906b62b754cd5326047894316021dcfe5a194c8ea52bdd94934a3457"
+checksum = "bddcadddf5e9015d310179a59bb28c4d4b9920ad0f11e8e14dbadf654890c9a6"

 [[package]]
 name = "archery"
@@ -2837,7 +2837,6 @@ dependencies = [
 "utils",
 "uuid",
 "workspace_hack",
- "x509-cert",
 ]

 [[package]]
@@ -3862,10 +3861,11 @@ dependencies = [

 [[package]]
 name = "num-bigint"
-version = "0.4.6"
+version = "0.4.3"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "a5e44f723f1133c9deac646763579fdb3ac745e418f2a7af9cd0c431da1f20b9"
+checksum = "f93ab6289c7b344a8a9f60f88d80aa20032336fe78da341afc91c8a2341fc75f"
 dependencies = [
+ "autocfg",
 "num-integer",
 "num-traits",
 ]
@@ -3914,10 +3914,11 @@ dependencies = [

 [[package]]
 name = "num-integer"
-version = "0.1.46"
+version = "0.1.45"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "7969661fd2958a5cb096e56c8e1ad0444ac2bbcd0061bd28660485a44879858f"
+checksum = "225d3389fb3509a24c93f5c29eb6bde2586b98d9f016636dff58d7c6f7569cd9"
 dependencies = [
+ "autocfg",
 "num-traits",
 ]

@@ -3946,9 +3947,9 @@ dependencies = [

 [[package]]
 name = "num-traits"
-version = "0.2.19"
+version = "0.2.15"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "071dfc062690e90b734c0b2273ce72ad0ffa95f0c74596bc250dcfd960262841"
+checksum = "578ede34cf02f8924ab9447f50c28075b4d3e5b269972345e7e0372b38c6cdcd"
 dependencies = [
 "autocfg",
 "libm",
@@ -3992,33 +3993,6 @@ dependencies = [
 "memchr",
 ]

-[[package]]
-name = "object_storage"
-version = "0.0.1"
-dependencies = [
- "anyhow",
- "axum",
- "axum-extra",
- "camino",
- "camino-tempfile",
- "futures",
- "http-body-util",
- "itertools 0.10.5",
- "jsonwebtoken",
- "prometheus",
- "rand 0.8.5",
- "remote_storage",
- "serde",
- "serde_json",
- "test-log",
- "tokio",
- "tokio-util",
- "tower 0.5.2",
- "tracing",
- "utils",
- "workspace_hack",
-]
-
 [[package]]
 name = "once_cell"
 version = "1.20.2"
@@ -4357,7 +4331,6 @@ dependencies = [
 "strum",
 "strum_macros",
 "thiserror 1.0.69",
- "tracing-utils",
 "utils",
 ]

@@ -4721,7 +4694,7 @@ dependencies = [
 [[package]]
 name = "postgres-protocol"
 version = "0.6.6"
-source = "git+https://github.com/neondatabase/rust-postgres.git?branch=neon#f3cf448febde5fd298071d54d568a9c875a7a62b"
+source = "git+https://github.com/neondatabase/rust-postgres.git?branch=neon#1f21e7959a96a34dcfbfce1b14b73286cdadffe9"
 dependencies = [
 "base64 0.22.1",
 "byteorder",
@@ -4755,7 +4728,7 @@ dependencies = [
 [[package]]
 name = "postgres-types"
 version = "0.2.6"
-source = "git+https://github.com/neondatabase/rust-postgres.git?branch=neon#f3cf448febde5fd298071d54d568a9c875a7a62b"
+source = "git+https://github.com/neondatabase/rust-postgres.git?branch=neon#1f21e7959a96a34dcfbfce1b14b73286cdadffe9"
 dependencies = [
 "bytes",
 "chrono",
@@ -5389,25 +5362,26 @@ dependencies = [

 [[package]]
 name = "redis"
-version = "0.29.2"
+version = "0.25.2"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "b110459d6e323b7cda23980c46c77157601199c9da6241552b284cd565a7a133"
+checksum = "71d64e978fd98a0e6b105d066ba4889a7301fca65aeac850a877d8797343feeb"
 dependencies = [
- "arc-swap",
+ "async-trait",
 "bytes",
 "combine",
 "futures-util",
 "itoa",
- "num-bigint",
 "percent-encoding",
 "pin-project-lite",
- "rustls 0.23.18",
- "rustls-native-certs 0.8.0",
+ "rustls 0.22.4",
+ "rustls-native-certs 0.7.0",
+ "rustls-pemfile 2.1.1",
+ "rustls-pki-types",
 "ryu",
 "sha1_smol",
 "socket2",
 "tokio",
- "tokio-rustls 0.26.0",
+ "tokio-rustls 0.25.0",
 "tokio-util",
 "url",
 ]
@@ -5495,17 +5469,6 @@ version = "1.9.0"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "c707298afce11da2efef2f600116fa93ffa7a032b5d7b628aa17711ec81383ca"

-[[package]]
-name = "remote_keys"
-version = "0.1.0"
-dependencies = [
- "anyhow",
- "aws-config",
- "aws-sdk-kms",
- "aws-smithy-types",
- "utils",
-]
-
 [[package]]
 name = "remote_storage"
 version = "0.1.0"
@@ -6964,28 +6927,6 @@ dependencies = [
 "syn 2.0.100",
 ]

-[[package]]
-name = "test-log"
-version = "0.2.17"
-source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "e7f46083d221181166e5b6f6b1e5f1d499f3a76888826e6cb1d057554157cd0f"
-dependencies = [
- "env_logger",
- "test-log-macros",
- "tracing-subscriber",
-]
-
-[[package]]
-name = "test-log-macros"
-version = "0.2.17"
-source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "888d0c3c6db53c0fdab160d2ed5e12ba745383d3e85813f2ea0f2b1475ab553f"
-dependencies = [
- "proc-macro2",
- "quote",
- "syn 2.0.100",
-]
-
 [[package]]
 name = "thiserror"
 version = "1.0.69"
@@ -7177,9 +7118,9 @@ dependencies = [

 [[package]]
 name = "tokio"
-version = "1.43.1"
+version = "1.43.0"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "492a604e2fd7f814268a378409e6c92b5525d747d10db9a229723f55a417958c"
+checksum = "3d61fa4ffa3de412bfea335c6ecff681de2b609ba3c77ef3e00e521813a9ed9e"
 dependencies = [
 "backtrace",
 "bytes",
@@ -7233,7 +7174,7 @@ dependencies = [
 [[package]]
 name = "tokio-postgres"
 version = "0.7.10"
-source = "git+https://github.com/neondatabase/rust-postgres.git?branch=neon#f3cf448febde5fd298071d54d568a9c875a7a62b"
+source = "git+https://github.com/neondatabase/rust-postgres.git?branch=neon#1f21e7959a96a34dcfbfce1b14b73286cdadffe9"
 dependencies = [
 "async-trait",
 "byteorder",
@@ -7276,14 +7217,15 @@ dependencies = [
 "bytes",
 "fallible-iterator",
 "futures-util",
+ "log",
 "parking_lot 0.12.1",
+ "phf",
 "pin-project-lite",
 "postgres-protocol2",
 "postgres-types2",
 "serde",
 "tokio",
 "tokio-util",
- "tracing",
 ]

 [[package]]
@@ -7665,7 +7607,6 @@ dependencies = [
 "opentelemetry-otlp",
 "opentelemetry-semantic-conventions",
 "opentelemetry_sdk",
- "pin-project-lite",
 "tokio",
 "tracing",
 "tracing-opentelemetry",
--- a/Cargo.toml
+++ b/Cargo.toml
@@ -30,7 +30,6 @@ members = [
    "libs/tenant_size_model",
    "libs/metrics",
    "libs/postgres_connection",
-    "libs/remote_keys",
    "libs/remote_storage",
    "libs/tracing-utils",
    "libs/postgres_ffi/wal_craft",
@@ -41,7 +40,6 @@ members = [
    "libs/proxy/postgres-protocol2",
    "libs/proxy/postgres-types2",
    "libs/proxy/tokio-postgres2",
-    "object_storage",
 ]

 [workspace.package]
@@ -52,7 +50,7 @@ license = "Apache-2.0"
 [workspace.dependencies]
 ahash = "0.8"
 anyhow = { version = "1.0", features = ["backtrace"] }
-arc-swap = "1.7"
+arc-swap = "1.6"
 async-compression = { version = "0.4.0", features = ["tokio", "gzip", "zstd"] }
 atomic-take = "1.1.0"
 flate2 = "1.0.26"
@@ -132,7 +130,7 @@ nix = { version = "0.27", features = ["dir", "fs", "process", "socket", "signal"
 # on compute startup metrics (start_postgres_ms), >= 25% degradation.
 notify = "6.0.0"
 num_cpus = "1.15"
-num-traits = "0.2.19"
+num-traits = "0.2.15"
 once_cell = "1.13"
 opentelemetry = "0.27"
 opentelemetry_sdk = "0.27"
@@ -148,7 +146,7 @@ procfs = "0.16"
 prometheus = {version = "0.13", default-features=false, features = ["process"]} # removes protobuf dependency
 prost = "0.13"
 rand = "0.8"
-redis = { version = "0.29.2", features = ["tokio-rustls-comp", "keep-alive"] }
+redis = { version = "0.25.2", features = ["tokio-rustls-comp", "keep-alive"] }
 regex = "1.10.2"
 reqwest = { version = "0.12", default-features = false, features = ["rustls-tls"] }
 reqwest-tracing = { version = "0.5", features = ["opentelemetry_0_27"] }
@@ -185,7 +183,7 @@ test-context = "0.3"
 thiserror = "1.0"
 tikv-jemallocator = { version = "0.6", features = ["profiling", "stats", "unprefixed_malloc_on_supported_platforms"] }
 tikv-jemalloc-ctl = { version = "0.6", features = ["stats"] }
-tokio = { version = "1.43.1", features = ["macros"] }
+tokio = { version = "1.41", features = ["macros"] }
 tokio-epoll-uring = { git = "https://github.com/neondatabase/tokio-epoll-uring.git" , branch = "main" }
 tokio-io-timeout = "1.2.0"
 tokio-postgres-rustls = "0.12.0"
@@ -210,7 +208,6 @@ tracing-opentelemetry = "0.28"
 tracing-serde = "0.2.0"
 tracing-subscriber = { version = "0.3", default-features = false, features = ["smallvec", "fmt", "tracing-log", "std", "env-filter", "json"] }
 try-lock = "0.2.5"
-test-log = { version = "0.2.17", default-features = false, features = ["log"] }
 twox-hash = { version = "1.6.3", default-features = false }
 typed-json = "0.1"
 url = "2.2"
--- a/2
+++ b/2
@@ -89,7 +89,6 @@ RUN set -e \
      --bin storage_broker  \
      --bin storage_controller  \
      --bin proxy  \
-      --bin object_storage \
      --bin neon_local \
      --bin storage_scrubber \
      --locked --release
@@ -122,7 +121,6 @@ COPY --from=build --chown=neon:neon /home/nonroot/target/release/safekeeper
 COPY --from=build --chown=neon:neon /home/nonroot/target/release/storage_broker      /usr/local/bin
 COPY --from=build --chown=neon:neon /home/nonroot/target/release/storage_controller  /usr/local/bin
 COPY --from=build --chown=neon:neon /home/nonroot/target/release/proxy               /usr/local/bin
-COPY --from=build --chown=neon:neon /home/nonroot/target/release/object_storage      /usr/local/bin
 COPY --from=build --chown=neon:neon /home/nonroot/target/release/neon_local          /usr/local/bin
 COPY --from=build --chown=neon:neon /home/nonroot/target/release/storage_scrubber    /usr/local/bin

--- a/build-tools.Dockerfile
+++ b/build-tools.Dockerfile
@@ -292,7 +292,7 @@ WORKDIR /home/nonroot

 # Rust
 # Please keep the version of llvm (installed above) in sync with rust llvm (`rustc --version --verbose | grep LLVM`)
-ENV RUSTC_VERSION=1.86.0
+ENV RUSTC_VERSION=1.85.0
 ENV RUSTUP_HOME="/home/nonroot/.rustup"
 ENV PATH="/home/nonroot/.cargo/bin:${PATH}"
 ARG RUSTFILT_VERSION=0.2.1
--- a/compute/compute-node.Dockerfile
+++ b/compute/compute-node.Dockerfile
@@ -369,7 +369,7 @@ FROM build-deps AS plv8-src
 ARG PG_VERSION
 WORKDIR /ext-src

-COPY compute/patches/plv8* .
+COPY compute/patches/plv8-3.1.10.patch .

 # plv8 3.2.3 supports v17
 # last release v3.2.3 - Sep 7, 2024
@@ -393,7 +393,7 @@ RUN case "${PG_VERSION:?}" in \
    git clone --recurse-submodules --depth 1 --branch ${PLV8_TAG} https://github.com/plv8/plv8.git plv8-src && \
    tar -czf plv8.tar.gz --exclude .git plv8-src && \
    cd plv8-src && \
-    if [[ "${PG_VERSION:?}" < "v17" ]]; then patch -p1 < /ext-src/plv8_v3.1.10.patch; else patch -p1 < /ext-src/plv8_v3.2.3.patch; fi
+    if [[ "${PG_VERSION:?}" < "v17" ]]; then patch -p1 < /ext-src/plv8-3.1.10.patch; fi

 # Step 1: Build the vendored V8 engine. It doesn't depend on PostgreSQL, so use
 # 'build-deps' as the base. This enables caching and avoids unnecessary rebuilds.
@@ -1022,6 +1022,67 @@ RUN make -j $(getconf _NPROCESSORS_ONLN) && \
    make -j $(getconf _NPROCESSORS_ONLN) install && \
    echo 'trusted = true' >> /usr/local/pgsql/share/extension/semver.control

+#########################################################################################
+#
+# Layer "pg_embedding-build"
+# compile pg_embedding extension
+#
+#########################################################################################
+FROM build-deps AS pg_embedding-src
+ARG PG_VERSION
+
+# This is our extension, support stopped in favor of pgvector
+# TODO: deprecate it
+WORKDIR /ext-src
+RUN case "${PG_VERSION:?}" in \
+      "v14" | "v15") \
+        export PG_EMBEDDING_VERSION=0.3.5 \
+        export PG_EMBEDDING_CHECKSUM=0e95b27b8b6196e2cf0a0c9ec143fe2219b82e54c5bb4ee064e76398cbe69ae9 \
+        ;; \
+      *) \
+        echo "pg_embedding not supported on this PostgreSQL version. Use pgvector instead." && exit 0;; \
+    esac && \
+    wget https://github.com/neondatabase/pg_embedding/archive/refs/tags/${PG_EMBEDDING_VERSION}.tar.gz -O pg_embedding.tar.gz && \
+    echo "${PG_EMBEDDING_CHECKSUM} pg_embedding.tar.gz" | sha256sum --check && \
+    mkdir pg_embedding-src && cd pg_embedding-src && tar xzf ../pg_embedding.tar.gz --strip-components=1 -C .
+
+FROM pg-build AS pg_embedding-build
+COPY --from=pg_embedding-src /ext-src/ /ext-src/
+WORKDIR /ext-src/
+RUN  if [ -d pg_embedding-src ]; then \
+        cd pg_embedding-src && \
+        make -j $(getconf _NPROCESSORS_ONLN) && \
+        make -j $(getconf _NPROCESSORS_ONLN) install; \
+    fi
+
+#########################################################################################
+#
+# Layer "pg_anon-build"
+# compile anon extension
+#
+#########################################################################################
+FROM build-deps AS pg_anon-src
+ARG PG_VERSION
+
+# This is an experimental extension, never got to real production.
+# !Do not remove! It can be present in shared_preload_libraries and compute will fail to start if library is not found.
+WORKDIR /ext-src
+RUN case "${PG_VERSION:?}" in "v17") \
+    echo "postgresql_anonymizer does not yet support PG17" && exit 0;; \
+    esac && \
+    wget  https://github.com/neondatabase/postgresql_anonymizer/archive/refs/tags/neon_1.1.1.tar.gz -O pg_anon.tar.gz && \
+    echo "321ea8d5c1648880aafde850a2c576e4a9e7b9933a34ce272efc839328999fa9  pg_anon.tar.gz" | sha256sum --check && \
+    mkdir pg_anon-src && cd pg_anon-src && tar xzf ../pg_anon.tar.gz --strip-components=1 -C .
+
+FROM pg-build AS pg_anon-build
+COPY --from=pg_anon-src /ext-src/ /ext-src/
+WORKDIR /ext-src
+RUN if [ -d pg_anon-src ]; then \
+        cd pg_anon-src && \
+        make -j $(getconf _NPROCESSORS_ONLN) install && \
+        echo 'trusted = true' >> /usr/local/pgsql/share/extension/anon.control; \
+    fi
+
 #########################################################################################
 #
 # Layer "pg build with nonroot user and cargo installed"
@@ -1305,8 +1366,8 @@ ARG PG_VERSION
 # Do not update without approve from proxy team
 # Make sure the version is reflected in proxy/src/serverless/local_conn_pool.rs
 WORKDIR /ext-src
-RUN wget https://github.com/neondatabase/pg_session_jwt/archive/refs/tags/v0.3.0.tar.gz -O pg_session_jwt.tar.gz && \
-    echo "19be2dc0b3834d643706ed430af998bb4c2cdf24b3c45e7b102bb3a550e8660c pg_session_jwt.tar.gz" | sha256sum --check && \
+RUN wget https://github.com/neondatabase/pg_session_jwt/archive/refs/tags/v0.2.0.tar.gz -O pg_session_jwt.tar.gz && \
+    echo "5ace028e591f2e000ca10afa5b1ca62203ebff014c2907c0ec3b29c36f28a1bb pg_session_jwt.tar.gz" | sha256sum --check && \
    mkdir pg_session_jwt-src && cd pg_session_jwt-src && tar xzf ../pg_session_jwt.tar.gz --strip-components=1 -C . && \
    sed -i 's/pgrx = "0.12.6"/pgrx = { version = "0.12.9", features = [ "unsafe-postgres" ] }/g' Cargo.toml && \
    sed -i 's/version = "0.12.6"/version = "0.12.9"/g' pgrx-tests/Cargo.toml && \
@@ -1614,7 +1675,9 @@ COPY --from=rdkit-build /usr/local/pgsql/ /usr/local/pgsql/
 COPY --from=pg_uuidv7-build /usr/local/pgsql/ /usr/local/pgsql/
 COPY --from=pg_roaringbitmap-build /usr/local/pgsql/ /usr/local/pgsql/
 COPY --from=pg_semver-build /usr/local/pgsql/ /usr/local/pgsql/
+COPY --from=pg_embedding-build /usr/local/pgsql/ /usr/local/pgsql/
 COPY --from=wal2json-build /usr/local/pgsql /usr/local/pgsql
+COPY --from=pg_anon-build /usr/local/pgsql/ /usr/local/pgsql/
 COPY --from=pg_ivm-build /usr/local/pgsql/ /usr/local/pgsql/
 COPY --from=pg_partman-build /usr/local/pgsql/ /usr/local/pgsql/
 COPY --from=pg_mooncake-build /usr/local/pgsql/ /usr/local/pgsql/
@@ -1790,6 +1853,7 @@ COPY --from=pg_cron-src /ext-src/ /ext-src/
 COPY --from=pg_uuidv7-src /ext-src/ /ext-src/
 COPY --from=pg_roaringbitmap-src /ext-src/ /ext-src/
 COPY --from=pg_semver-src /ext-src/ /ext-src/
+#COPY --from=pg_embedding-src /ext-src/ /ext-src/
 #COPY --from=wal2json-src /ext-src/ /ext-src/
 COPY --from=pg_ivm-src /ext-src/ /ext-src/
 COPY --from=pg_partman-src /ext-src/ /ext-src/
@@ -1852,30 +1916,26 @@ RUN apt update && \
      ;; \
    esac && \
    apt install --no-install-recommends -y \
-        ca-certificates \
        gdb \
-        iproute2 \
+        liblz4-1 \
+        libreadline8 \
        libboost-iostreams1.74.0 \
        libboost-regex1.74.0 \
        libboost-serialization1.74.0 \
        libboost-system1.74.0 \
-        libcurl4 \
-        libevent-2.1-7 \
-        libgeos-c1v5 \
-        liblz4-1 \
        libossp-uuid16 \
+        libgeos-c1v5 \
        libprotobuf-c1 \
-        libreadline8 \
        libsfcgal1 \
        libxml2 \
        libxslt1.1 \
        libzstd1 \
+        libcurl4 \
+        libevent-2.1-7 \
        locales \
-        lsof \
        procps \
+        ca-certificates \
        rsyslog \
-        screen \
-        tcpdump \
        $VERSION_INSTALLS && \
    apt clean && rm -rf /var/lib/apt/lists/* /tmp/* /var/tmp/* && \
    localedef -i en_US -c -f UTF-8 -A /usr/share/locale/locale.alias en_US.UTF-8
--- a/compute/etc/neon_collector.jsonnet
+++ b/compute/etc/neon_collector.jsonnet
@@ -33,7 +33,6 @@
    import 'sql_exporter/lfc_hits.libsonnet',
    import 'sql_exporter/lfc_misses.libsonnet',
    import 'sql_exporter/lfc_used.libsonnet',
-    import 'sql_exporter/lfc_used_pages.libsonnet',
    import 'sql_exporter/lfc_writes.libsonnet',
    import 'sql_exporter/logical_slot_restart_lsn.libsonnet',
    import 'sql_exporter/max_cluster_size.libsonnet',
--- a/compute/etc/sql_exporter/lfc_used_pages.libsonnet
+++ b/compute/etc/sql_exporter/lfc_used_pages.libsonnet
@@ -1,10 +0,0 @@
-{
-  metric_name: 'lfc_used_pages',
-  type: 'gauge',
-  help: 'LFC pages used',
-  key_labels: null,
-  values: [
-    'lfc_used_pages',
-  ],
-  query: importstr 'sql_exporter/lfc_used_pages.sql',
-}
--- a/compute/etc/sql_exporter/lfc_used_pages.sql
+++ b/compute/etc/sql_exporter/lfc_used_pages.sql
@@ -1 +0,0 @@
-SELECT lfc_value AS lfc_used_pages FROM neon.neon_lfc_stats WHERE lfc_key = 'file_cache_used_pages';
--- a/compute/patches/cloud_regress_pg16.patch
+++ b/compute/patches/cloud_regress_pg16.patch
@@ -202,10 +202,10 @@ index cf0b80d616..e8e2a14a4a 100644
 COMMENT ON CONSTRAINT the_constraint ON constraint_comments_tbl IS 'no, the comment';
 ERROR:  must be owner of relation constraint_comments_tbl
 diff --git a/src/test/regress/expected/conversion.out b/src/test/regress/expected/conversion.out
-index d785f92561..16377e5ac9 100644
+index 442e7aff2b..525f732b03 100644
 --- a/src/test/regress/expected/conversion.out
 +++ b/src/test/regress/expected/conversion.out
-@@ -15,7 +15,7 @@ SELECT FROM test_enc_setup();
+@@ -8,7 +8,7 @@
 CREATE FUNCTION test_enc_conversion(bytea, name, name, bool, validlen OUT int, result OUT bytea)
     AS :'regresslib', 'test_enc_conversion'
     LANGUAGE C STRICT;
@@ -587,15 +587,16 @@ index f551624afb..57f1e432d4 100644
 SELECT *
    INTO TABLE ramp
 diff --git a/src/test/regress/expected/database.out b/src/test/regress/expected/database.out
-index 4cbdbdf84d..573362850e 100644
+index 454db91ec0..01378d7081 100644
 --- a/src/test/regress/expected/database.out
 +++ b/src/test/regress/expected/database.out
-@@ -1,8 +1,6 @@
+@@ -1,8 +1,7 @@
 CREATE DATABASE regression_tbd
 	ENCODING utf8 LC_COLLATE "C" LC_CTYPE "C" TEMPLATE template0;
 ALTER DATABASE regression_tbd RENAME TO regression_utf8;
 -ALTER DATABASE regression_utf8 SET TABLESPACE regress_tblspace;
 -ALTER DATABASE regression_utf8 RESET TABLESPACE;
+WARNING:  you need to manually restart any running background workers after this command
 ALTER DATABASE regression_utf8 CONNECTION_LIMIT 123;
 -- Test PgDatabaseToastTable.  Doing this with GRANT would be slow.
 BEGIN;
@@ -699,7 +700,7 @@ index 6ed50fdcfa..caa00a345d 100644
 COMMENT ON FOREIGN DATA WRAPPER dummy IS 'useless';
 CREATE FOREIGN DATA WRAPPER postgresql VALIDATOR postgresql_fdw_validator;
 diff --git a/src/test/regress/expected/foreign_key.out b/src/test/regress/expected/foreign_key.out
-index 84745b9f60..4883c12351 100644
+index 6b8c2f2414..8e13b7fa46 100644
 --- a/src/test/regress/expected/foreign_key.out
 +++ b/src/test/regress/expected/foreign_key.out
@@ -1985,7 +1985,7 @@ ALTER TABLE fk_partitioned_fk_6 ATTACH PARTITION fk_partitioned_pk_6 FOR VALUES
@@ -1111,7 +1112,7 @@ index 8475231735..0653946337 100644
 DROP ROLE regress_passwd_sha_len1;
 DROP ROLE regress_passwd_sha_len2;
 diff --git a/src/test/regress/expected/privileges.out b/src/test/regress/expected/privileges.out
-index 620fbe8c52..0570102357 100644
+index 5b9dba7b32..cc408dad42 100644
 --- a/src/test/regress/expected/privileges.out
 +++ b/src/test/regress/expected/privileges.out
@@ -20,19 +20,19 @@ SELECT lo_unlink(oid) FROM pg_largeobject_metadata WHERE oid >= 1000 AND oid < 3
@@ -1173,8 +1174,8 @@ index 620fbe8c52..0570102357 100644
 +CREATE GROUP regress_priv_group2 WITH ADMIN regress_priv_user1 PASSWORD NEON_PASSWORD_PLACEHOLDER USER regress_priv_user2;
 ALTER GROUP regress_priv_group1 ADD USER regress_priv_user4;
 GRANT regress_priv_group2 TO regress_priv_user2 GRANTED BY regress_priv_user1;
- SET SESSION AUTHORIZATION regress_priv_user3;
-@@ -246,12 +246,16 @@ GRANT regress_priv_role TO regress_priv_user1 WITH ADMIN OPTION GRANTED BY regre
+ SET SESSION AUTHORIZATION regress_priv_user1;
+@@ -239,12 +239,16 @@ GRANT regress_priv_role TO regress_priv_user1 WITH ADMIN OPTION GRANTED BY regre
 ERROR:  permission denied to grant privileges as role "regress_priv_role"
 DETAIL:  The grantor must have the ADMIN option on role "regress_priv_role".
 GRANT regress_priv_role TO regress_priv_user1 WITH ADMIN OPTION GRANTED BY CURRENT_ROLE;
@@ -1191,7 +1192,7 @@ index 620fbe8c52..0570102357 100644
 DROP ROLE regress_priv_role;
 SET SESSION AUTHORIZATION regress_priv_user1;
 SELECT session_user, current_user;
-@@ -1783,7 +1787,7 @@ SELECT has_table_privilege('regress_priv_user1', 'atest4', 'SELECT WITH GRANT OP
+@@ -1776,7 +1780,7 @@ SELECT has_table_privilege('regress_priv_user1', 'atest4', 'SELECT WITH GRANT OP
 
 -- security-restricted operations
 \c -
@@ -1200,7 +1201,7 @@ index 620fbe8c52..0570102357 100644
 -- Check that index expressions and predicates are run as the table's owner
 -- A dummy index function checking current_user
 CREATE FUNCTION sro_ifun(int) RETURNS int AS $$
-@@ -2675,8 +2679,8 @@ drop cascades to function testns.priv_testagg(integer)
+@@ -2668,8 +2672,8 @@ drop cascades to function testns.priv_testagg(integer)
 drop cascades to function testns.priv_testproc(integer)
 -- Change owner of the schema & and rename of new schema owner
 \c -
@@ -1211,7 +1212,7 @@ index 620fbe8c52..0570102357 100644
 SET SESSION ROLE regress_schemauser1;
 CREATE SCHEMA testns;
 SELECT nspname, rolname FROM pg_namespace, pg_roles WHERE pg_namespace.nspname = 'testns' AND pg_namespace.nspowner = pg_roles.oid;
-@@ -2799,7 +2803,7 @@ DROP USER regress_priv_user7;
+@@ -2792,7 +2796,7 @@ DROP USER regress_priv_user7;
 DROP USER regress_priv_user8; -- does not exist
 ERROR:  role "regress_priv_user8" does not exist
 -- permissions with LOCK TABLE
@@ -1220,7 +1221,7 @@ index 620fbe8c52..0570102357 100644
 CREATE TABLE lock_table (a int);
 -- LOCK TABLE and SELECT permission
 GRANT SELECT ON lock_table TO regress_locktable_user;
-@@ -2881,7 +2885,7 @@ DROP USER regress_locktable_user;
+@@ -2874,7 +2878,7 @@ DROP USER regress_locktable_user;
 -- pg_backend_memory_contexts.
 -- switch to superuser
 \c -
@@ -1229,7 +1230,7 @@ index 620fbe8c52..0570102357 100644
 SELECT has_table_privilege('regress_readallstats','pg_backend_memory_contexts','SELECT'); -- no
  has_table_privilege 
 ---------------------
-@@ -2925,10 +2929,10 @@ RESET ROLE;
+@@ -2918,10 +2922,10 @@ RESET ROLE;
 -- clean up
 DROP ROLE regress_readallstats;
 -- test role grantor machinery
@@ -1244,7 +1245,7 @@ index 620fbe8c52..0570102357 100644
 GRANT regress_group TO regress_group_direct_manager WITH INHERIT FALSE, ADMIN TRUE;
 GRANT regress_group_direct_manager TO regress_group_indirect_manager;
 SET SESSION AUTHORIZATION regress_group_direct_manager;
-@@ -2957,9 +2961,9 @@ DROP ROLE regress_group_direct_manager;
+@@ -2950,9 +2954,9 @@ DROP ROLE regress_group_direct_manager;
 DROP ROLE regress_group_indirect_manager;
 DROP ROLE regress_group_member;
 -- test SET and INHERIT options with object ownership changes
@@ -1840,7 +1841,7 @@ index 09a255649b..15895f0c53 100644
 CREATE TABLE ruletest_t2 (x int);
 CREATE VIEW ruletest_v1 WITH (security_invoker=true) AS
 diff --git a/src/test/regress/expected/security_label.out b/src/test/regress/expected/security_label.out
-index a8e01a6220..83543b250a 100644
+index a8e01a6220..5a9cef4ede 100644
 --- a/src/test/regress/expected/security_label.out
 +++ b/src/test/regress/expected/security_label.out
@@ -6,8 +6,8 @@ SET client_min_messages TO 'warning';
@@ -1854,6 +1855,34 @@ index a8e01a6220..83543b250a 100644
 CREATE TABLE seclabel_tbl1 (a int, b text);
 CREATE TABLE seclabel_tbl2 (x int, y text);
 CREATE VIEW seclabel_view1 AS SELECT * FROM seclabel_tbl2;
+@@ -19,21 +19,21 @@ ALTER TABLE seclabel_tbl2 OWNER TO regress_seclabel_user2;
+ -- Test of SECURITY LABEL statement without a plugin
+ --
+ SECURITY LABEL ON TABLE seclabel_tbl1 IS 'classified';			-- fail
+-ERROR:  no security label providers have been loaded
+ERROR:  must specify provider when multiple security label providers have been loaded
+ SECURITY LABEL FOR 'dummy' ON TABLE seclabel_tbl1 IS 'classified';		-- fail
+ ERROR:  security label provider "dummy" is not loaded
+ SECURITY LABEL ON TABLE seclabel_tbl1 IS '...invalid label...';		-- fail
+-ERROR:  no security label providers have been loaded
+ERROR:  must specify provider when multiple security label providers have been loaded
+ SECURITY LABEL ON TABLE seclabel_tbl3 IS 'unclassified';			-- fail
+-ERROR:  no security label providers have been loaded
+ERROR:  must specify provider when multiple security label providers have been loaded
+ SECURITY LABEL ON ROLE regress_seclabel_user1 IS 'classified';			-- fail
+-ERROR:  no security label providers have been loaded
+ERROR:  must specify provider when multiple security label providers have been loaded
+ SECURITY LABEL FOR 'dummy' ON ROLE regress_seclabel_user1 IS 'classified';		-- fail
+ ERROR:  security label provider "dummy" is not loaded
+ SECURITY LABEL ON ROLE regress_seclabel_user1 IS '...invalid label...';		-- fail
+-ERROR:  no security label providers have been loaded
+ERROR:  must specify provider when multiple security label providers have been loaded
+ SECURITY LABEL ON ROLE regress_seclabel_user3 IS 'unclassified';			-- fail
+-ERROR:  no security label providers have been loaded
+ERROR:  must specify provider when multiple security label providers have been loaded
+ -- clean up objects
+ DROP FUNCTION seclabel_four();
+ DROP DOMAIN seclabel_domain;
 diff --git a/src/test/regress/expected/select_into.out b/src/test/regress/expected/select_into.out
 index b79fe9a1c0..e29fab88ab 100644
 --- a/src/test/regress/expected/select_into.out
@@ -2384,10 +2413,10 @@ index e3e3bea709..fa86ddc326 100644
 COMMENT ON CONSTRAINT the_constraint ON constraint_comments_tbl IS 'no, the comment';
 COMMENT ON CONSTRAINT the_constraint ON DOMAIN constraint_comments_dom IS 'no, another comment';
 diff --git a/src/test/regress/sql/conversion.sql b/src/test/regress/sql/conversion.sql
-index b567a1a572..4d1ac2e631 100644
+index 9a65fca91f..58431a3056 100644
 --- a/src/test/regress/sql/conversion.sql
 +++ b/src/test/regress/sql/conversion.sql
-@@ -17,7 +17,7 @@ CREATE FUNCTION test_enc_conversion(bytea, name, name, bool, validlen OUT int, r
+@@ -12,7 +12,7 @@ CREATE FUNCTION test_enc_conversion(bytea, name, name, bool, validlen OUT int, r
     AS :'regresslib', 'test_enc_conversion'
     LANGUAGE C STRICT;
 
@@ -2751,7 +2780,7 @@ index ae6841308b..47bc792e30 100644
 
 SELECT *
 diff --git a/src/test/regress/sql/database.sql b/src/test/regress/sql/database.sql
-index 46ad263478..eb05584ed5 100644
+index 0367c0e37a..a23b98c4bd 100644
 --- a/src/test/regress/sql/database.sql
 +++ b/src/test/regress/sql/database.sql
@@ -1,8 +1,6 @@
@@ -2864,7 +2893,7 @@ index aa147b14a9..370e0dd570 100644
 CREATE FOREIGN DATA WRAPPER dummy;
 COMMENT ON FOREIGN DATA WRAPPER dummy IS 'useless';
 diff --git a/src/test/regress/sql/foreign_key.sql b/src/test/regress/sql/foreign_key.sql
-index 9f4210b26e..620d3fc87e 100644
+index 45c7a534cb..32dd26b8cd 100644
 --- a/src/test/regress/sql/foreign_key.sql
 +++ b/src/test/regress/sql/foreign_key.sql
@@ -1435,7 +1435,7 @@ ALTER TABLE fk_partitioned_fk_6 ATTACH PARTITION fk_partitioned_pk_6 FOR VALUES
@@ -3217,7 +3246,7 @@ index 53e86b0b6c..0303fdfe96 100644
 -- Check that the invalid secrets were re-hashed. A re-hashed secret
 -- should not contain the original salt.
 diff --git a/src/test/regress/sql/privileges.sql b/src/test/regress/sql/privileges.sql
-index 259f1aedd1..6e1a3d17b7 100644
+index 249df17a58..b258e7f26a 100644
 --- a/src/test/regress/sql/privileges.sql
 +++ b/src/test/regress/sql/privileges.sql
@@ -24,18 +24,18 @@ RESET client_min_messages;
@@ -3279,7 +3308,7 @@ index 259f1aedd1..6e1a3d17b7 100644
 
 ALTER GROUP regress_priv_group1 ADD USER regress_priv_user4;
 
-@@ -1160,7 +1160,7 @@ SELECT has_table_privilege('regress_priv_user1', 'atest4', 'SELECT WITH GRANT OP
+@@ -1157,7 +1157,7 @@ SELECT has_table_privilege('regress_priv_user1', 'atest4', 'SELECT WITH GRANT OP
 
 -- security-restricted operations
 \c -
@@ -3288,7 +3317,7 @@ index 259f1aedd1..6e1a3d17b7 100644
 
 -- Check that index expressions and predicates are run as the table's owner
 
-@@ -1656,8 +1656,8 @@ DROP SCHEMA testns CASCADE;
+@@ -1653,8 +1653,8 @@ DROP SCHEMA testns CASCADE;
 -- Change owner of the schema & and rename of new schema owner
 \c -
 
@@ -3299,7 +3328,7 @@ index 259f1aedd1..6e1a3d17b7 100644
 
 SET SESSION ROLE regress_schemauser1;
 CREATE SCHEMA testns;
-@@ -1751,7 +1751,7 @@ DROP USER regress_priv_user8; -- does not exist
+@@ -1748,7 +1748,7 @@ DROP USER regress_priv_user8; -- does not exist
 
 
 -- permissions with LOCK TABLE
@@ -3308,7 +3337,7 @@ index 259f1aedd1..6e1a3d17b7 100644
 CREATE TABLE lock_table (a int);
 
 -- LOCK TABLE and SELECT permission
-@@ -1839,7 +1839,7 @@ DROP USER regress_locktable_user;
+@@ -1836,7 +1836,7 @@ DROP USER regress_locktable_user;
 -- switch to superuser
 \c -
 
@@ -3317,7 +3346,7 @@ index 259f1aedd1..6e1a3d17b7 100644
 
 SELECT has_table_privilege('regress_readallstats','pg_backend_memory_contexts','SELECT'); -- no
 SELECT has_table_privilege('regress_readallstats','pg_shmem_allocations','SELECT'); -- no
-@@ -1859,10 +1859,10 @@ RESET ROLE;
+@@ -1856,10 +1856,10 @@ RESET ROLE;
 DROP ROLE regress_readallstats;
 
 -- test role grantor machinery
@@ -3332,7 +3361,7 @@ index 259f1aedd1..6e1a3d17b7 100644
 
 GRANT regress_group TO regress_group_direct_manager WITH INHERIT FALSE, ADMIN TRUE;
 GRANT regress_group_direct_manager TO regress_group_indirect_manager;
-@@ -1884,9 +1884,9 @@ DROP ROLE regress_group_indirect_manager;
+@@ -1881,9 +1881,9 @@ DROP ROLE regress_group_indirect_manager;
 DROP ROLE regress_group_member;
 
 -- test SET and INHERIT options with object ownership changes
--- a/compute/patches/cloud_regress_pg17.patch
+++ b/compute/patches/cloud_regress_pg17.patch
@@ -202,10 +202,10 @@ index cf0b80d616..e8e2a14a4a 100644
 COMMENT ON CONSTRAINT the_constraint ON constraint_comments_tbl IS 'no, the comment';
 ERROR:  must be owner of relation constraint_comments_tbl
 diff --git a/src/test/regress/expected/conversion.out b/src/test/regress/expected/conversion.out
-index d785f92561..16377e5ac9 100644
+index 442e7aff2b..525f732b03 100644
 --- a/src/test/regress/expected/conversion.out
 +++ b/src/test/regress/expected/conversion.out
-@@ -15,7 +15,7 @@ SELECT FROM test_enc_setup();
+@@ -8,7 +8,7 @@
 CREATE FUNCTION test_enc_conversion(bytea, name, name, bool, validlen OUT int, result OUT bytea)
     AS :'regresslib', 'test_enc_conversion'
     LANGUAGE C STRICT;
@@ -587,15 +587,16 @@ index f551624afb..57f1e432d4 100644
 SELECT *
    INTO TABLE ramp
 diff --git a/src/test/regress/expected/database.out b/src/test/regress/expected/database.out
-index 4cbdbdf84d..573362850e 100644
+index 454db91ec0..01378d7081 100644
 --- a/src/test/regress/expected/database.out
 +++ b/src/test/regress/expected/database.out
-@@ -1,8 +1,6 @@
+@@ -1,8 +1,7 @@
 CREATE DATABASE regression_tbd
 	ENCODING utf8 LC_COLLATE "C" LC_CTYPE "C" TEMPLATE template0;
 ALTER DATABASE regression_tbd RENAME TO regression_utf8;
 -ALTER DATABASE regression_utf8 SET TABLESPACE regress_tblspace;
 -ALTER DATABASE regression_utf8 RESET TABLESPACE;
+WARNING:  you need to manually restart any running background workers after this command
 ALTER DATABASE regression_utf8 CONNECTION_LIMIT 123;
 -- Test PgDatabaseToastTable.  Doing this with GRANT would be slow.
 BEGIN;
@@ -699,7 +700,7 @@ index 6ed50fdcfa..caa00a345d 100644
 COMMENT ON FOREIGN DATA WRAPPER dummy IS 'useless';
 CREATE FOREIGN DATA WRAPPER postgresql VALIDATOR postgresql_fdw_validator;
 diff --git a/src/test/regress/expected/foreign_key.out b/src/test/regress/expected/foreign_key.out
-index fe6a1015f2..614b387b7d 100644
+index 69994c98e3..129abcfbe8 100644
 --- a/src/test/regress/expected/foreign_key.out
 +++ b/src/test/regress/expected/foreign_key.out
@@ -1985,7 +1985,7 @@ ALTER TABLE fk_partitioned_fk_6 ATTACH PARTITION fk_partitioned_pk_6 FOR VALUES
@@ -1146,7 +1147,7 @@ index 924d6e001d..7fdda73439 100644
 DROP ROLE regress_passwd_sha_len1;
 DROP ROLE regress_passwd_sha_len2;
 diff --git a/src/test/regress/expected/privileges.out b/src/test/regress/expected/privileges.out
-index e8c668e0a1..03be5c2120 100644
+index 1296da0d57..f43fffa44c 100644
 --- a/src/test/regress/expected/privileges.out
 +++ b/src/test/regress/expected/privileges.out
@@ -20,19 +20,19 @@ SELECT lo_unlink(oid) FROM pg_largeobject_metadata WHERE oid >= 1000 AND oid < 3
@@ -1208,8 +1209,8 @@ index e8c668e0a1..03be5c2120 100644
 +CREATE GROUP regress_priv_group2 WITH ADMIN regress_priv_user1 PASSWORD NEON_PASSWORD_PLACEHOLDER USER regress_priv_user2;
 ALTER GROUP regress_priv_group1 ADD USER regress_priv_user4;
 GRANT regress_priv_group2 TO regress_priv_user2 GRANTED BY regress_priv_user1;
- SET SESSION AUTHORIZATION regress_priv_user3;
-@@ -246,12 +246,16 @@ GRANT regress_priv_role TO regress_priv_user1 WITH ADMIN OPTION GRANTED BY regre
+ SET SESSION AUTHORIZATION regress_priv_user1;
+@@ -239,12 +239,16 @@ GRANT regress_priv_role TO regress_priv_user1 WITH ADMIN OPTION GRANTED BY regre
 ERROR:  permission denied to grant privileges as role "regress_priv_role"
 DETAIL:  The grantor must have the ADMIN option on role "regress_priv_role".
 GRANT regress_priv_role TO regress_priv_user1 WITH ADMIN OPTION GRANTED BY CURRENT_ROLE;
@@ -1226,7 +1227,7 @@ index e8c668e0a1..03be5c2120 100644
 DROP ROLE regress_priv_role;
 SET SESSION AUTHORIZATION regress_priv_user1;
 SELECT session_user, current_user;
-@@ -1783,7 +1787,7 @@ SELECT has_table_privilege('regress_priv_user1', 'atest4', 'SELECT WITH GRANT OP
+@@ -1776,7 +1780,7 @@ SELECT has_table_privilege('regress_priv_user1', 'atest4', 'SELECT WITH GRANT OP
 
 -- security-restricted operations
 \c -
@@ -1235,7 +1236,7 @@ index e8c668e0a1..03be5c2120 100644
 -- Check that index expressions and predicates are run as the table's owner
 -- A dummy index function checking current_user
 CREATE FUNCTION sro_ifun(int) RETURNS int AS $$
-@@ -2675,8 +2679,8 @@ drop cascades to function testns.priv_testagg(integer)
+@@ -2668,8 +2672,8 @@ drop cascades to function testns.priv_testagg(integer)
 drop cascades to function testns.priv_testproc(integer)
 -- Change owner of the schema & and rename of new schema owner
 \c -
@@ -1246,7 +1247,7 @@ index e8c668e0a1..03be5c2120 100644
 SET SESSION ROLE regress_schemauser1;
 CREATE SCHEMA testns;
 SELECT nspname, rolname FROM pg_namespace, pg_roles WHERE pg_namespace.nspname = 'testns' AND pg_namespace.nspowner = pg_roles.oid;
-@@ -2799,7 +2803,7 @@ DROP USER regress_priv_user7;
+@@ -2792,7 +2796,7 @@ DROP USER regress_priv_user7;
 DROP USER regress_priv_user8; -- does not exist
 ERROR:  role "regress_priv_user8" does not exist
 -- permissions with LOCK TABLE
@@ -1255,7 +1256,7 @@ index e8c668e0a1..03be5c2120 100644
 CREATE TABLE lock_table (a int);
 -- LOCK TABLE and SELECT permission
 GRANT SELECT ON lock_table TO regress_locktable_user;
-@@ -2895,7 +2899,7 @@ DROP USER regress_locktable_user;
+@@ -2888,7 +2892,7 @@ DROP USER regress_locktable_user;
 -- pg_backend_memory_contexts.
 -- switch to superuser
 \c -
@@ -1264,7 +1265,7 @@ index e8c668e0a1..03be5c2120 100644
 SELECT has_table_privilege('regress_readallstats','pg_backend_memory_contexts','SELECT'); -- no
  has_table_privilege 
 ---------------------
-@@ -2939,10 +2943,10 @@ RESET ROLE;
+@@ -2932,10 +2936,10 @@ RESET ROLE;
 -- clean up
 DROP ROLE regress_readallstats;
 -- test role grantor machinery
@@ -1279,7 +1280,7 @@ index e8c668e0a1..03be5c2120 100644
 GRANT regress_group TO regress_group_direct_manager WITH INHERIT FALSE, ADMIN TRUE;
 GRANT regress_group_direct_manager TO regress_group_indirect_manager;
 SET SESSION AUTHORIZATION regress_group_direct_manager;
-@@ -2971,9 +2975,9 @@ DROP ROLE regress_group_direct_manager;
+@@ -2964,9 +2968,9 @@ DROP ROLE regress_group_direct_manager;
 DROP ROLE regress_group_indirect_manager;
 DROP ROLE regress_group_member;
 -- test SET and INHERIT options with object ownership changes
@@ -1292,7 +1293,7 @@ index e8c668e0a1..03be5c2120 100644
 CREATE SCHEMA regress_roleoption;
 GRANT CREATE, USAGE ON SCHEMA regress_roleoption TO PUBLIC;
 GRANT regress_roleoption_donor TO regress_roleoption_protagonist WITH INHERIT TRUE, SET FALSE;
-@@ -3002,9 +3006,9 @@ DROP ROLE regress_roleoption_protagonist;
+@@ -2995,9 +2999,9 @@ DROP ROLE regress_roleoption_protagonist;
 DROP ROLE regress_roleoption_donor;
 DROP ROLE regress_roleoption_recipient;
 -- MAINTAIN
@@ -2432,10 +2433,10 @@ index e3e3bea709..fa86ddc326 100644
 COMMENT ON CONSTRAINT the_constraint ON constraint_comments_tbl IS 'no, the comment';
 COMMENT ON CONSTRAINT the_constraint ON DOMAIN constraint_comments_dom IS 'no, another comment';
 diff --git a/src/test/regress/sql/conversion.sql b/src/test/regress/sql/conversion.sql
-index b567a1a572..4d1ac2e631 100644
+index 9a65fca91f..58431a3056 100644
 --- a/src/test/regress/sql/conversion.sql
 +++ b/src/test/regress/sql/conversion.sql
-@@ -17,7 +17,7 @@ CREATE FUNCTION test_enc_conversion(bytea, name, name, bool, validlen OUT int, r
+@@ -12,7 +12,7 @@ CREATE FUNCTION test_enc_conversion(bytea, name, name, bool, validlen OUT int, r
     AS :'regresslib', 'test_enc_conversion'
     LANGUAGE C STRICT;
 
@@ -2799,7 +2800,7 @@ index ae6841308b..47bc792e30 100644
 
 SELECT *
 diff --git a/src/test/regress/sql/database.sql b/src/test/regress/sql/database.sql
-index 46ad263478..eb05584ed5 100644
+index 0367c0e37a..a23b98c4bd 100644
 --- a/src/test/regress/sql/database.sql
 +++ b/src/test/regress/sql/database.sql
@@ -1,8 +1,6 @@
@@ -2912,7 +2913,7 @@ index aa147b14a9..370e0dd570 100644
 CREATE FOREIGN DATA WRAPPER dummy;
 COMMENT ON FOREIGN DATA WRAPPER dummy IS 'useless';
 diff --git a/src/test/regress/sql/foreign_key.sql b/src/test/regress/sql/foreign_key.sql
-index 8c4e4c7c83..e946cd2119 100644
+index 2e710e419c..89cd481a54 100644
 --- a/src/test/regress/sql/foreign_key.sql
 +++ b/src/test/regress/sql/foreign_key.sql
@@ -1435,7 +1435,7 @@ ALTER TABLE fk_partitioned_fk_6 ATTACH PARTITION fk_partitioned_pk_6 FOR VALUES
@@ -3300,7 +3301,7 @@ index bb82aa4aa2..dd8a05e24d 100644
 -- Check that the invalid secrets were re-hashed. A re-hashed secret
 -- should not contain the original salt.
 diff --git a/src/test/regress/sql/privileges.sql b/src/test/regress/sql/privileges.sql
-index b7e1cb6cdd..6e5a2217f1 100644
+index 5880bc018d..27aa952b18 100644
 --- a/src/test/regress/sql/privileges.sql
 +++ b/src/test/regress/sql/privileges.sql
@@ -24,18 +24,18 @@ RESET client_min_messages;
@@ -3362,7 +3363,7 @@ index b7e1cb6cdd..6e5a2217f1 100644
 
 ALTER GROUP regress_priv_group1 ADD USER regress_priv_user4;
 
-@@ -1160,7 +1160,7 @@ SELECT has_table_privilege('regress_priv_user1', 'atest4', 'SELECT WITH GRANT OP
+@@ -1157,7 +1157,7 @@ SELECT has_table_privilege('regress_priv_user1', 'atest4', 'SELECT WITH GRANT OP
 
 -- security-restricted operations
 \c -
@@ -3371,7 +3372,7 @@ index b7e1cb6cdd..6e5a2217f1 100644
 
 -- Check that index expressions and predicates are run as the table's owner
 
-@@ -1656,8 +1656,8 @@ DROP SCHEMA testns CASCADE;
+@@ -1653,8 +1653,8 @@ DROP SCHEMA testns CASCADE;
 -- Change owner of the schema & and rename of new schema owner
 \c -
 
@@ -3382,7 +3383,7 @@ index b7e1cb6cdd..6e5a2217f1 100644
 
 SET SESSION ROLE regress_schemauser1;
 CREATE SCHEMA testns;
-@@ -1751,7 +1751,7 @@ DROP USER regress_priv_user8; -- does not exist
+@@ -1748,7 +1748,7 @@ DROP USER regress_priv_user8; -- does not exist
 
 
 -- permissions with LOCK TABLE
@@ -3391,7 +3392,7 @@ index b7e1cb6cdd..6e5a2217f1 100644
 CREATE TABLE lock_table (a int);
 
 -- LOCK TABLE and SELECT permission
-@@ -1854,7 +1854,7 @@ DROP USER regress_locktable_user;
+@@ -1851,7 +1851,7 @@ DROP USER regress_locktable_user;
 -- switch to superuser
 \c -
 
@@ -3400,7 +3401,7 @@ index b7e1cb6cdd..6e5a2217f1 100644
 
 SELECT has_table_privilege('regress_readallstats','pg_backend_memory_contexts','SELECT'); -- no
 SELECT has_table_privilege('regress_readallstats','pg_shmem_allocations','SELECT'); -- no
-@@ -1874,10 +1874,10 @@ RESET ROLE;
+@@ -1871,10 +1871,10 @@ RESET ROLE;
 DROP ROLE regress_readallstats;
 
 -- test role grantor machinery
@@ -3415,7 +3416,7 @@ index b7e1cb6cdd..6e5a2217f1 100644
 
 GRANT regress_group TO regress_group_direct_manager WITH INHERIT FALSE, ADMIN TRUE;
 GRANT regress_group_direct_manager TO regress_group_indirect_manager;
-@@ -1899,9 +1899,9 @@ DROP ROLE regress_group_indirect_manager;
+@@ -1896,9 +1896,9 @@ DROP ROLE regress_group_indirect_manager;
 DROP ROLE regress_group_member;
 
 -- test SET and INHERIT options with object ownership changes
@@ -3428,7 +3429,7 @@ index b7e1cb6cdd..6e5a2217f1 100644
 CREATE SCHEMA regress_roleoption;
 GRANT CREATE, USAGE ON SCHEMA regress_roleoption TO PUBLIC;
 GRANT regress_roleoption_donor TO regress_roleoption_protagonist WITH INHERIT TRUE, SET FALSE;
-@@ -1929,9 +1929,9 @@ DROP ROLE regress_roleoption_donor;
+@@ -1926,9 +1926,9 @@ DROP ROLE regress_roleoption_donor;
 DROP ROLE regress_roleoption_recipient;
 
 -- MAINTAIN
--- a/compute/patches/pg_hint_plan_v16.patch
+++ b/compute/patches/pg_hint_plan_v16.patch
@@ -2,6 +2,23 @@ diff --git a/expected/ut-A.out b/expected/ut-A.out
 index da723b8..5328114 100644
 --- a/expected/ut-A.out
 +++ b/expected/ut-A.out
+@@ -9,13 +9,16 @@ SET search_path TO public;
+ ----
+ -- No.A-1-1-3
+ CREATE EXTENSION pg_hint_plan;
+LOG:  Sending request to compute_ctl: http://localhost:3081/extension_server/pg_hint_plan
+ -- No.A-1-2-3
+ DROP EXTENSION pg_hint_plan;
+ -- No.A-1-1-4
+ CREATE SCHEMA other_schema;
+ CREATE EXTENSION pg_hint_plan SCHEMA other_schema;
+LOG:  Sending request to compute_ctl: http://localhost:3081/extension_server/pg_hint_plan
+ ERROR:  extension "pg_hint_plan" must be installed in schema "hint_plan"
+ CREATE EXTENSION pg_hint_plan;
+LOG:  Sending request to compute_ctl: http://localhost:3081/extension_server/pg_hint_plan
+ DROP SCHEMA other_schema;
+ ----
+ ---- No. A-5-1 comment pattern
@@ -3175,6 +3178,7 @@ SELECT s.query, s.calls
   FROM public.pg_stat_statements s
   JOIN pg_catalog.pg_database d
@@ -10,6 +27,18 @@ index da723b8..5328114 100644
  ORDER BY 1;
                 query                 | calls 
 --------------------------------------+-------
+diff --git a/expected/ut-fdw.out b/expected/ut-fdw.out
+index d372459..6282afe 100644
+--- a/expected/ut-fdw.out
+++ b/expected/ut-fdw.out
+@@ -7,6 +7,7 @@ SET pg_hint_plan.debug_print TO on;
+ SET client_min_messages TO LOG;
+ SET pg_hint_plan.enable_hint TO on;
+ CREATE EXTENSION file_fdw;
+LOG:  Sending request to compute_ctl: http://localhost:3081/extension_server/file_fdw
+ CREATE SERVER file_server FOREIGN DATA WRAPPER file_fdw;
+ CREATE USER MAPPING FOR PUBLIC SERVER file_server;
+ CREATE FOREIGN TABLE ft1 (id int, val int) SERVER file_server OPTIONS (format 'csv', filename :'filename');
 diff --git a/sql/ut-A.sql b/sql/ut-A.sql
 index 7c7d58a..4fd1a07 100644
 --- a/sql/ut-A.sql
--- a/compute/patches/pg_hint_plan_v17.patch
+++ b/compute/patches/pg_hint_plan_v17.patch
@@ -1,3 +1,24 @@
+diff --git a/expected/ut-A.out b/expected/ut-A.out
+index e7d68a1..65a056c 100644
+--- a/expected/ut-A.out
+++ b/expected/ut-A.out
+@@ -9,13 +9,16 @@ SET search_path TO public;
+ ----
+ -- No.A-1-1-3
+ CREATE EXTENSION pg_hint_plan;
+LOG:  Sending request to compute_ctl: http://localhost:3081/extension_server/pg_hint_plan
+ -- No.A-1-2-3
+ DROP EXTENSION pg_hint_plan;
+ -- No.A-1-1-4
+ CREATE SCHEMA other_schema;
+ CREATE EXTENSION pg_hint_plan SCHEMA other_schema;
+LOG:  Sending request to compute_ctl: http://localhost:3081/extension_server/pg_hint_plan
+ ERROR:  extension "pg_hint_plan" must be installed in schema "hint_plan"
+ CREATE EXTENSION pg_hint_plan;
+LOG:  Sending request to compute_ctl: http://localhost:3081/extension_server/pg_hint_plan
+ DROP SCHEMA other_schema;
+ ----
+ ---- No. A-5-1 comment pattern
 diff --git a/expected/ut-J.out b/expected/ut-J.out
 index 2fa3c70..314e929 100644
 --- a/expected/ut-J.out
@@ -139,3 +160,15 @@ index a09bd34..0ad227c 100644
 error hint:
 
                     explain_filter                    
+diff --git a/expected/ut-fdw.out b/expected/ut-fdw.out
+index 017fa4b..98d989b 100644
+--- a/expected/ut-fdw.out
+++ b/expected/ut-fdw.out
+@@ -7,6 +7,7 @@ SET pg_hint_plan.debug_print TO on;
+ SET client_min_messages TO LOG;
+ SET pg_hint_plan.enable_hint TO on;
+ CREATE EXTENSION file_fdw;
+LOG:  Sending request to compute_ctl: http://localhost:3081/extension_server/file_fdw
+ CREATE SERVER file_server FOREIGN DATA WRAPPER file_fdw;
+ CREATE USER MAPPING FOR PUBLIC SERVER file_server;
+ CREATE FOREIGN TABLE ft1 (id int, val int) SERVER file_server OPTIONS (format 'csv', filename :'filename');
--- a/compute/patches/plv8_v3.1.10.patch
+++ b/compute/patches/plv8_v3.1.10.patch
@@ -1,6 +1,12 @@
+commit 46b38d3e46f9cd6c70d9b189dd6ff4abaa17cf5e
+Author: Alexander Bayandin <alexander@neon.tech>
+Date:   Sat Nov 30 18:29:32 2024 +0000
+
+    Fix v8 9.7.37 compilation on Debian 12
+
 diff --git a/patches/code/84cf3230a9680aac3b73c410c2b758760b6d3066.patch b/patches/code/84cf3230a9680aac3b73c410c2b758760b6d3066.patch
 new file mode 100644
-index 0000000..fae1cb3
+index 0000000..f0a5dc7
 --- /dev/null
 +++ b/patches/code/84cf3230a9680aac3b73c410c2b758760b6d3066.patch
@@ -0,0 +1,30 @@
@@ -29,21 +35,8 @@ index 0000000..fae1cb3
 +@@ -5,6 +5,7 @@
 + #ifndef V8_HEAP_CPPGC_PREFINALIZER_HANDLER_H_
 + #define V8_HEAP_CPPGC_PREFINALIZER_HANDLER_H_
-+
+ 
 ++#include <utility>
 + #include <vector>
-+
+ 
 + #include "include/cppgc/prefinalizer.h"
-diff --git a/plv8.cc b/plv8.cc
-index c1ce883..6e47e94 100644
--- a/plv8.cc
-+++ b/plv8.cc
-@@ -379,7 +379,7 @@ _PG_init(void)
- 							   NULL,
- 							   &plv8_v8_flags,
- 							   NULL,
-							   PGC_USERSET, 0,
-+							   PGC_SUSET, 0,
- #if PG_VERSION_NUM >= 90100
- 							   NULL,
- #endif
--- a/compute/patches/plv8_v3.2.3.patch
+++ b/compute/patches/plv8_v3.2.3.patch
@@ -1,13 +0,0 @@
-diff --git a/plv8.cc b/plv8.cc
-index edfa2aa..623e7f2 100644
--- a/plv8.cc
-+++ b/plv8.cc
-@@ -385,7 +385,7 @@ _PG_init(void)
-                                    NULL,
-                                    &plv8_v8_flags,
-                                    NULL,
-                                   PGC_USERSET, 0,
-+                                   PGC_SUSET, 0,
- #if PG_VERSION_NUM >= 90100
-                                    NULL,
- #endif
--- a/compute_tools/src/bin/compute_ctl.rs
+++ b/compute_tools/src/bin/compute_ctl.rs
@@ -29,12 +29,13 @@
 //! ```sh
 //! compute_ctl -D /var/db/postgres/compute \
 //!             -C 'postgresql://cloud_admin@localhost/postgres' \
-//!             -c /var/db/postgres/configs/config.json \
+//!             -S /var/db/postgres/specs/current.json \
 //!             -b /usr/local/bin/postgres \
 //!             -r http://pg-ext-s3-gateway \
 //! ```
 use std::ffi::OsString;
 use std::fs::File;
+use std::path::Path;
 use std::process::exit;
 use std::sync::mpsc;
 use std::thread;
@@ -42,10 +43,9 @@ use std::time::Duration;

 use anyhow::{Context, Result};
 use clap::Parser;
-use compute_api::responses::ComputeConfig;
-use compute_tools::compute::{
-    BUILD_TAG, ComputeNode, ComputeNodeParams, forward_termination_signal,
-};
+use compute_api::responses::ComputeCtlConfig;
+use compute_api::spec::ComputeSpec;
+use compute_tools::compute::{ComputeNode, ComputeNodeParams, forward_termination_signal};
 use compute_tools::extension_server::get_pg_version_string;
 use compute_tools::logger::*;
 use compute_tools::params::*;
@@ -57,6 +57,10 @@ use tracing::{error, info};
 use url::Url;
 use utils::failpoint_support;

+// this is an arbitrary build tag. Fine as a default / for testing purposes
+// in-case of not-set environment var
+const BUILD_TAG_DEFAULT: &str = "latest";
+
 // Compatibility hack: if the control plane specified any remote-ext-config
 // use the default value for extension storage proxy gateway.
 // Remove this once the control plane is updated to pass the gateway URL
@@ -116,21 +120,16 @@ struct Cli {
    #[arg(long)]
    pub set_disk_quota_for_fs: Option<String>,

-    // TODO(tristan957): remove alias after compatibility tests are no longer
-    // an issue
-    #[arg(short = 'c', long, alias = "spec-path")]
-    pub config: Option<OsString>,
+    #[arg(short = 's', long = "spec", group = "spec")]
+    pub spec_json: Option<String>,
+
+    #[arg(short = 'S', long, group = "spec-path")]
+    pub spec_path: Option<OsString>,

    #[arg(short = 'i', long, group = "compute-id")]
    pub compute_id: String,

-    #[arg(
-        short = 'p',
-        long,
-        conflicts_with = "config",
-        value_name = "CONTROL_PLANE_API_BASE_URL",
-        requires = "compute-id"
-    )]
+    #[arg(short = 'p', long, conflicts_with_all = ["spec", "spec-path"], value_name = "CONTROL_PLANE_API_BASE_URL")]
    pub control_plane_uri: Option<String>,
 }

@@ -139,7 +138,7 @@ fn main() -> Result<()> {

    let scenario = failpoint_support::init();

-    // For historical reasons, the main thread that processes the config and launches postgres
+    // For historical reasons, the main thread that processes the spec and launches postgres
    // is synchronous, but we always have this tokio runtime available and we "enter" it so
    // that you can use tokio::spawn() and tokio::runtime::Handle::current().block_on(...)
    // from all parts of compute_ctl.
@@ -148,14 +147,14 @@ fn main() -> Result<()> {
        .build()?;
    let _rt_guard = runtime.enter();

-    runtime.block_on(init())?;
+    let build_tag = runtime.block_on(init())?;

    // enable core dumping for all child processes
    setrlimit(Resource::CORE, rlimit::INFINITY, rlimit::INFINITY)?;

    let connstr = Url::parse(&cli.connstr).context("cannot parse connstr as a URL")?;

-    let config = get_config(&cli)?;
+    let cli_spec = try_spec_from_cli(&cli)?;

    let compute_node = ComputeNode::new(
        ComputeNodeParams {
@@ -175,8 +174,12 @@ fn main() -> Result<()> {
            cgroup: cli.cgroup,
            #[cfg(target_os = "linux")]
            vm_monitor_addr: cli.vm_monitor_addr,
+            build_tag,
+
+            live_config_allowed: cli_spec.live_config_allowed,
        },
-        config,
+        cli_spec.spec,
+        cli_spec.compute_ctl_config,
    )?;

    let exit_code = compute_node.run()?;
@@ -186,7 +189,7 @@ fn main() -> Result<()> {
    deinit_and_exit(exit_code);
 }

-async fn init() -> Result<()> {
+async fn init() -> Result<String> {
    init_tracing_and_logging(DEFAULT_LOG_LEVEL).await?;

    let mut signals = Signals::new([SIGINT, SIGTERM, SIGQUIT])?;
@@ -196,22 +199,45 @@ async fn init() -> Result<()> {
        }
    });

-    info!("compute build_tag: {}", &BUILD_TAG.to_string());
+    let build_tag = option_env!("BUILD_TAG")
+        .unwrap_or(BUILD_TAG_DEFAULT)
+        .to_string();
+    info!("build_tag: {build_tag}");

-    Ok(())
+    Ok(build_tag)
 }

-fn get_config(cli: &Cli) -> Result<ComputeConfig> {
-    // First, read the config from the path if provided
-    if let Some(ref config) = cli.config {
-        let file = File::open(config)?;
-        return Ok(serde_json::from_reader(&file)?);
+fn try_spec_from_cli(cli: &Cli) -> Result<CliSpecParams> {
+    // First, try to get cluster spec from the cli argument
+    if let Some(ref spec_json) = cli.spec_json {
+        info!("got spec from cli argument {}", spec_json);
+        return Ok(CliSpecParams {
+            spec: Some(serde_json::from_str(spec_json)?),
+            compute_ctl_config: ComputeCtlConfig::default(),
+            live_config_allowed: false,
+        });
    }

-    // If the config wasn't provided in the CLI arguments, then retrieve it from
-    // the control plane
-    match get_config_from_control_plane(cli.control_plane_uri.as_ref().unwrap(), &cli.compute_id) {
-        Ok(config) => Ok(config),
+    // Second, try to read it from the file if path is provided
+    if let Some(ref spec_path) = cli.spec_path {
+        let file = File::open(Path::new(spec_path))?;
+        return Ok(CliSpecParams {
+            spec: Some(serde_json::from_reader(file)?),
+            compute_ctl_config: ComputeCtlConfig::default(),
+            live_config_allowed: true,
+        });
+    }
+
+    if cli.control_plane_uri.is_none() {
+        panic!("must specify --control-plane-uri");
+    };
+
+    match get_spec_from_control_plane(cli.control_plane_uri.as_ref().unwrap(), &cli.compute_id) {
+        Ok(resp) => Ok(CliSpecParams {
+            spec: resp.0,
+            compute_ctl_config: resp.1,
+            live_config_allowed: true,
+        }),
        Err(e) => {
            error!(
                "cannot get response from control plane: {}\n\
@@ -223,6 +249,14 @@ fn get_config(cli: &Cli) -> Result<ComputeConfig> {
    }
 }

+struct CliSpecParams {
+    /// If a spec was provided via CLI or file, the [`ComputeSpec`]
+    spec: Option<ComputeSpec>,
+    #[allow(dead_code)]
+    compute_ctl_config: ComputeCtlConfig,
+    live_config_allowed: bool,
+}
+
 fn deinit_and_exit(exit_code: Option<i32>) -> ! {
    // Shutdown trace pipeline gracefully, so that it has a chance to send any
    // pending traces before we exit. Shutting down OTEL tracing provider may
--- a/compute_tools/src/catalog.rs
+++ b/compute_tools/src/catalog.rs
@@ -98,15 +98,13 @@ pub async fn get_database_schema(
        .kill_on_drop(true)
        .spawn()?;

-    let stdout = cmd
-        .stdout
-        .take()
-        .ok_or_else(|| std::io::Error::other("Failed to capture stdout."))?;
+    let stdout = cmd.stdout.take().ok_or_else(|| {
+        std::io::Error::new(std::io::ErrorKind::Other, "Failed to capture stdout.")
+    })?;

-    let stderr = cmd
-        .stderr
-        .take()
-        .ok_or_else(|| std::io::Error::other("Failed to capture stderr."))?;
+    let stderr = cmd.stderr.take().ok_or_else(|| {
+        std::io::Error::new(std::io::ErrorKind::Other, "Failed to capture stderr.")
+    })?;

    let mut stdout_reader = FramedRead::new(stdout, BytesCodec::new());
    let stderr_reader = BufReader::new(stderr);
@@ -130,7 +128,8 @@ pub async fn get_database_schema(
                }
            });

-            return Err(SchemaDumpError::IO(std::io::Error::other(
+            return Err(SchemaDumpError::IO(std::io::Error::new(
+                std::io::ErrorKind::Other,
                "failed to start pg_dump",
            )));
        }
--- a/compute_tools/src/compute.rs
+++ b/compute_tools/src/compute.rs
@@ -11,7 +11,7 @@ use std::{env, fs};
 use anyhow::{Context, Result};
 use chrono::{DateTime, Utc};
 use compute_api::privilege::Privilege;
-use compute_api::responses::{ComputeConfig, ComputeCtlConfig, ComputeMetrics, ComputeStatus};
+use compute_api::responses::{ComputeCtlConfig, ComputeMetrics, ComputeStatus};
 use compute_api::spec::{
    ComputeAudit, ComputeFeature, ComputeMode, ComputeSpec, ExtVersion, PgIdent,
 };
@@ -20,7 +20,6 @@ use futures::future::join_all;
 use futures::stream::FuturesUnordered;
 use nix::sys::signal::{Signal, kill};
 use nix::unistd::Pid;
-use once_cell::sync::Lazy;
 use postgres;
 use postgres::NoTls;
 use postgres::error::SqlState;
@@ -36,7 +35,6 @@ use crate::disk_quota::set_disk_quota;
 use crate::installed_extensions::get_installed_extensions;
 use crate::logger::startup_context_from_env;
 use crate::lsn_lease::launch_lsn_lease_bg_task_for_static;
-use crate::metrics::COMPUTE_CTL_UP;
 use crate::monitor::launch_monitor;
 use crate::pg_helpers::*;
 use crate::rsyslog::{
@@ -51,17 +49,6 @@ use crate::{config, extension_server, local_proxy};

 pub static SYNC_SAFEKEEPERS_PID: AtomicU32 = AtomicU32::new(0);
 pub static PG_PID: AtomicU32 = AtomicU32::new(0);
-// This is an arbitrary build tag. Fine as a default / for testing purposes
-// in-case of not-set environment var
-const BUILD_TAG_DEFAULT: &str = "latest";
-/// Build tag/version of the compute node binaries/image. It's tricky and ugly
-/// to pass it everywhere as a part of `ComputeNodeParams`, so we use a
-/// global static variable.
-pub static BUILD_TAG: Lazy<String> = Lazy::new(|| {
-    option_env!("BUILD_TAG")
-        .unwrap_or(BUILD_TAG_DEFAULT)
-        .to_string()
-});

 /// Static configuration params that don't change after startup. These mostly
 /// come from the CLI args, or are derived from them.
@@ -85,6 +72,7 @@ pub struct ComputeNodeParams {
    pub pgdata: String,
    pub pgbin: String,
    pub pgversion: String,
+    pub build_tag: String,

    /// The port that the compute's external HTTP server listens on
    pub external_http_port: u16,
@@ -93,6 +81,20 @@ pub struct ComputeNodeParams {

    /// the address of extension storage proxy gateway
    pub ext_remote_storage: Option<String>,
+
+    /// We should only allow live re- / configuration of the compute node if
+    /// it uses 'pull model', i.e. it can go to control-plane and fetch
+    /// the latest configuration. Otherwise, there could be a case:
+    /// - we start compute with some spec provided as argument
+    /// - we push new spec and it does reconfiguration
+    /// - but then something happens and compute pod / VM is destroyed,
+    ///   so k8s controller starts it again with the **old** spec
+    ///
+    /// and the same for empty computes:
+    /// - we started compute without any spec
+    /// - we push spec and it does configuration
+    /// - but then it is restarted without any spec again
+    pub live_config_allowed: bool,
 }

 /// Compute node info shared across several `compute_ctl` threads.
@@ -171,11 +173,6 @@ impl ComputeState {
        info!("Changing compute status from {} to {}", prev, status);
        self.status = status;
        state_changed.notify_all();
-
-        COMPUTE_CTL_UP.reset();
-        COMPUTE_CTL_UP
-            .with_label_values(&[&BUILD_TAG, status.to_string().as_str()])
-            .set(1);
    }

    pub fn set_failed_status(&mut self, err: anyhow::Error, state_changed: &Condvar) {
@@ -303,7 +300,11 @@ struct StartVmMonitorResult {
 }

 impl ComputeNode {
-    pub fn new(params: ComputeNodeParams, config: ComputeConfig) -> Result<Self> {
+    pub fn new(
+        params: ComputeNodeParams,
+        cli_spec: Option<ComputeSpec>,
+        compute_ctl_config: ComputeCtlConfig,
+    ) -> Result<Self> {
        let connstr = params.connstr.as_str();
        let conn_conf = postgres::config::Config::from_str(connstr)
            .context("cannot build postgres config from connstr")?;
@@ -311,8 +312,8 @@ impl ComputeNode {
            .context("cannot build tokio postgres config from connstr")?;

        let mut new_state = ComputeState::new();
-        if let Some(spec) = config.spec {
-            let pspec = ParsedSpec::try_from(spec).map_err(|msg| anyhow::anyhow!(msg))?;
+        if let Some(cli_spec) = cli_spec {
+            let pspec = ParsedSpec::try_from(cli_spec).map_err(|msg| anyhow::anyhow!(msg))?;
            new_state.pspec = Some(pspec);
        }

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

@@ -342,14 +343,6 @@ impl ComputeNode {
            this.prewarm_postgres()?;
        }

-        // Set the up metric with Empty status before starting the HTTP server.
-        // That way on the first metric scrape, an external observer will see us
-        // as 'up' and 'empty' (unless the compute was started with a spec or
-        // already configured by control plane).
-        COMPUTE_CTL_UP
-            .with_label_values(&[&BUILD_TAG, ComputeStatus::Empty.to_string().as_str()])
-            .set(1);
-
        // Launch the external HTTP server first, so that we can serve control plane
        // requests while configuration is still in progress.
        crate::http::server::Server::External {
@@ -519,14 +512,11 @@ impl ComputeNode {

        let pspec = compute_state.pspec.as_ref().expect("spec must be set");
        info!(
-            "starting compute for project {}, operation {}, tenant {}, timeline {}, project {}, branch {}, endpoint {}, features {:?}, spec.remote_extensions {:?}",
+            "starting compute for project {}, operation {}, tenant {}, timeline {}, features {:?}, spec.remote_extensions {:?}",
            pspec.spec.cluster.cluster_id.as_deref().unwrap_or("None"),
            pspec.spec.operation_uuid.as_deref().unwrap_or("None"),
            pspec.tenant_id,
            pspec.timeline_id,
-            pspec.spec.project_id.as_deref().unwrap_or("None"),
-            pspec.spec.branch_id.as_deref().unwrap_or("None"),
-            pspec.spec.endpoint_id.as_deref().unwrap_or("None"),
            pspec.spec.features,
            pspec.spec.remote_extensions,
        );
@@ -630,28 +620,31 @@ impl ComputeNode {
            });
        }

-        // Configure and start rsyslog for compliance audit logging
-        match pspec.spec.audit_log_level {
-            ComputeAudit::Hipaa | ComputeAudit::Extended | ComputeAudit::Full => {
-                let remote_endpoint =
-                    std::env::var("AUDIT_LOGGING_ENDPOINT").unwrap_or("".to_string());
-                if remote_endpoint.is_empty() {
-                    anyhow::bail!("AUDIT_LOGGING_ENDPOINT is empty");
-                }
-
-                let log_directory_path = Path::new(&self.params.pgdata).join("log");
-                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 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() {
+                anyhow::bail!("AUDIT_LOGGING_ENDPOINT is empty");
            }
-            _ => {}
+
+            let log_directory_path = Path::new(&self.params.pgdata).join("log");
+            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
-        let conf = PostgresLogsRsyslogConfig::new(pspec.spec.logs_export_host.as_deref());
-        configure_postgres_logs_export(conf)?;
+        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
        let _monitor_handle = launch_monitor(self);
@@ -1504,6 +1497,27 @@ impl ComputeNode {
        Ok::<(), anyhow::Error>(())
    }

+    /// Apply config operations that are not covered by `skip_pg_catalog_updates`
+    #[instrument(skip_all)]
+    pub fn apply_config_non_skippable(&self, compute_state: &ComputeState) -> Result<()> {
+        let conf = self.get_tokio_conn_conf(Some("compute_ctl:apply_config"));
+
+        let conf = Arc::new(conf);
+        let spec = Arc::new(
+            compute_state
+                .pspec
+                .as_ref()
+                .expect("spec must be set")
+                .spec
+                .clone(),
+        );
+
+        // Merge-apply spec & changes to PostgreSQL state.
+        self.apply_spec_sql_non_skippable(spec.clone(), conf.clone())?;
+
+        Ok::<(), anyhow::Error>(())
+    }
+
    // Wrapped this around `pg_ctl reload`, but right now we don't use
    // `pg_ctl` for start / stop.
    #[instrument(skip_all)]
@@ -1555,10 +1569,6 @@ impl ComputeNode {
            });
        }

-        // Reconfigure rsyslog for Postgres logs export
-        let conf = PostgresLogsRsyslogConfig::new(spec.logs_export_host.as_deref());
-        configure_postgres_logs_export(conf)?;
-
        // Write new config
        let pgdata_path = Path::new(&self.params.pgdata);
        config::write_postgres_conf(
@@ -1630,8 +1640,24 @@ impl ComputeNode {
                    "updated postgresql.conf to set neon.disable_logical_replication_subscribers=false"
                );
            }
+            self.pg_reload_conf()?;
+        } else {
+            // We need to run some operations even if skip_pg_catalog_updates is set
+            let pgdata_path = Path::new(&self.params.pgdata);
+            // temporarily reset max_cluster_size in config
+            // to avoid the possibility of hitting the limit, while we are applying config:
+            // creating new extensions, roles, etc...
+            config::with_compute_ctl_tmp_override(pgdata_path, "neon.max_cluster_size=-1", || {
+                self.pg_reload_conf()?;
+
+                self.apply_config_non_skippable(compute_state)?;
+
+                Ok(())
+            })?;
+
            self.pg_reload_conf()?;
        }
+
        self.post_apply_config()?;

        Ok(())
@@ -2043,8 +2069,12 @@ LIMIT 100",

        let mut download_tasks = Vec::new();
        for library in &libs_vec {
-            let (ext_name, ext_path) =
-                remote_extensions.get_ext(library, true, &BUILD_TAG, &self.params.pgversion)?;
+            let (ext_name, ext_path) = remote_extensions.get_ext(
+                library,
+                true,
+                &self.params.build_tag,
+                &self.params.pgversion,
+            )?;
            download_tasks.push(self.download_extension(ext_name, ext_path));
        }
        let results = join_all(download_tasks).await;
--- a/compute_tools/src/config.rs
+++ b/compute_tools/src/config.rs
@@ -7,7 +7,7 @@ use std::io::prelude::*;
 use std::path::Path;

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

 use crate::pg_helpers::{
    GenericOptionExt, GenericOptionsSearch, PgOptionsSerialize, escape_conf_value,
@@ -89,15 +89,6 @@ pub fn write_postgres_conf(
            escape_conf_value(&s.to_string())
        )?;
    }
-    if let Some(s) = &spec.project_id {
-        writeln!(file, "neon.project_id={}", escape_conf_value(s))?;
-    }
-    if let Some(s) = &spec.branch_id {
-        writeln!(file, "neon.branch_id={}", escape_conf_value(s))?;
-    }
-    if let Some(s) = &spec.endpoint_id {
-        writeln!(file, "neon.endpoint_id={}", escape_conf_value(s))?;
-    }

    // tls
    if let Some(tls_config) = tls_config {
@@ -178,7 +169,7 @@ pub fn write_postgres_conf(
    // and don't allow the user or the control plane admin to change them.
    match spec.audit_log_level {
        ComputeAudit::Disabled => {}
-        ComputeAudit::Log | ComputeAudit::Base => {
+        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
@@ -202,20 +193,16 @@ pub fn write_postgres_conf(
            }
            writeln!(file, "# Managed by compute_ctl base audit settings: end")?;
        }
-        ComputeAudit::Hipaa | ComputeAudit::Extended | ComputeAudit::Full => {
+        ComputeAudit::Hipaa => {
            writeln!(
                file,
                "# Managed by compute_ctl compliance audit settings: begin"
            )?;
-            // Enable logging of parameters.
-            // This is very verbose and may contain sensitive data.
-            if spec.audit_log_level == ComputeAudit::Full {
-                writeln!(file, "pgaudit.log_parameter=on")?;
-                writeln!(file, "pgaudit.log='all'")?;
-            } else {
-                writeln!(file, "pgaudit.log_parameter=off")?;
-                writeln!(file, "pgaudit.log='all, -misc'")?;
-            }
+            // 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")?;
@@ -268,7 +255,7 @@ pub fn write_postgres_conf(

    // We need Postgres to send logs to rsyslog so that we can forward them
    // further to customers' log aggregation systems.
-    if spec.logs_export_host.is_some() {
+    if spec.features.contains(&ComputeFeature::PostgresLogsExport) {
        writeln!(file, "log_destination='stderr,syslog'")?;
    }

--- a/compute_tools/src/http/middleware/authorize.rs
+++ b/compute_tools/src/http/middleware/authorize.rs
@@ -6,15 +6,20 @@ use axum_extra::{
    TypedHeader,
    headers::{Authorization, authorization::Bearer},
 };
-use compute_api::requests::ComputeClaims;
 use futures::future::BoxFuture;
 use http::{Request, Response, StatusCode};
 use jsonwebtoken::{Algorithm, DecodingKey, TokenData, Validation, jwk::JwkSet};
+use serde::Deserialize;
 use tower_http::auth::AsyncAuthorizeRequest;
-use tracing::{debug, warn};
+use tracing::warn;

 use crate::http::{JsonResponse, extract::RequestId};

+#[derive(Clone, Debug, Deserialize)]
+pub(in crate::http) struct Claims {
+    compute_id: String,
+}
+
 #[derive(Clone, Debug)]
 pub(in crate::http) struct Authorize {
    compute_id: String,
@@ -54,12 +59,9 @@ impl AsyncAuthorizeRequest<Body> for Authorize {
        Box::pin(async move {
            let request_id = request.extract_parts::<RequestId>().await.unwrap();

-            // TODO: Remove this stanza after teaching neon_local and the
-            // regression tests to use a JWT + JWKS.
-            //
-            // https://github.com/neondatabase/neon/issues/11316
-            if cfg!(feature = "testing") {
-                warn!(%request_id, "Skipping compute_ctl authorization check");
+            // TODO: Remove this check after a successful rollout
+            if jwks.keys.is_empty() {
+                warn!(%request_id, "Authorization has not been configured");

                return Ok(request);
            }
@@ -92,7 +94,7 @@ impl AsyncAuthorizeRequest<Body> for Authorize {
            if data.claims.compute_id != compute_id {
                return Err(JsonResponse::error(
                    StatusCode::UNAUTHORIZED,
-                    "invalid compute ID in authorization token claims",
+                    "invalid claims in authorization token",
                ));
            }

@@ -107,19 +109,15 @@ impl AsyncAuthorizeRequest<Body> for Authorize {

 impl Authorize {
    /// Verify the token using the JSON Web Key set and return the token data.
-    fn verify(
-        jwks: &JwkSet,
-        token: &str,
-        validation: &Validation,
-    ) -> Result<TokenData<ComputeClaims>> {
-        debug!("verifying token {}", token);
+    fn verify(jwks: &JwkSet, token: &str, validation: &Validation) -> Result<TokenData<Claims>> {
+        debug_assert!(!jwks.keys.is_empty());

        for jwk in jwks.keys.iter() {
            let decoding_key = match DecodingKey::from_jwk(jwk) {
                Ok(key) => key,
                Err(e) => {
                    warn!(
-                        "failed to construct decoding key from {}: {}",
+                        "Failed to construct decoding key from {}: {}",
                        jwk.common.key_id.as_ref().unwrap(),
                        e
                    );
@@ -128,11 +126,11 @@ impl Authorize {
                }
            };

-            match jsonwebtoken::decode::<ComputeClaims>(token, &decoding_key, validation) {
+            match jsonwebtoken::decode::<Claims>(token, &decoding_key, validation) {
                Ok(data) => return Ok(data),
                Err(e) => {
                    warn!(
-                        "failed to decode authorization token using {}: {}",
+                        "Failed to decode authorization token using {}: {}",
                        jwk.common.key_id.as_ref().unwrap(),
                        e
                    );
@@ -142,6 +140,6 @@ impl Authorize {
            }
        }

-        Err(anyhow!("failed to verify authorization token"))
+        Err(anyhow!("Failed to verify authorization token"))
    }
 }
--- 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
@@ -22,6 +25,13 @@ pub(in crate::http) async fn configure(
    State(compute): State<Arc<ComputeNode>>,
    request: Json<ConfigurationRequest>,
 ) -> Response {
+    if !compute.params.live_config_allowed {
+        return JsonResponse::error(
+            StatusCode::PRECONDITION_FAILED,
+            "live configuration is not allowed for this compute node".to_string(),
+        );
+    }
+
    let pspec = match ParsedSpec::try_from(request.spec.clone()) {
        Ok(p) => p,
        Err(e) => return JsonResponse::error(StatusCode::BAD_REQUEST, e),
@@ -85,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/routes/extension_server.rs
+++ b/compute_tools/src/http/routes/extension_server.rs
@@ -5,7 +5,7 @@ use axum::response::{IntoResponse, Response};
 use http::StatusCode;
 use serde::Deserialize;

-use crate::compute::{BUILD_TAG, ComputeNode};
+use crate::compute::ComputeNode;
 use crate::http::JsonResponse;
 use crate::http::extract::{Path, Query};

@@ -47,7 +47,7 @@ pub(in crate::http) async fn download_extension(
        remote_extensions.get_ext(
            &filename,
            ext_server_params.is_library,
-            &BUILD_TAG,
+            &compute.params.build_tag,
            &compute.params.pgversion,
        )
    };
--- a/compute_tools/src/http/server.rs
+++ b/compute_tools/src/http/server.rs
@@ -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))
--- a/compute_tools/src/metrics.rs
+++ b/compute_tools/src/metrics.rs
@@ -1,8 +1,7 @@
 use metrics::core::{AtomicF64, Collector, GenericGauge};
 use metrics::proto::MetricFamily;
 use metrics::{
-    IntCounterVec, IntGaugeVec, UIntGaugeVec, register_gauge, register_int_counter_vec,
-    register_int_gauge_vec, register_uint_gauge_vec,
+    IntCounterVec, UIntGaugeVec, register_gauge, register_int_counter_vec, register_uint_gauge_vec,
 };
 use once_cell::sync::Lazy;

@@ -19,13 +18,13 @@ pub(crate) static INSTALLED_EXTENSIONS: Lazy<UIntGaugeVec> = Lazy::new(|| {
 // but for all our APIs we defined a 'slug'/method/operationId in the OpenAPI spec.
 // And it's fair to call it a 'RPC' (Remote Procedure Call).
 pub enum CPlaneRequestRPC {
-    GetConfig,
+    GetSpec,
 }

 impl CPlaneRequestRPC {
    pub fn as_str(&self) -> &str {
        match self {
-            CPlaneRequestRPC::GetConfig => "GetConfig",
+            CPlaneRequestRPC::GetSpec => "GetSpec",
        }
    }
 }
@@ -71,19 +70,8 @@ pub(crate) static AUDIT_LOG_DIR_SIZE: Lazy<GenericGauge<AtomicF64>> = Lazy::new(
    .expect("failed to define a metric")
 });

-// Report that `compute_ctl` is up and what's the current compute status.
-pub(crate) static COMPUTE_CTL_UP: Lazy<IntGaugeVec> = Lazy::new(|| {
-    register_int_gauge_vec!(
-        "compute_ctl_up",
-        "Whether compute_ctl is running",
-        &["build_tag", "status"]
-    )
-    .expect("failed to define a metric")
-});
-
 pub fn collect() -> Vec<MetricFamily> {
-    let mut metrics = COMPUTE_CTL_UP.collect();
-    metrics.extend(INSTALLED_EXTENSIONS.collect());
+    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());
--- a/compute_tools/src/rsyslog.rs
+++ b/compute_tools/src/rsyslog.rs
@@ -119,9 +119,16 @@ impl<'a> PostgresLogsRsyslogConfig<'a> {
        };
        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
+        )
+    }
 }

-/// Writes rsyslogd configuration for Postgres logs export and restarts rsyslog.
 pub fn configure_postgres_logs_export(conf: PostgresLogsRsyslogConfig) -> Result<()> {
    let new_config = conf.build()?;
    let current_config = PostgresLogsRsyslogConfig::current_config()?;
@@ -254,5 +261,16 @@ mod tests {
            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
@@ -3,8 +3,9 @@ use std::path::Path;

 use anyhow::{Result, anyhow, bail};
 use compute_api::responses::{
-    ComputeConfig, ControlPlaneComputeStatus, ControlPlaneConfigResponse,
+    ComputeCtlConfig, ControlPlaneComputeStatus, ControlPlaneSpecResponse,
 };
+use compute_api::spec::ComputeSpec;
 use reqwest::StatusCode;
 use tokio_postgres::Client;
 use tracing::{error, info, instrument};
@@ -20,7 +21,7 @@ use crate::params::PG_HBA_ALL_MD5;
 fn do_control_plane_request(
    uri: &str,
    jwt: &str,
-) -> Result<ControlPlaneConfigResponse, (bool, String, String)> {
+) -> Result<ControlPlaneSpecResponse, (bool, String, String)> {
    let resp = reqwest::blocking::Client::new()
        .get(uri)
        .header("Authorization", format!("Bearer {}", jwt))
@@ -28,14 +29,14 @@ fn do_control_plane_request(
        .map_err(|e| {
            (
                true,
-                format!("could not perform request to control plane: {:?}", e),
+                format!("could not perform spec request to control plane: {:?}", e),
                UNKNOWN_HTTP_STATUS.to_string(),
            )
        })?;

    let status = resp.status();
    match status {
-        StatusCode::OK => match resp.json::<ControlPlaneConfigResponse>() {
+        StatusCode::OK => match resp.json::<ControlPlaneSpecResponse>() {
            Ok(spec_resp) => Ok(spec_resp),
            Err(e) => Err((
                true,
@@ -68,35 +69,40 @@ fn do_control_plane_request(
    }
 }

-/// Request config from the control-plane by compute_id. If
-/// `NEON_CONTROL_PLANE_TOKEN` env variable is set, it will be used for
-/// authorization.
-pub fn get_config_from_control_plane(base_uri: &str, compute_id: &str) -> Result<ComputeConfig> {
+/// Request spec from the control-plane by compute_id. If `NEON_CONTROL_PLANE_TOKEN`
+/// env variable is set, it will be used for authorization.
+pub fn get_spec_from_control_plane(
+    base_uri: &str,
+    compute_id: &str,
+) -> Result<(Option<ComputeSpec>, ComputeCtlConfig)> {
    let cp_uri = format!("{base_uri}/compute/api/v2/computes/{compute_id}/spec");
-    let jwt: String = std::env::var("NEON_CONTROL_PLANE_TOKEN").unwrap_or_default();
+    let jwt: String = match std::env::var("NEON_CONTROL_PLANE_TOKEN") {
+        Ok(v) => v,
+        Err(_) => "".to_string(),
+    };
    let mut attempt = 1;

-    info!("getting config from control plane: {}", cp_uri);
+    info!("getting spec from control plane: {}", cp_uri);

    // Do 3 attempts to get spec from the control plane using the following logic:
    // - network error -> then retry
    // - compute id is unknown or any other error -> bail out
    // - no spec for compute yet (Empty state) -> return Ok(None)
-    // - got config -> return Ok(Some(config))
+    // - got spec -> return Ok(Some(spec))
    while attempt < 4 {
        let result = match do_control_plane_request(&cp_uri, &jwt) {
-            Ok(config_resp) => {
+            Ok(spec_resp) => {
                CPLANE_REQUESTS_TOTAL
                    .with_label_values(&[
-                        CPlaneRequestRPC::GetConfig.as_str(),
+                        CPlaneRequestRPC::GetSpec.as_str(),
                        &StatusCode::OK.to_string(),
                    ])
                    .inc();
-                match config_resp.status {
-                    ControlPlaneComputeStatus::Empty => Ok(config_resp.into()),
+                match spec_resp.status {
+                    ControlPlaneComputeStatus::Empty => Ok((None, spec_resp.compute_ctl_config)),
                    ControlPlaneComputeStatus::Attached => {
-                        if config_resp.spec.is_some() {
-                            Ok(config_resp.into())
+                        if let Some(spec) = spec_resp.spec {
+                            Ok((Some(spec), spec_resp.compute_ctl_config))
                        } else {
                            bail!("compute is attached, but spec is empty")
                        }
@@ -105,7 +111,7 @@ pub fn get_config_from_control_plane(base_uri: &str, compute_id: &str) -> Result
            }
            Err((retry, msg, status)) => {
                CPLANE_REQUESTS_TOTAL
-                    .with_label_values(&[CPlaneRequestRPC::GetConfig.as_str(), &status])
+                    .with_label_values(&[CPlaneRequestRPC::GetSpec.as_str(), &status])
                    .inc();
                if retry {
                    Err(anyhow!(msg))
@@ -116,7 +122,7 @@ pub fn get_config_from_control_plane(base_uri: &str, compute_id: &str) -> Result
        };

        if let Err(e) = &result {
-            error!("attempt {} to get config failed with: {}", attempt, e);
+            error!("attempt {} to get spec failed with: {}", attempt, e);
        } else {
            return result;
        }
@@ -127,13 +133,13 @@ pub fn get_config_from_control_plane(base_uri: &str, compute_id: &str) -> Result

    // All attempts failed, return error.
    Err(anyhow::anyhow!(
-        "Exhausted all attempts to retrieve the config from the control plane"
+        "Exhausted all attempts to retrieve the spec from the control plane"
    ))
 }

 /// Check `pg_hba.conf` and update if needed to allow external connections.
 pub fn update_pg_hba(pgdata_path: &Path) -> Result<()> {
-    // XXX: consider making it a part of config.json
+    // XXX: consider making it a part of spec.json
    let pghba_path = pgdata_path.join("pg_hba.conf");

    if config::line_in_file(&pghba_path, PG_HBA_ALL_MD5)? {
@@ -147,7 +153,7 @@ pub fn update_pg_hba(pgdata_path: &Path) -> Result<()> {

 /// Create a standby.signal file
 pub fn add_standby_signal(pgdata_path: &Path) -> Result<()> {
-    // XXX: consider making it a part of config.json
+    // XXX: consider making it a part of spec.json
    let signalfile = pgdata_path.join("standby.signal");

    if !signalfile.exists() {
--- a/compute_tools/src/spec_apply.rs
+++ b/compute_tools/src/spec_apply.rs
@@ -278,12 +278,81 @@ impl ComputeNode {
            // so that all config operations are audit logged.
            match spec.audit_log_level
            {
-                ComputeAudit::Hipaa | ComputeAudit::Extended | ComputeAudit::Full => {
+                ComputeAudit::Hipaa => {
                    phases.push(CreatePgauditExtension);
                    phases.push(CreatePgauditlogtofileExtension);
                    phases.push(DisablePostgresDBPgAudit);
                }
-                ComputeAudit::Log | ComputeAudit::Base => {
+                ComputeAudit::Log => {
+                    phases.push(CreatePgauditExtension);
+                    phases.push(DisablePostgresDBPgAudit);
+                }
+                ComputeAudit::Disabled => {}
+            }
+
+            for phase in phases {
+                debug!("Applying phase {:?}", &phase);
+                apply_operations(
+                    spec.clone(),
+                    ctx.clone(),
+                    jwks_roles.clone(),
+                    phase,
+                    || async { Ok(&client) },
+                )
+                .await?;
+            }
+
+            Ok::<(), anyhow::Error>(())
+        })?;
+
+        Ok(())
+    }
+
+    // Similar to apply_spec_sql, but for the simplified set of operations
+    // that we perform even when `pg_skip_catalog_updates` is set.
+    //
+    // Keep the list of operations as small as possible,
+    // as it will be run on every spec change and affect compute start time.
+    pub fn apply_spec_sql_non_skippable(
+        &self,
+        spec: Arc<ComputeSpec>,
+        conf: Arc<tokio_postgres::Config>,
+    ) -> Result<()> {
+        info!("Applying non_skippable config",);
+        debug!("Config: {:?}", spec);
+
+        let rt = tokio::runtime::Handle::current();
+        rt.block_on(async {
+            let client = Self::get_maintenance_client(&conf).await?;
+            let spec = spec.clone();
+
+            let jwks_roles = Arc::new(
+                spec.as_ref()
+                    .local_proxy_config
+                    .iter()
+                    .flat_map(|it| &it.jwks)
+                    .flatten()
+                    .flat_map(|setting| &setting.role_names)
+                    .cloned()
+                    .collect::<HashSet<_>>(),
+            );
+
+            // NOTE: Here we assume that operations below don't use ctx
+            // TODO: refactor apply_operations() to accept ctx as option.
+            let ctx = Arc::new(tokio::sync::RwLock::new(MutableApplyContext {
+                roles: HashMap::new(),
+                dbs: HashMap::new(),
+            }));
+
+            let mut phases = vec![];
+
+            match spec.audit_log_level {
+                ComputeAudit::Hipaa => {
+                    phases.push(CreatePgauditExtension);
+                    phases.push(CreatePgauditlogtofileExtension);
+                    phases.push(DisablePostgresDBPgAudit);
+                }
+                ComputeAudit::Log => {
                    phases.push(CreatePgauditExtension);
                    phases.push(DisablePostgresDBPgAudit);
                }
@@ -419,7 +488,7 @@ impl ComputeNode {
                .iter()
                .filter_map(|val| val.parse::<usize>().ok())
                .map(|val| if val > 1 { val - 1 } else { 1 })
-                .next_back()
+                .last()
                .unwrap_or(3)
        }
    }
--- a/control_plane/src/bin/neon_local.rs
+++ b/control_plane/src/bin/neon_local.rs
@@ -20,10 +20,8 @@ use compute_api::spec::ComputeMode;
 use control_plane::endpoint::ComputeControlPlane;
 use control_plane::local_env::{
    InitForceMode, LocalEnv, NeonBroker, NeonLocalInitConf, NeonLocalInitPageserverConf,
-    ObjectStorageConf, SafekeeperConf,
+    SafekeeperConf,
 };
-use control_plane::object_storage::OBJECT_STORAGE_DEFAULT_PORT;
-use control_plane::object_storage::ObjectStorage;
 use control_plane::pageserver::PageServerNode;
 use control_plane::safekeeper::SafekeeperNode;
 use control_plane::storage_controller::{
@@ -41,7 +39,7 @@ use pageserver_api::controller_api::{
 use pageserver_api::models::{
    ShardParameters, TenantConfigRequest, TimelineCreateRequest, TimelineInfo,
 };
-use pageserver_api::shard::{DEFAULT_STRIPE_SIZE, ShardCount, ShardStripeSize, TenantShardId};
+use pageserver_api::shard::{ShardCount, ShardStripeSize, TenantShardId};
 use postgres_backend::AuthType;
 use postgres_connection::parse_host_port;
 use safekeeper_api::membership::SafekeeperGeneration;
@@ -93,8 +91,6 @@ enum NeonLocalCmd {
    #[command(subcommand)]
    Safekeeper(SafekeeperCmd),
    #[command(subcommand)]
-    ObjectStorage(ObjectStorageCmd),
-    #[command(subcommand)]
    Endpoint(EndpointCmd),
    #[command(subcommand)]
    Mappings(MappingsCmd),
@@ -458,32 +454,6 @@ enum SafekeeperCmd {
    Restart(SafekeeperRestartCmdArgs),
 }

-#[derive(clap::Subcommand)]
-#[clap(about = "Manage object storage")]
-enum ObjectStorageCmd {
-    Start(ObjectStorageStartCmd),
-    Stop(ObjectStorageStopCmd),
-}
-
-#[derive(clap::Args)]
-#[clap(about = "Start object storage")]
-struct ObjectStorageStartCmd {
-    #[clap(short = 't', long, help = "timeout until we fail the command")]
-    #[arg(default_value = "10s")]
-    start_timeout: humantime::Duration,
-}
-
-#[derive(clap::Args)]
-#[clap(about = "Stop object storage")]
-struct ObjectStorageStopCmd {
-    #[arg(value_enum, default_value = "fast")]
-    #[clap(
-        short = 'm',
-        help = "If 'immediate', don't flush repository data at shutdown"
-    )]
-    stop_mode: StopMode,
-}
-
 #[derive(clap::Args)]
 #[clap(about = "Start local safekeeper")]
 struct SafekeeperStartCmdArgs {
@@ -789,7 +759,6 @@ fn main() -> Result<()> {
            }
            NeonLocalCmd::StorageBroker(subcmd) => rt.block_on(handle_storage_broker(&subcmd, env)),
            NeonLocalCmd::Safekeeper(subcmd) => rt.block_on(handle_safekeeper(&subcmd, env)),
-            NeonLocalCmd::ObjectStorage(subcmd) => rt.block_on(handle_object_storage(&subcmd, env)),
            NeonLocalCmd::Endpoint(subcmd) => rt.block_on(handle_endpoint(&subcmd, env)),
            NeonLocalCmd::Mappings(subcmd) => handle_mappings(&subcmd, env),
        };
@@ -1006,9 +975,6 @@ fn handle_init(args: &InitCmdArgs) -> anyhow::Result<LocalEnv> {
                    }
                })
                .collect(),
-            object_storage: ObjectStorageConf {
-                port: OBJECT_STORAGE_DEFAULT_PORT,
-            },
            pg_distrib_dir: None,
            neon_distrib_dir: None,
            default_tenant_id: TenantId::from_array(std::array::from_fn(|_| 0)),
@@ -1117,7 +1083,7 @@ async fn handle_tenant(subcmd: &TenantCmd, env: &mut local_env::LocalEnv) -> any
                        stripe_size: args
                            .shard_stripe_size
                            .map(ShardStripeSize)
-                            .unwrap_or(DEFAULT_STRIPE_SIZE),
+                            .unwrap_or(ShardParameters::DEFAULT_STRIPE_SIZE),
                    },
                    placement_policy: args.placement_policy.clone(),
                    config: tenant_conf,
@@ -1430,7 +1396,7 @@ async fn handle_endpoint(subcmd: &EndpointCmd, env: &local_env::LocalEnv) -> Res
                    vec![(parsed.0, parsed.1.unwrap_or(5432))],
                    // If caller is telling us what pageserver to use, this is not a tenant which is
                    // full managed by storage controller, therefore not sharded.
-                    DEFAULT_STRIPE_SIZE,
+                    ShardParameters::DEFAULT_STRIPE_SIZE,
                )
            } else {
                // Look up the currently attached location of the tenant, and its striping metadata,
@@ -1717,41 +1683,6 @@ async fn handle_safekeeper(subcmd: &SafekeeperCmd, env: &local_env::LocalEnv) ->
    Ok(())
 }

-async fn handle_object_storage(subcmd: &ObjectStorageCmd, env: &local_env::LocalEnv) -> Result<()> {
-    use ObjectStorageCmd::*;
-    let storage = ObjectStorage::from_env(env);
-
-    // In tests like test_forward_compatibility or test_graceful_cluster_restart
-    // old neon binaries (without object_storage) are present
-    if !storage.bin.exists() {
-        eprintln!(
-            "{} binary not found. Ignore if this is a compatibility test",
-            storage.bin
-        );
-        return Ok(());
-    }
-
-    match subcmd {
-        Start(ObjectStorageStartCmd { start_timeout }) => {
-            if let Err(e) = storage.start(start_timeout).await {
-                eprintln!("object_storage start failed: {e}");
-                exit(1);
-            }
-        }
-        Stop(ObjectStorageStopCmd { stop_mode }) => {
-            let immediate = match stop_mode {
-                StopMode::Fast => false,
-                StopMode::Immediate => true,
-            };
-            if let Err(e) = storage.stop(immediate) {
-                eprintln!("proxy stop failed: {e}");
-                exit(1);
-            }
-        }
-    };
-    Ok(())
-}
-
 async fn handle_storage_broker(subcmd: &StorageBrokerCmd, env: &local_env::LocalEnv) -> Result<()> {
    match subcmd {
        StorageBrokerCmd::Start(args) => {
@@ -1846,13 +1777,6 @@ async fn handle_start_all_impl(
                    .map_err(|e| e.context(format!("start safekeeper {}", safekeeper.id)))
            });
        }
-
-        js.spawn(async move {
-            ObjectStorage::from_env(env)
-                .start(&retry_timeout)
-                .await
-                .map_err(|e| e.context("start object_storage"))
-        });
    })();

    let mut errors = Vec::new();
@@ -1950,11 +1874,6 @@ async fn try_stop_all(env: &local_env::LocalEnv, immediate: bool) {
        }
    }

-    let storage = ObjectStorage::from_env(env);
-    if let Err(e) = storage.stop(immediate) {
-        eprintln!("object_storage stop failed: {:#}", e);
-    }
-
    for ps_conf in &env.pageservers {
        let pageserver = PageServerNode::from_env(env, ps_conf);
        if let Err(e) = pageserver.stop(immediate) {
--- a/control_plane/src/endpoint.rs
+++ b/control_plane/src/endpoint.rs
@@ -29,7 +29,7 @@
 //!     compute.log               - log output of `compute_ctl` and `postgres`
 //!     endpoint.json             - serialized `EndpointConf` struct
 //!     postgresql.conf           - postgresql settings
-//!     config.json                 - passed to `compute_ctl`
+//!     spec.json                 - passed to `compute_ctl`
 //!     pgdata/
 //!         postgresql.conf       - copy of postgresql.conf created by `compute_ctl`
 //!         zenith.signal
@@ -46,9 +46,7 @@ use std::time::{Duration, Instant, SystemTime, UNIX_EPOCH};

 use anyhow::{Context, Result, anyhow, bail};
 use compute_api::requests::ConfigurationRequest;
-use compute_api::responses::{
-    ComputeConfig, ComputeCtlConfig, ComputeStatus, ComputeStatusResponse,
-};
+use compute_api::responses::{ComputeCtlConfig, ComputeStatus, ComputeStatusResponse};
 use compute_api::spec::{
    Cluster, ComputeAudit, ComputeFeature, ComputeMode, ComputeSpec, Database, PgIdent,
    RemoteExtSpec, Role,
@@ -621,101 +619,86 @@ impl Endpoint {
            remote_extensions = None;
        };

-        // Create config file
-        let config = {
-            let mut spec = ComputeSpec {
-                skip_pg_catalog_updates: self.skip_pg_catalog_updates,
-                format_version: 1.0,
-                operation_uuid: None,
-                features: self.features.clone(),
-                swap_size_bytes: None,
-                disk_quota_bytes: None,
-                disable_lfc_resizing: None,
-                cluster: Cluster {
-                    cluster_id: None, // project ID: not used
-                    name: None,       // project name: not used
-                    state: None,
-                    roles: if create_test_user {
-                        vec![Role {
-                            name: PgIdent::from_str("test").unwrap(),
-                            encrypted_password: None,
-                            options: None,
-                        }]
-                    } else {
-                        Vec::new()
-                    },
-                    databases: if create_test_user {
-                        vec![Database {
-                            name: PgIdent::from_str("neondb").unwrap(),
-                            owner: PgIdent::from_str("test").unwrap(),
-                            options: None,
-                            restrict_conn: false,
-                            invalid: false,
-                        }]
-                    } else {
-                        Vec::new()
-                    },
-                    settings: None,
-                    postgresql_conf: Some(postgresql_conf.clone()),
-                },
-                delta_operations: None,
-                tenant_id: Some(self.tenant_id),
-                timeline_id: Some(self.timeline_id),
-                project_id: None,
-                branch_id: None,
-                endpoint_id: Some(self.endpoint_id.clone()),
-                mode: self.mode,
-                pageserver_connstring: Some(pageserver_connstring),
-                safekeepers_generation: safekeepers_generation.map(|g| g.into_inner()),
-                safekeeper_connstrings,
-                storage_auth_token: auth_token.clone(),
-                remote_extensions,
-                pgbouncer_settings: None,
-                shard_stripe_size: Some(shard_stripe_size),
-                local_proxy_config: None,
-                reconfigure_concurrency: self.reconfigure_concurrency,
-                drop_subscriptions_before_start: self.drop_subscriptions_before_start,
-                audit_log_level: ComputeAudit::Disabled,
-                logs_export_host: None::<String>,
-            };
-
-            // this strange code is needed to support respec() in tests
-            if self.cluster.is_some() {
-                debug!("Cluster is already set in the endpoint spec, using it");
-                spec.cluster = self.cluster.clone().unwrap();
-
-                debug!("spec.cluster {:?}", spec.cluster);
-
-                // fill missing fields again
-                if create_test_user {
-                    spec.cluster.roles.push(Role {
+        // Create spec file
+        let mut spec = ComputeSpec {
+            skip_pg_catalog_updates: self.skip_pg_catalog_updates,
+            format_version: 1.0,
+            operation_uuid: None,
+            features: self.features.clone(),
+            swap_size_bytes: None,
+            disk_quota_bytes: None,
+            disable_lfc_resizing: None,
+            cluster: Cluster {
+                cluster_id: None, // project ID: not used
+                name: None,       // project name: not used
+                state: None,
+                roles: if create_test_user {
+                    vec![Role {
                        name: PgIdent::from_str("test").unwrap(),
                        encrypted_password: None,
                        options: None,
-                    });
-                    spec.cluster.databases.push(Database {
+                    }]
+                } else {
+                    Vec::new()
+                },
+                databases: if create_test_user {
+                    vec![Database {
                        name: PgIdent::from_str("neondb").unwrap(),
                        owner: PgIdent::from_str("test").unwrap(),
                        options: None,
                        restrict_conn: false,
                        invalid: false,
-                    });
-                }
-                spec.cluster.postgresql_conf = Some(postgresql_conf);
-            }
-
-            ComputeConfig {
-                spec: Some(spec),
-                compute_ctl_config: ComputeCtlConfig::default(),
-            }
+                    }]
+                } else {
+                    Vec::new()
+                },
+                settings: None,
+                postgresql_conf: Some(postgresql_conf.clone()),
+            },
+            delta_operations: None,
+            tenant_id: Some(self.tenant_id),
+            timeline_id: Some(self.timeline_id),
+            mode: self.mode,
+            pageserver_connstring: Some(pageserver_connstring),
+            safekeepers_generation: safekeepers_generation.map(|g| g.into_inner()),
+            safekeeper_connstrings,
+            storage_auth_token: auth_token.clone(),
+            remote_extensions,
+            pgbouncer_settings: None,
+            shard_stripe_size: Some(shard_stripe_size),
+            local_proxy_config: None,
+            reconfigure_concurrency: self.reconfigure_concurrency,
+            drop_subscriptions_before_start: self.drop_subscriptions_before_start,
+            audit_log_level: ComputeAudit::Disabled,
        };

-        // TODO(tristan957): Remove the write to spec.json after compatibility
-        // tests work themselves out
+        // this strange code is needed to support respec() in tests
+        if self.cluster.is_some() {
+            debug!("Cluster is already set in the endpoint spec, using it");
+            spec.cluster = self.cluster.clone().unwrap();
+
+            debug!("spec.cluster {:?}", spec.cluster);
+
+            // fill missing fields again
+            if create_test_user {
+                spec.cluster.roles.push(Role {
+                    name: PgIdent::from_str("test").unwrap(),
+                    encrypted_password: None,
+                    options: None,
+                });
+                spec.cluster.databases.push(Database {
+                    name: PgIdent::from_str("neondb").unwrap(),
+                    owner: PgIdent::from_str("test").unwrap(),
+                    options: None,
+                    restrict_conn: false,
+                    invalid: false,
+                });
+            }
+            spec.cluster.postgresql_conf = Some(postgresql_conf);
+        }
+
        let spec_path = self.endpoint_path().join("spec.json");
-        std::fs::write(spec_path, serde_json::to_string_pretty(&config.spec)?)?;
-        let config_path = self.endpoint_path().join("config.json");
-        std::fs::write(config_path, serde_json::to_string_pretty(&config)?)?;
+        std::fs::write(spec_path, serde_json::to_string_pretty(&spec)?)?;

        // Open log file. We'll redirect the stdout and stderr of `compute_ctl` to it.
        let logfile = std::fs::OpenOptions::new()
@@ -723,16 +706,6 @@ impl Endpoint {
            .append(true)
            .open(self.endpoint_path().join("compute.log"))?;

-        // TODO(tristan957): Remove when compatibility tests are no longer an
-        // issue
-        let old_compute_ctl = {
-            let mut cmd = Command::new(self.env.neon_distrib_dir.join("compute_ctl"));
-            let help_output = cmd.arg("--help").output()?;
-            let help_output = String::from_utf8_lossy(&help_output.stdout);
-
-            !help_output.contains("--config")
-        };
-
        // Launch compute_ctl
        let conn_str = self.connstr("cloud_admin", "postgres");
        println!("Starting postgres node at '{}'", conn_str);
@@ -751,18 +724,9 @@ impl Endpoint {
        ])
        .args(["--pgdata", self.pgdata().to_str().unwrap()])
        .args(["--connstr", &conn_str])
-        // TODO(tristan957): Change this to --config when compatibility tests
-        // are no longer an issue
        .args([
            "--spec-path",
-            self.endpoint_path()
-                .join(if old_compute_ctl {
-                    "spec.json"
-                } else {
-                    "config.json"
-                })
-                .to_str()
-                .unwrap(),
+            self.endpoint_path().join("spec.json").to_str().unwrap(),
        ])
        .args([
            "--pgbin",
@@ -905,12 +869,10 @@ impl Endpoint {
        stripe_size: Option<ShardStripeSize>,
        safekeepers: Option<Vec<NodeId>>,
    ) -> Result<()> {
-        let (mut spec, compute_ctl_config) = {
-            let config_path = self.endpoint_path().join("config.json");
-            let file = std::fs::File::open(config_path)?;
-            let config: ComputeConfig = serde_json::from_reader(file)?;
-
-            (config.spec.unwrap(), config.compute_ctl_config)
+        let mut spec: ComputeSpec = {
+            let spec_path = self.endpoint_path().join("spec.json");
+            let file = std::fs::File::open(spec_path)?;
+            serde_json::from_reader(file)?
        };

        let postgresql_conf = self.read_postgresql_conf()?;
@@ -960,7 +922,7 @@ impl Endpoint {
            .body(
                serde_json::to_string(&ConfigurationRequest {
                    spec,
-                    compute_ctl_config,
+                    compute_ctl_config: ComputeCtlConfig::default(),
                })
                .unwrap(),
            )
--- a/control_plane/src/lib.rs
+++ b/control_plane/src/lib.rs
@@ -10,7 +10,6 @@ mod background_process;
 pub mod broker;
 pub mod endpoint;
 pub mod local_env;
-pub mod object_storage;
 pub mod pageserver;
 pub mod postgresql_conf;
 pub mod safekeeper;
--- a/control_plane/src/local_env.rs
+++ b/control_plane/src/local_env.rs
@@ -15,10 +15,9 @@ use clap::ValueEnum;
 use postgres_backend::AuthType;
 use reqwest::Url;
 use serde::{Deserialize, Serialize};
-use utils::auth::encode_from_key_file;
+use utils::auth::{Claims, encode_from_key_file};
 use utils::id::{NodeId, TenantId, TenantTimelineId, TimelineId};

-use crate::object_storage::{OBJECT_STORAGE_REMOTE_STORAGE_DIR, ObjectStorage};
 use crate::pageserver::{PAGESERVER_REMOTE_STORAGE_DIR, PageServerNode};
 use crate::safekeeper::SafekeeperNode;

@@ -56,7 +55,6 @@ pub struct LocalEnv {

    // used to issue tokens during e.g pg start
    pub private_key_path: PathBuf,
-    pub public_key_path: PathBuf,

    pub broker: NeonBroker,

@@ -70,8 +68,6 @@ pub struct LocalEnv {

    pub safekeepers: Vec<SafekeeperConf>,

-    pub object_storage: ObjectStorageConf,
-
    // Control plane upcall API for pageserver: if None, we will not run storage_controller  If set, this will
    // be propagated into each pageserver's configuration.
    pub control_plane_api: Url,
@@ -99,7 +95,6 @@ pub struct OnDiskConfig {
    pub neon_distrib_dir: PathBuf,
    pub default_tenant_id: Option<TenantId>,
    pub private_key_path: PathBuf,
-    pub public_key_path: PathBuf,
    pub broker: NeonBroker,
    pub storage_controller: NeonStorageControllerConf,
    #[serde(
@@ -108,7 +103,6 @@ pub struct OnDiskConfig {
    )]
    pub pageservers: Vec<PageServerConf>,
    pub safekeepers: Vec<SafekeeperConf>,
-    pub object_storage: ObjectStorageConf,
    pub control_plane_api: Option<Url>,
    pub control_plane_hooks_api: Option<Url>,
    pub control_plane_compute_hook_api: Option<Url>,
@@ -142,18 +136,11 @@ pub struct NeonLocalInitConf {
    pub storage_controller: Option<NeonStorageControllerConf>,
    pub pageservers: Vec<NeonLocalInitPageserverConf>,
    pub safekeepers: Vec<SafekeeperConf>,
-    pub object_storage: ObjectStorageConf,
    pub control_plane_api: Option<Url>,
    pub control_plane_hooks_api: Option<Url>,
    pub generate_local_ssl_certs: bool,
 }

-#[derive(Serialize, Default, Deserialize, PartialEq, Eq, Clone, Debug)]
-#[serde(default)]
-pub struct ObjectStorageConf {
-    pub port: u16,
-}
-
 /// Broker config for cluster internal communication.
 #[derive(Serialize, Deserialize, PartialEq, Eq, Clone, Debug)]
 #[serde(default)]
@@ -197,8 +184,6 @@ pub struct NeonStorageControllerConf {
    pub timelines_onto_safekeepers: bool,

    pub use_https_safekeeper_api: bool,
-
-    pub use_local_compute_notifications: bool,
 }

 impl NeonStorageControllerConf {
@@ -228,7 +213,6 @@ impl Default for NeonStorageControllerConf {
            use_https_pageserver_api: false,
            timelines_onto_safekeepers: false,
            use_https_safekeeper_api: false,
-            use_local_compute_notifications: true,
        }
    }
 }
@@ -411,10 +395,6 @@ impl LocalEnv {
        self.pg_dir(pg_version, "lib")
    }

-    pub fn object_storage_bin(&self) -> PathBuf {
-        self.neon_distrib_dir.join("object_storage")
-    }
-
    pub fn pageserver_bin(&self) -> PathBuf {
        self.neon_distrib_dir.join("pageserver")
    }
@@ -448,10 +428,6 @@ impl LocalEnv {
        self.base_data_dir.join("safekeepers").join(data_dir_name)
    }

-    pub fn object_storage_data_dir(&self) -> PathBuf {
-        self.base_data_dir.join("object_storage")
-    }
-
    pub fn get_pageserver_conf(&self, id: NodeId) -> anyhow::Result<&PageServerConf> {
        if let Some(conf) = self.pageservers.iter().find(|node| node.id == id) {
            Ok(conf)
@@ -603,7 +579,6 @@ impl LocalEnv {
                neon_distrib_dir,
                default_tenant_id,
                private_key_path,
-                public_key_path,
                broker,
                storage_controller,
                pageservers,
@@ -613,7 +588,6 @@ impl LocalEnv {
                control_plane_compute_hook_api: _,
                branch_name_mappings,
                generate_local_ssl_certs,
-                object_storage,
            } = on_disk_config;
            LocalEnv {
                base_data_dir: repopath.to_owned(),
@@ -621,7 +595,6 @@ impl LocalEnv {
                neon_distrib_dir,
                default_tenant_id,
                private_key_path,
-                public_key_path,
                broker,
                storage_controller,
                pageservers,
@@ -630,7 +603,6 @@ impl LocalEnv {
                control_plane_hooks_api,
                branch_name_mappings,
                generate_local_ssl_certs,
-                object_storage,
            }
        };

@@ -730,7 +702,6 @@ impl LocalEnv {
                neon_distrib_dir: self.neon_distrib_dir.clone(),
                default_tenant_id: self.default_tenant_id,
                private_key_path: self.private_key_path.clone(),
-                public_key_path: self.public_key_path.clone(),
                broker: self.broker.clone(),
                storage_controller: self.storage_controller.clone(),
                pageservers: vec![], // it's skip_serializing anyway
@@ -740,7 +711,6 @@ impl LocalEnv {
                control_plane_compute_hook_api: None,
                branch_name_mappings: self.branch_name_mappings.clone(),
                generate_local_ssl_certs: self.generate_local_ssl_certs,
-                object_storage: self.object_storage.clone(),
            },
        )
    }
@@ -757,7 +727,7 @@ impl LocalEnv {
    }

    // this function is used only for testing purposes in CLI e g generate tokens during init
-    pub fn generate_auth_token<S: Serialize>(&self, claims: &S) -> anyhow::Result<String> {
+    pub fn generate_auth_token(&self, claims: &Claims) -> anyhow::Result<String> {
        let private_key_path = self.get_private_key_path();
        let key_data = fs::read(private_key_path)?;
        encode_from_key_file(claims, &key_data)
@@ -824,7 +794,6 @@ impl LocalEnv {
            control_plane_api,
            generate_local_ssl_certs,
            control_plane_hooks_api,
-            object_storage,
        } = conf;

        // Find postgres binaries.
@@ -856,7 +825,6 @@ impl LocalEnv {
        )
        .context("generate auth keys")?;
        let private_key_path = PathBuf::from("auth_private_key.pem");
-        let public_key_path = PathBuf::from("auth_public_key.pem");

        // create the runtime type because the remaining initialization code below needs
        // a LocalEnv instance op operation
@@ -867,7 +835,6 @@ impl LocalEnv {
            neon_distrib_dir,
            default_tenant_id: Some(default_tenant_id),
            private_key_path,
-            public_key_path,
            broker,
            storage_controller: storage_controller.unwrap_or_default(),
            pageservers: pageservers.iter().map(Into::into).collect(),
@@ -876,7 +843,6 @@ impl LocalEnv {
            control_plane_hooks_api,
            branch_name_mappings: Default::default(),
            generate_local_ssl_certs,
-            object_storage,
        };

        if generate_local_ssl_certs {
@@ -904,13 +870,8 @@ impl LocalEnv {
                .context("pageserver init failed")?;
        }

-        ObjectStorage::from_env(&env)
-            .init()
-            .context("object storage init failed")?;
-
        // setup remote remote location for default LocalFs remote storage
        std::fs::create_dir_all(env.base_data_dir.join(PAGESERVER_REMOTE_STORAGE_DIR))?;
-        std::fs::create_dir_all(env.base_data_dir.join(OBJECT_STORAGE_REMOTE_STORAGE_DIR))?;

        env.persist_config()
    }
@@ -980,7 +941,7 @@ fn generate_ssl_ca_cert(cert_path: &Path, key_path: &Path) -> anyhow::Result<()>
    // -out rootCA.crt -keyout rootCA.key
    let keygen_output = Command::new("openssl")
        .args([
-            "req", "-x509", "-newkey", "ed25519", "-nodes", "-days", "36500",
+            "req", "-x509", "-newkey", "rsa:2048", "-nodes", "-days", "36500",
        ])
        .args(["-subj", "/CN=Neon Local CA"])
        .args(["-out", cert_path.to_str().unwrap()])
@@ -1010,7 +971,7 @@ fn generate_ssl_cert(
    // -subj "/CN=localhost" -addext "subjectAltName=DNS:localhost,IP:127.0.0.1"
    let keygen_output = Command::new("openssl")
        .args(["req", "-new", "-nodes"])
-        .args(["-newkey", "ed25519"])
+        .args(["-newkey", "rsa:2048"])
        .args(["-subj", "/CN=localhost"])
        .args(["-addext", "subjectAltName=DNS:localhost,IP:127.0.0.1"])
        .args(["-keyout", key_path.to_str().unwrap()])
--- a/control_plane/src/object_storage.rs
+++ b/control_plane/src/object_storage.rs
@@ -1,107 +0,0 @@
-use crate::background_process::{self, start_process, stop_process};
-use crate::local_env::LocalEnv;
-use anyhow::anyhow;
-use anyhow::{Context, Result};
-use camino::Utf8PathBuf;
-use std::io::Write;
-use std::time::Duration;
-
-/// Directory within .neon which will be used by default for LocalFs remote storage.
-pub const OBJECT_STORAGE_REMOTE_STORAGE_DIR: &str = "local_fs_remote_storage/object_storage";
-pub const OBJECT_STORAGE_DEFAULT_PORT: u16 = 9993;
-
-pub struct ObjectStorage {
-    pub bin: Utf8PathBuf,
-    pub data_dir: Utf8PathBuf,
-    pub pemfile: Utf8PathBuf,
-    pub port: u16,
-}
-
-impl ObjectStorage {
-    pub fn from_env(env: &LocalEnv) -> ObjectStorage {
-        ObjectStorage {
-            bin: Utf8PathBuf::from_path_buf(env.object_storage_bin()).unwrap(),
-            data_dir: Utf8PathBuf::from_path_buf(env.object_storage_data_dir()).unwrap(),
-            pemfile: Utf8PathBuf::from_path_buf(env.public_key_path.clone()).unwrap(),
-            port: env.object_storage.port,
-        }
-    }
-
-    fn config_path(&self) -> Utf8PathBuf {
-        self.data_dir.join("object_storage.json")
-    }
-
-    fn listen_addr(&self) -> Utf8PathBuf {
-        format!("127.0.0.1:{}", self.port).into()
-    }
-
-    pub fn init(&self) -> Result<()> {
-        println!("Initializing object storage in {:?}", self.data_dir);
-        let parent = self.data_dir.parent().unwrap();
-
-        #[derive(serde::Serialize)]
-        struct Cfg {
-            listen: Utf8PathBuf,
-            pemfile: Utf8PathBuf,
-            local_path: Utf8PathBuf,
-            r#type: String,
-        }
-        let cfg = Cfg {
-            listen: self.listen_addr(),
-            pemfile: parent.join(self.pemfile.clone()),
-            local_path: parent.join(OBJECT_STORAGE_REMOTE_STORAGE_DIR),
-            r#type: "LocalFs".to_string(),
-        };
-        std::fs::create_dir_all(self.config_path().parent().unwrap())?;
-        std::fs::write(self.config_path(), serde_json::to_string(&cfg)?)
-            .context("write object storage config")?;
-        Ok(())
-    }
-
-    pub async fn start(&self, retry_timeout: &Duration) -> Result<()> {
-        println!("Starting s3 proxy at {}", self.listen_addr());
-        std::io::stdout().flush().context("flush stdout")?;
-
-        let process_status_check = || async {
-            tokio::time::sleep(Duration::from_millis(500)).await;
-            let res = reqwest::Client::new()
-                .get(format!("http://{}/metrics", self.listen_addr()))
-                .send()
-                .await;
-            match res {
-                Ok(response) if response.status().is_success() => Ok(true),
-                Ok(_) => Err(anyhow!("Failed to query /metrics")),
-                Err(e) => Err(anyhow!("Failed to check node status: {e}")),
-            }
-        };
-
-        let res = start_process(
-            "object_storage",
-            &self.data_dir.clone().into_std_path_buf(),
-            &self.bin.clone().into_std_path_buf(),
-            vec![self.config_path().to_string()],
-            vec![("RUST_LOG".into(), "debug".into())],
-            background_process::InitialPidFile::Create(self.pid_file()),
-            retry_timeout,
-            process_status_check,
-        )
-        .await;
-        if res.is_err() {
-            eprintln!("Logs:\n{}", std::fs::read_to_string(self.log_file())?);
-        }
-
-        res
-    }
-
-    pub fn stop(&self, immediate: bool) -> anyhow::Result<()> {
-        stop_process(immediate, "object_storage", &self.pid_file())
-    }
-
-    fn log_file(&self) -> Utf8PathBuf {
-        self.data_dir.join("object_storage.log")
-    }
-
-    fn pid_file(&self) -> Utf8PathBuf {
-        self.data_dir.join("object_storage.pid")
-    }
-}
--- a/control_plane/src/pageserver.rs
+++ b/control_plane/src/pageserver.rs
@@ -428,6 +428,11 @@ impl PageServerNode {
                .map(|x| x.parse::<usize>())
                .transpose()
                .context("Failed to parse 'l0_flush_delay_threshold' as an integer")?,
+            l0_flush_wait_upload: settings
+                .remove("l0_flush_wait_upload")
+                .map(|x| x.parse::<bool>())
+                .transpose()
+                .context("Failed to parse 'l0_flush_wait_upload' as a boolean")?,
            l0_flush_stall_threshold: settings
                .remove("l0_flush_stall_threshold")
                .map(|x| x.parse::<usize>())
@@ -535,11 +540,6 @@ impl PageServerNode {
                .map(|x| x.parse::<bool>())
                .transpose()
                .context("Failed to parse 'gc_compaction_enabled' as bool")?,
-            gc_compaction_verification: settings
-                .remove("gc_compaction_verification")
-                .map(|x| x.parse::<bool>())
-                .transpose()
-                .context("Failed to parse 'gc_compaction_verification' as bool")?,
            gc_compaction_initial_threshold_kb: settings
                .remove("gc_compaction_initial_threshold_kb")
                .map(|x| x.parse::<u64>())
@@ -550,11 +550,6 @@ impl PageServerNode {
                .map(|x| x.parse::<u64>())
                .transpose()
                .context("Failed to parse 'gc_compaction_ratio_percent' as integer")?,
-            sampling_ratio: settings
-                .remove("sampling_ratio")
-                .map(serde_json::from_str)
-                .transpose()
-                .context("Falied to parse 'sampling_ratio'")?,
        };
        if !settings.is_empty() {
            bail!("Unrecognized tenant settings: {settings:?}")
--- a/control_plane/src/storage_controller.rs
+++ b/control_plane/src/storage_controller.rs
@@ -13,9 +13,7 @@ use pageserver_api::controller_api::{
    NodeConfigureRequest, NodeDescribeResponse, NodeRegisterRequest, TenantCreateRequest,
    TenantCreateResponse, TenantLocateResponse,
 };
-use pageserver_api::models::{
-    TenantConfig, TenantConfigRequest, TimelineCreateRequest, TimelineInfo,
-};
+use pageserver_api::models::{TenantConfigRequest, TimelineCreateRequest, TimelineInfo};
 use pageserver_api::shard::TenantShardId;
 use pageserver_client::mgmt_api::ResponseErrorMessageExt;
 use postgres_backend::AuthType;
@@ -84,8 +82,7 @@ impl NeonStorageControllerStopArgs {
 pub struct AttachHookRequest {
    pub tenant_shard_id: TenantShardId,
    pub node_id: Option<NodeId>,
-    pub generation_override: Option<i32>, // only new tenants
-    pub config: Option<TenantConfig>,     // only new tenants
+    pub generation_override: Option<i32>,
 }

 #[derive(Serialize, Deserialize)]
@@ -558,10 +555,6 @@ impl StorageController {
            args.push("--use-https-safekeeper-api".to_string());
        }

-        if self.config.use_local_compute_notifications {
-            args.push("--use-local-compute-notifications".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()));
        }
@@ -808,7 +801,6 @@ impl StorageController {
            tenant_shard_id,
            node_id: Some(pageserver_id),
            generation_override: None,
-            config: None,
        };

        let response = self
--- a/control_plane/storcon_cli/src/main.rs
+++ b/control_plane/storcon_cli/src/main.rs
@@ -385,6 +385,8 @@ where
 async fn main() -> anyhow::Result<()> {
    let cli = Cli::parse();

+    let storcon_client = Client::new(cli.api.clone(), cli.jwt.clone());
+
    let ssl_ca_certs = match &cli.ssl_ca_file {
        Some(ssl_ca_file) => {
            let buf = tokio::fs::read(ssl_ca_file).await?;
@@ -399,11 +401,9 @@ async fn main() -> anyhow::Result<()> {
    }
    let http_client = http_client.build()?;

-    let storcon_client = Client::new(http_client.clone(), cli.api.clone(), cli.jwt.clone());
-
    let mut trimmed = cli.api.to_string();
    trimmed.pop();
-    let vps_client = mgmt_api::Client::new(http_client.clone(), trimmed, cli.jwt.as_deref());
+    let vps_client = mgmt_api::Client::new(http_client, trimmed, cli.jwt.as_deref());

    match cli.command {
        Command::NodeRegister {
@@ -941,7 +941,7 @@ async fn main() -> anyhow::Result<()> {
            let mut node_to_fill_descs = Vec::new();

            for desc in node_descs {
-                let to_drain = nodes.contains(&desc.id);
+                let to_drain = nodes.iter().any(|id| *id == desc.id);
                if to_drain {
                    node_to_drain_descs.push(desc);
                } else {
@@ -1056,7 +1056,7 @@ async fn main() -> anyhow::Result<()> {
            const DEFAULT_MIGRATE_CONCURRENCY: usize = 8;
            let mut stream = futures::stream::iter(moves)
                .map(|mv| {
-                    let client = Client::new(http_client.clone(), cli.api.clone(), cli.jwt.clone());
+                    let client = Client::new(cli.api.clone(), cli.jwt.clone());
                    async move {
                        client
                            .dispatch::<TenantShardMigrateRequest, TenantShardMigrateResponse>(
--- a/docker-compose/compute_wrapper/shell/compute.sh
+++ b/docker-compose/compute_wrapper/shell/compute.sh
@@ -11,8 +11,8 @@ generate_id() {

 PG_VERSION=${PG_VERSION:-14}

-CONFIG_FILE_ORG=/var/db/postgres/configs/config.json
-CONFIG_FILE=/tmp/config.json
+SPEC_FILE_ORG=/var/db/postgres/specs/spec.json
+SPEC_FILE=/tmp/spec.json

 echo "Waiting pageserver become ready."
 while ! nc -z pageserver 6400; do
@@ -20,7 +20,7 @@ while ! nc -z pageserver 6400; do
 done
 echo "Page server is ready."

-cp ${CONFIG_FILE_ORG} ${CONFIG_FILE}
+cp ${SPEC_FILE_ORG} ${SPEC_FILE}

 if [ -n "${TENANT_ID:-}" ] && [ -n "${TIMELINE_ID:-}" ]; then
   tenant_id=${TENANT_ID}
@@ -73,27 +73,17 @@ else
  ulid_extension=ulid
 fi
 echo "Adding pgx_ulid"
-shared_libraries=$(jq -r '.spec.cluster.settings[] | select(.name=="shared_preload_libraries").value' ${CONFIG_FILE})
-sed -i "s/${shared_libraries}/${shared_libraries},${ulid_extension}/" ${CONFIG_FILE}
+shared_libraries=$(jq -r '.cluster.settings[] | select(.name=="shared_preload_libraries").value' ${SPEC_FILE})
+sed -i "s/${shared_libraries}/${shared_libraries},${ulid_extension}/" ${SPEC_FILE}
 echo "Overwrite tenant id and timeline id in spec file"
-sed -i "s/TENANT_ID/${tenant_id}/" ${CONFIG_FILE}
-sed -i "s/TIMELINE_ID/${timeline_id}/" ${CONFIG_FILE}
+sed -i "s/TENANT_ID/${tenant_id}/" ${SPEC_FILE}
+sed -i "s/TIMELINE_ID/${timeline_id}/" ${SPEC_FILE}

-cat ${CONFIG_FILE}
-
-# TODO(tristan957): Remove these workarounds for backwards compatibility after
-# the next compute release. That includes these next few lines and the
-# --spec-path in the compute_ctl invocation.
-if compute_ctl --help | grep --quiet -- '--config'; then
-  SPEC_PATH="$CONFIG_FILE"
-else
-  jq '.spec' < "$CONFIG_FILE" > /tmp/spec.json
-  SPEC_PATH=/tmp/spec.json
-fi
+cat ${SPEC_FILE}

 echo "Start compute node"
 /usr/local/bin/compute_ctl --pgdata /var/db/postgres/compute \
     -C "postgresql://cloud_admin@localhost:55433/postgres"  \
     -b /usr/local/bin/postgres                              \
     --compute-id "compute-$RANDOM"                          \
-     --spec-path "$SPEC_PATH"
+     -S ${SPEC_FILE}
--- a/docker-compose/compute_wrapper/var/db/postgres/configs/config.json
+++ b/docker-compose/compute_wrapper/var/db/postgres/configs/config.json
@@ -1,148 +0,0 @@
-{
-    "spec": {
-        "format_version": 1.0,
-
-        "timestamp": "2022-10-12T18:00:00.000Z",
-        "operation_uuid": "0f657b36-4b0f-4a2d-9c2e-1dcd615e7d8c",
-
-        "cluster": {
-            "cluster_id": "docker_compose",
-            "name": "docker_compose_test",
-            "state": "restarted",
-            "roles": [
-                {
-                    "name": "cloud_admin",
-                    "encrypted_password": "b093c0d3b281ba6da1eacc608620abd8",
-                    "options": null
-                }
-            ],
-            "databases": [
-            ],
-            "settings": [
-                {
-                    "name": "fsync",
-                    "value": "off",
-                    "vartype": "bool"
-                },
-                {
-                    "name": "wal_level",
-                    "value": "logical",
-                    "vartype": "enum"
-                },
-                {
-                    "name": "wal_log_hints",
-                    "value": "on",
-                    "vartype": "bool"
-                },
-                {
-                    "name": "log_connections",
-                    "value": "on",
-                    "vartype": "bool"
-                },
-                {
-                    "name": "port",
-                    "value": "55433",
-                    "vartype": "integer"
-                },
-                {
-                    "name": "shared_buffers",
-                    "value": "1MB",
-                    "vartype": "string"
-                },
-                {
-                    "name": "max_connections",
-                    "value": "100",
-                    "vartype": "integer"
-                },
-                {
-                    "name": "listen_addresses",
-                    "value": "0.0.0.0",
-                    "vartype": "string"
-                },
-                {
-                    "name": "max_wal_senders",
-                    "value": "10",
-                    "vartype": "integer"
-                },
-                {
-                    "name": "max_replication_slots",
-                    "value": "10",
-                    "vartype": "integer"
-                },
-                {
-                    "name": "wal_sender_timeout",
-                    "value": "5s",
-                    "vartype": "string"
-                },
-                {
-                    "name": "wal_keep_size",
-                    "value": "0",
-                    "vartype": "integer"
-                },
-                {
-                    "name": "password_encryption",
-                    "value": "md5",
-                    "vartype": "enum"
-                },
-                {
-                    "name": "restart_after_crash",
-                    "value": "off",
-                    "vartype": "bool"
-                },
-                {
-                    "name": "synchronous_standby_names",
-                    "value": "walproposer",
-                    "vartype": "string"
-                },
-                {
-                    "name": "shared_preload_libraries",
-                    "value": "neon,pg_cron,timescaledb,pg_stat_statements",
-                    "vartype": "string"
-                },
-                {
-                    "name": "neon.safekeepers",
-                    "value": "safekeeper1:5454,safekeeper2:5454,safekeeper3:5454",
-                    "vartype": "string"
-                },
-                {
-                    "name": "neon.timeline_id",
-                    "value": "TIMELINE_ID",
-                    "vartype": "string"
-                },
-                {
-                    "name": "neon.tenant_id",
-                    "value": "TENANT_ID",
-                    "vartype": "string"
-                },
-                {
-                    "name": "neon.pageserver_connstring",
-                    "value": "host=pageserver port=6400",
-                    "vartype": "string"
-                },
-                {
-                    "name": "max_replication_write_lag",
-                    "value": "500MB",
-                    "vartype": "string"
-                },
-                {
-                    "name": "max_replication_flush_lag",
-                    "value": "10GB",
-                    "vartype": "string"
-                },
-                {
-                    "name": "cron.database",
-                    "value": "postgres",
-                    "vartype": "string"
-                }
-            ]
-        },
-
-        "delta_operations": [
-        ]
-    },
-    "compute_ctl_config": {
-        "jwks": {
-            "keys": []
-        }
-    }
-}
--- a/docker-compose/compute_wrapper/var/db/postgres/specs/spec.json
+++ b/docker-compose/compute_wrapper/var/db/postgres/specs/spec.json
@@ -0,0 +1,141 @@
+{
+    "format_version": 1.0,
+
+    "timestamp": "2022-10-12T18:00:00.000Z",
+    "operation_uuid": "0f657b36-4b0f-4a2d-9c2e-1dcd615e7d8c",
+
+    "cluster": {
+        "cluster_id": "docker_compose",
+        "name": "docker_compose_test",
+        "state": "restarted",
+        "roles": [
+            {
+                "name": "cloud_admin",
+                "encrypted_password": "b093c0d3b281ba6da1eacc608620abd8",
+                "options": null
+            }
+        ],
+        "databases": [
+        ],
+        "settings": [
+            {
+                "name": "fsync",
+                "value": "off",
+                "vartype": "bool"
+            },
+            {
+                "name": "wal_level",
+                "value": "logical",
+                "vartype": "enum"
+            },
+            {
+                "name": "wal_log_hints",
+                "value": "on",
+                "vartype": "bool"
+            },
+            {
+                "name": "log_connections",
+                "value": "on",
+                "vartype": "bool"
+            },
+            {
+                "name": "port",
+                "value": "55433",
+                "vartype": "integer"
+            },
+            {
+                "name": "shared_buffers",
+                "value": "1MB",
+                "vartype": "string"
+            },
+            {
+                "name": "max_connections",
+                "value": "100",
+                "vartype": "integer"
+            },
+            {
+                "name": "listen_addresses",
+                "value": "0.0.0.0",
+                "vartype": "string"
+            },
+            {
+                "name": "max_wal_senders",
+                "value": "10",
+                "vartype": "integer"
+            },
+            {
+                "name": "max_replication_slots",
+                "value": "10",
+                "vartype": "integer"
+            },
+            {
+                "name": "wal_sender_timeout",
+                "value": "5s",
+                "vartype": "string"
+            },
+            {
+                "name": "wal_keep_size",
+                "value": "0",
+                "vartype": "integer"
+            },
+            {
+                "name": "password_encryption",
+                "value": "md5",
+                "vartype": "enum"
+            },
+            {
+                "name": "restart_after_crash",
+                "value": "off",
+                "vartype": "bool"
+            },
+            {
+                "name": "synchronous_standby_names",
+                "value": "walproposer",
+                "vartype": "string"
+            },
+            {
+                "name": "shared_preload_libraries",
+                "value": "neon,pg_cron,timescaledb,pg_stat_statements",
+                "vartype": "string"
+            },
+            {
+                "name": "neon.safekeepers",
+                "value": "safekeeper1:5454,safekeeper2:5454,safekeeper3:5454",
+                "vartype": "string"
+            },
+            {
+                "name": "neon.timeline_id",
+                "value": "TIMELINE_ID",
+                "vartype": "string"
+            },
+            {
+                "name": "neon.tenant_id",
+                "value": "TENANT_ID",
+                "vartype": "string"
+            },
+            {
+                "name": "neon.pageserver_connstring",
+                "value": "host=pageserver port=6400",
+                "vartype": "string"
+            },
+            {
+                "name": "max_replication_write_lag",
+                "value": "500MB",
+                "vartype": "string"
+            },
+            {
+                "name": "max_replication_flush_lag",
+                "value": "10GB",
+                "vartype": "string"
+            },
+            {
+                "name": "cron.database",
+                "value": "postgres",
+                "vartype": "string"
+            }
+        ]
+    },
+
+    "delta_operations": [
+    ]
+}
--- a/docker-compose/docker-compose.yml
+++ b/docker-compose/docker-compose.yml
@@ -159,7 +159,7 @@ services:
      #- RUST_BACKTRACE=1
    # Mount the test files directly, for faster editing cycle.
    volumes:
-      - ./compute_wrapper/var/db/postgres/configs/:/var/db/postgres/configs/
+      - ./compute_wrapper/var/db/postgres/specs/:/var/db/postgres/specs/
      - ./compute_wrapper/shell/:/shell/
    ports:
      - 55433:55433 # pg protocol handler
--- a/docs/SUMMARY.md
+++ b/docs/SUMMARY.md
@@ -21,7 +21,6 @@ in this repository.
    - [WAL Redo](./pageserver-walredo.md)
    - [Page cache](./pageserver-pagecache.md)
    - [Storage](./pageserver-storage.md)
-    - [Compaction](./pageserver-compaction.md)
    - [Processing a GetPage request](./pageserver-processing-getpage.md)
    - [Processing WAL](./pageserver-processing-wal.md)

--- a/docs/pageserver-compaction.md
+++ b/docs/pageserver-compaction.md
@@ -1,110 +0,0 @@
-# Pageserver Compaction
-
-Lifted from <https://www.notion.so/neondatabase/Rough-Notes-on-Compaction-1baf189e004780859e65ef63b85cfa81?pvs=4>.
-
-Updated 2025-03-26.
-
-## Pages and WAL
-
-Postgres stores data in 8 KB pages, identified by a page number.
-
-The WAL contains a sequence of page writes: either images (complete page contents) or deltas (patches applied to images). Each write is identified by its byte position in the WAL, aka LSN. 
-
-Each page version is thus identified by page@LSN. Postgres may read pages at past LSNs.
-
-Pageservers ingest WAL by writing WAL records into a key/value store keyed by page@LSN.
-
-Pageservers materialize pages for Postgres reads by finding the most recent page image and applying all subsequent page deltas, up to the read LSN.
-
-## Compaction: Why?
-
-Pageservers store page@LSN keys in a key/value store using a custom variant of an LSM tree. Each timeline on each tenant shard has its own LSM tree.
-
-When Pageservers write new page@LSN entries, they are appended unordered to an ephemeral layer file. When the ephemeral layer file exceeds `checkpoint_distance` (default 256 MB), the key/value pairs are sorted by key and written out to a layer file (for efficient lookups).
-
-As WAL writes continue, more layer files accumulate.
-
-Reads must search through the layer files to find the page’s image and deltas. The more layer files accumulate, the more la yer files reads must search through before they find a page image, aka read amplification.
-
-Compaction’s job is to:
-
- Reduce read amplification by reorganizing and combining layer files.
- Remove old garbage from layer files.
-
-As part of this, it may combine several page deltas into a single page image where possible.
-
-## Compaction: How?
-
-Neon uses a non-standard variant of an LSM tree made up of two levels of layer files: L0 and L1.
-
-Compaction runs in two phases: L0→L1 compaction, and L1 image compaction.
-
-L0 contains a stack of L0 layers at decreasing LSN ranges. These have been flushed sequentially from ephemeral layers. Each L0 layer covers the entire page space (page 0 to ~infinity) and the LSN range that was ingested into it. L0 layers are therefore particularly bad for read amp, since every read must search all L0 layers below the read LSN. For example:
-
-```
-| Page 0-99 @ LSN 0400-04ff |
-| Page 0-99 @ LSN 0300-03ff |
-| Page 0-99 @ LSN 0200-02ff |
-| Page 0-99 @ LSN 0100-01ff |
-| Page 0-99 @ LSN 0000-00ff |
-```
-
-L0→L1 compaction takes the bottom-most chunk of L0 layer files of between `compaction_threshold` (default 10) and `compaction_upper_limit` (default 20) layers. It uses merge-sort to write out sorted L1 delta layers of size `compaction_target_size` (default 128 MB).
-
-L1 typically consists of a “bed” of image layers with materialized page images at a specific LSN, and then delta layers of various page/LSN ranges above them with page deltas. For example:
-
-```
-Delta layers:               |     30-84@0310-04ff      |
-Delta layers:    | 10-42@0200-02ff |           | 65-92@0174-02aa |
-Image layers: |    0-39@0100    |    40-79@0100    |    80-99@0100    |
-```
-
-L1 image compaction scans across the L1 keyspace at some LSN, materializes page images by reading the image and delta layers below the LSN (via vectored reads), and writes out new sorted image layers of roughly size `compaction_target_size` (default 128 MB) at that LSN.
-
-Layer files below the new image files’ LSN can be garbage collected when they are no longer needed for PITR.
-
-Even though the old layer files are not immediately garbage collected, the new image layers help with read amp because reads can stop traversing the layer stack as soon as they encounter a page image.
-
-## Compaction: When?
-
-Pageservers run a `compaction_loop` background task for each tenant shard. Every `compaction_period` (default 20 seconds) it will wake up and check if any of the shard’s timelines need compaction. Additionally, L0 layer flushes will eagerly wake the compaction loop if the L0 count exceeds `compaction_threshold` (default 10).
-
-L0 compaction runs if the number of L0 layers exceeds `compaction_threshold` (default 10).
-
-L1 image compaction runs across sections of the L1 keyspace that have at least `image_creation_threshold` (default 3) delta layers overlapping image layers.
-
-At most `CONCURRENT_BACKGROUND_TASKS` (default 3 / 4 * CPUs = 6) background tasks can run concurrently on a Pageserver, including compaction. Further compaction tasks must wait.
-
-Because L0 layers cause the most read amp (they overlap the entire keyspace and only contain page deltas), they are aggressively compacted down:
-
- L0 is compacted down across all tenant timelines before L1 compaction is attempted (`compaction_l0_first`).
- L0 compaction uses a separate concurrency limit of `CONCURRENT_L0_COMPACTION_TASKS` (default 3 / 4 * CPUs = 6) to avoid waiting for other tasks (`compaction_l0_semaphore`).
- If L0 compaction is needed on any tenant timeline, L1 image compaction will yield to start an immediate L0 compaction run (except for compaction run via admin APIs).
-
-## Backpressure
-
-With sustained heavy write loads, new L0 layers may be flushed faster than they can be compacted down. This can cause an unbounded buildup of read amplification and compaction debt, which can take hours to resolve even after the writes stop.
-
-To avoid this and allow compaction to keep up, layer flushes will slow writes down to apply backpressure on the workload:
-
- At `l0_flush_delay_threshold` (default 30) L0 layers, layer flushes are delayed by the flush duration, such that they take 2x as long.
- At `l0_flush_stall_threshold` (default disabled) L0 layers, layer flushes stall entirely until the L0 count falls back below the threshold. This is currently disabled because we don’t trust L0 compaction to be responsive enough.
-
-This backpressure is propagated to the compute by waiting for layer flushes when WAL ingestion rolls the ephemeral layer. The compute will significantly slow down WAL writes at:
-
- `max_replication_write_lag` (default 500 MB), when Pageserver WAL ingestion lags
- `max_replication_flush_lag` (default 10 GB), when Pageserver L0 flushes lag
-
-Combined, this means that the compute will backpressure when there are 30 L0 layers (30 * 256 MB = 7.7 GB) and the Pageserver WAL ingestion lags the compute by 500 MB, for a total of ~8 GB L0+ephemeral compaction debt on a single shard.
-
-Since we only delay L0 flushes by 2x when backpressuring, and haven’t enabled stalls, it is still possible for read amp to increase unbounded if compaction is too slow (although we haven’t seen this in practice). But this is considered better than stalling flushes and causing unavailability for as long as it takes L0 compaction to react, since we don’t trust it to be fast enough — at the expense of continually increasing read latency and CPU usage for this tenant. We should either enable stalls when we have enough confidence in L0 compaction, or scale the flush delay by the number of L0 layers to apply increasing backpressure.
-
-## Circuit Breaker
-
-Compaction can fail, often repeatedly. This can happen e.g. due to data corruption, faulty hardware, S3 outages, etc.
-
-If compaction fails, the compaction loop will naïvely try and fail again almost immediately. It may only fail after doing a significant amount of wasted work, while holding onto the background task semaphore.
-
-To avoid repeatedly doing wasted work and starving out other compaction jobs, each tenant has a compaction circuit breaker. After 5 repeated compaction failures, the circuit breaker trips and disables compaction for the next 24 hours, before resetting the breaker and trying again. This disables compaction across all tenant timelines (faulty or not).
-
-Disabling compaction for a long time is dangerous, since it can lead to unbounded read amp and compaction debt, and continuous workload backpressure. However, continually failing would not help either. Tripped circuit breakers trigger an alert and must be investigated promptly.
--- a/docs/storage_controller.md
+++ b/docs/storage_controller.md
@@ -151,7 +151,7 @@ Example body:
 ```
 {
  "tenant_id": "1f359dd625e519a1a4e8d7509690f6fc",
-  "stripe_size": 2048,
+  "stripe_size": 32768,
  "shards": [
      {"node_id": 344, "shard_number": 0},
      {"node_id": 722, "shard_number": 1},
--- a/explained_queries.sql
+++ b/explained_queries.sql
--- a/libs/compute_api/src/requests.rs
+++ b/libs/compute_api/src/requests.rs
@@ -5,14 +5,6 @@ use crate::privilege::Privilege;
 use crate::responses::ComputeCtlConfig;
 use crate::spec::{ComputeSpec, ExtVersion, PgIdent};

-/// When making requests to the `compute_ctl` external HTTP server, the client
-/// must specify a set of claims in `Authorization` header JWTs such that
-/// `compute_ctl` can authorize the request.
-#[derive(Clone, Debug, Deserialize, Serialize)]
-pub struct ComputeClaims {
-    pub compute_id: String,
-}
-
 /// Request of the /configure API
 ///
 /// We now pass only `spec` in the configuration request, but later we can
@@ -38,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
@@ -14,32 +14,6 @@ pub struct GenericAPIError {
    pub error: String,
 }

-/// All configuration parameters necessary for a compute. When
-/// [`ComputeConfig::spec`] is provided, it means that the compute is attached
-/// to a tenant. [`ComputeConfig::compute_ctl_config`] will always be provided
-/// and contains parameters necessary for operating `compute_ctl` independently
-/// of whether a tenant is attached to the compute or not.
-///
-/// This also happens to be the body of `compute_ctl`'s /configure request.
-#[derive(Debug, Deserialize, Serialize)]
-pub struct ComputeConfig {
-    /// The compute spec
-    pub spec: Option<ComputeSpec>,
-
-    /// The compute_ctl configuration
-    #[allow(dead_code)]
-    pub compute_ctl_config: ComputeCtlConfig,
-}
-
-impl From<ControlPlaneConfigResponse> for ComputeConfig {
-    fn from(value: ControlPlaneConfigResponse) -> Self {
-        Self {
-            spec: value.spec,
-            compute_ctl_config: value.compute_ctl_config,
-        }
-    }
-}
-
 #[derive(Debug, Clone, Serialize)]
 pub struct ExtensionInstallResponse {
    pub extension: PgIdent,
@@ -187,7 +161,7 @@ pub struct TlsConfig {

 /// Response of the `/computes/{compute_id}/spec` control-plane API.
 #[derive(Deserialize, Debug)]
-pub struct ControlPlaneConfigResponse {
+pub struct ControlPlaneSpecResponse {
    pub spec: Option<ComputeSpec>,
    pub status: ControlPlaneComputeStatus,
    pub compute_ctl_config: ComputeCtlConfig,
--- a/libs/compute_api/src/spec.rs
+++ b/libs/compute_api/src/spec.rs
@@ -1,8 +1,8 @@
-//! The ComputeSpec contains all the information needed to start up
-//! the right version of PostgreSQL, and connect it to the storage nodes.
-//! It can be passed as part of the `config.json`, or the control plane can
-//! provide it by calling the compute_ctl's `/compute_ctl` endpoint, or
-//! compute_ctl can fetch it by calling the control plane's API.
+//! `ComputeSpec` represents the contents of the spec.json file.
+//!
+//! The spec.json file is used to pass information to 'compute_ctl'. It contains
+//! all the information needed to start up the right version of PostgreSQL,
+//! and connect it to the storage nodes.
 use std::collections::HashMap;

 use indexmap::IndexMap;
@@ -104,12 +104,6 @@ pub struct ComputeSpec {
    pub timeline_id: Option<TimelineId>,
    pub pageserver_connstring: Option<String>,

-    // More neon ids that we expose to the compute_ctl
-    // and to postgres as neon extension GUCs.
-    pub project_id: Option<String>,
-    pub branch_id: Option<String>,
-    pub endpoint_id: Option<String>,
-
    /// Safekeeper membership config generation. It is put in
    /// neon.safekeepers GUC and serves two purposes:
    /// 1) Non zero value forces walproposer to use membership configurations.
@@ -165,13 +159,15 @@ pub struct ComputeSpec {
    #[serde(default)] // Default false
    pub drop_subscriptions_before_start: bool,

-    /// Log level for compute audit logging
+    /// Log level for audit logging:
+    ///
+    /// Disabled - no audit logging. This is the default.
+    /// log - log masked statements to the postgres log using pgaudit extension
+    /// hipaa - log unmasked statements to the file using pgaudit and pgauditlogtofile extension
+    ///
+    /// Extensions should be present in shared_preload_libraries
    #[serde(default)]
    pub audit_log_level: ComputeAudit,
-
-    /// Hostname and the port of the otel collector. Leave empty to disable Postgres logs forwarding.
-    /// Example: config-shy-breeze-123-collector-monitoring.neon-telemetry.svc.cluster.local:10514
-    pub logs_export_host: Option<String>,
 }

 /// Feature flag to signal `compute_ctl` to enable certain experimental functionality.
@@ -183,6 +179,9 @@ pub enum ComputeFeature {
    /// track short-lived connections as user activity.
    ActivityMonitorExperimental,

+    /// 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
    /// `parse_unknown_features()` for more details.
@@ -289,25 +288,14 @@ impl ComputeMode {
 }

 /// Log level for audit logging
+/// Disabled, log, hipaa
+/// Default is Disabled
 #[derive(Clone, Debug, Default, Eq, PartialEq, Deserialize, Serialize)]
 pub enum ComputeAudit {
    #[default]
    Disabled,
-    // Deprecated, use Base instead
    Log,
-    // (pgaudit.log = 'ddl', pgaudit.log_parameter='off')
-    // logged to the standard postgresql log stream
-    Base,
-    // Deprecated, use Full or Extended instead
    Hipaa,
-    // (pgaudit.log = 'all, -misc', pgaudit.log_parameter='off')
-    // logged to separate files collected by rsyslog
-    // into dedicated log storage with strict access
-    Extended,
-    // (pgaudit.log='all', pgaudit.log_parameter='on'),
-    // logged to separate files collected by rsyslog
-    // into dedicated log storage with strict access.
-    Full,
 }

 #[derive(Clone, Debug, Default, Deserialize, Serialize, PartialEq, Eq)]
--- a/libs/http-utils/Cargo.toml
+++ b/libs/http-utils/Cargo.toml
@@ -30,7 +30,6 @@ tokio.workspace = true
 tracing.workspace = true
 url.workspace = true
 uuid.workspace = true
-x509-cert.workspace = true

 # to use tokio channels as streams, this is faster to compile than async_stream
 # why is it only here? no other crate should use it, streams are rarely needed.
--- a/libs/http-utils/src/server.rs
+++ b/libs/http-utils/src/server.rs
@@ -4,8 +4,6 @@ use futures::StreamExt;
 use futures::stream::FuturesUnordered;
 use hyper0::Body;
 use hyper0::server::conn::Http;
-use metrics::{IntCounterVec, register_int_counter_vec};
-use once_cell::sync::Lazy;
 use routerify::{RequestService, RequestServiceBuilder};
 use tokio::io::{AsyncRead, AsyncWrite};
 use tokio_rustls::TlsAcceptor;
@@ -28,24 +26,6 @@ pub struct Server {
    tls_acceptor: Option<TlsAcceptor>,
 }

-static CONNECTION_STARTED_COUNT: Lazy<IntCounterVec> = Lazy::new(|| {
-    register_int_counter_vec!(
-        "http_server_connection_started_total",
-        "Number of established http/https connections",
-        &["scheme"]
-    )
-    .expect("failed to define a metric")
-});
-
-static CONNECTION_ERROR_COUNT: Lazy<IntCounterVec> = Lazy::new(|| {
-    register_int_counter_vec!(
-        "http_server_connection_errors_total",
-        "Number of occured connection errors by type",
-        &["type"]
-    )
-    .expect("failed to define a metric")
-});
-
 impl Server {
    pub fn new(
        request_service: Arc<RequestServiceBuilder<Body, ApiError>>,
@@ -80,15 +60,6 @@ impl Server {
            false
        }

-        let tcp_error_cnt = CONNECTION_ERROR_COUNT.with_label_values(&["tcp"]);
-        let tls_error_cnt = CONNECTION_ERROR_COUNT.with_label_values(&["tls"]);
-        let http_error_cnt = CONNECTION_ERROR_COUNT.with_label_values(&["http"]);
-        let https_error_cnt = CONNECTION_ERROR_COUNT.with_label_values(&["https"]);
-        let panic_error_cnt = CONNECTION_ERROR_COUNT.with_label_values(&["panic"]);
-
-        let http_connection_cnt = CONNECTION_STARTED_COUNT.with_label_values(&["http"]);
-        let https_connection_cnt = CONNECTION_STARTED_COUNT.with_label_values(&["https"]);
-
        let mut connections = FuturesUnordered::new();
        loop {
            tokio::select! {
@@ -96,7 +67,6 @@ impl Server {
                    let (tcp_stream, remote_addr) = match stream {
                        Ok(stream) => stream,
                        Err(err) => {
-                            tcp_error_cnt.inc();
                            if !suppress_io_error(&err) {
                                info!("Failed to accept TCP connection: {err:#}");
                            }
@@ -108,18 +78,11 @@ impl Server {
                    let tls_acceptor = self.tls_acceptor.clone();
                    let cancel = cancel.clone();

-                    let tls_error_cnt = tls_error_cnt.clone();
-                    let http_error_cnt = http_error_cnt.clone();
-                    let https_error_cnt = https_error_cnt.clone();
-                    let http_connection_cnt = http_connection_cnt.clone();
-                    let https_connection_cnt = https_connection_cnt.clone();
-
                    connections.push(tokio::spawn(
                        async move {
                            match tls_acceptor {
                                Some(tls_acceptor) => {
                                    // Handle HTTPS connection.
-                                    https_connection_cnt.inc();
                                    let tls_stream = tokio::select! {
                                        tls_stream = tls_acceptor.accept(tcp_stream) => tls_stream,
                                        _ = cancel.cancelled() => return,
@@ -127,27 +90,23 @@ impl Server {
                                    let tls_stream = match tls_stream {
                                        Ok(tls_stream) => tls_stream,
                                        Err(err) => {
-                                            tls_error_cnt.inc();
                                            if !suppress_io_error(&err) {
-                                                info!(%remote_addr, "Failed to accept TLS connection: {err:#}");
+                                                info!("Failed to accept TLS connection: {err:#}");
                                            }
                                            return;
                                        }
                                    };
                                    if let Err(err) = Self::serve_connection(tls_stream, service, cancel).await {
-                                        https_error_cnt.inc();
                                        if !suppress_hyper_error(&err) {
-                                            info!(%remote_addr, "Failed to serve HTTPS connection: {err:#}");
+                                            info!("Failed to serve HTTPS connection: {err:#}");
                                        }
                                    }
                                }
                                None => {
                                    // Handle HTTP connection.
-                                    http_connection_cnt.inc();
                                    if let Err(err) = Self::serve_connection(tcp_stream, service, cancel).await {
-                                        http_error_cnt.inc();
                                        if !suppress_hyper_error(&err) {
-                                            info!(%remote_addr, "Failed to serve HTTP connection: {err:#}");
+                                            info!("Failed to serve HTTP connection: {err:#}");
                                        }
                                    }
                                }
@@ -156,7 +115,6 @@ impl Server {
                 }
                Some(conn) = connections.next() => {
                    if let Err(err) = conn {
-                        panic_error_cnt.inc();
                        error!("Connection panicked: {err:#}");
                    }
                }
@@ -164,7 +122,6 @@ impl Server {
                    // Wait for graceful shutdown of all connections.
                    while let Some(conn) = connections.next().await {
                        if let Err(err) = conn {
-                            panic_error_cnt.inc();
                            error!("Connection panicked: {err:#}");
                        }
                    }
--- a/libs/http-utils/src/tls_certs.rs
+++ b/libs/http-utils/src/tls_certs.rs
@@ -3,14 +3,11 @@ use std::{sync::Arc, time::Duration};
 use anyhow::Context;
 use arc_swap::ArcSwap;
 use camino::Utf8Path;
-use metrics::{IntCounterVec, UIntGaugeVec, register_int_counter_vec, register_uint_gauge_vec};
-use once_cell::sync::Lazy;
 use rustls::{
-    pki_types::{CertificateDer, PrivateKeyDer, UnixTime},
+    pki_types::{CertificateDer, PrivateKeyDer},
    server::{ClientHello, ResolvesServerCert},
    sign::CertifiedKey,
 };
-use x509_cert::der::Reader;

 pub async fn load_cert_chain(filename: &Utf8Path) -> anyhow::Result<Vec<CertificateDer<'static>>> {
    let cert_data = tokio::fs::read(filename)
@@ -56,76 +53,6 @@ pub async fn load_certified_key(
    Ok(certified_key)
 }

-/// rustls's CertifiedKey with extra parsed fields used for metrics.
-struct ParsedCertifiedKey {
-    certified_key: CertifiedKey,
-    expiration_time: UnixTime,
-}
-
-/// Parse expiration time from an X509 certificate.
-fn parse_expiration_time(cert: &CertificateDer<'_>) -> anyhow::Result<UnixTime> {
-    let parsed_cert = x509_cert::der::SliceReader::new(cert)
-        .context("Failed to parse cerficiate")?
-        .decode::<x509_cert::Certificate>()
-        .context("Failed to parse cerficiate")?;
-
-    Ok(UnixTime::since_unix_epoch(
-        parsed_cert
-            .tbs_certificate
-            .validity
-            .not_after
-            .to_unix_duration(),
-    ))
-}
-
-async fn load_and_parse_certified_key(
-    key_filename: &Utf8Path,
-    cert_filename: &Utf8Path,
-) -> anyhow::Result<ParsedCertifiedKey> {
-    let certified_key = load_certified_key(key_filename, cert_filename).await?;
-    let expiration_time = parse_expiration_time(certified_key.end_entity_cert()?)?;
-    Ok(ParsedCertifiedKey {
-        certified_key,
-        expiration_time,
-    })
-}
-
-static CERT_EXPIRATION_TIME: Lazy<UIntGaugeVec> = Lazy::new(|| {
-    register_uint_gauge_vec!(
-        "tls_certs_expiration_time_seconds",
-        "Expiration time of the loaded certificate since unix epoch in seconds",
-        &["resolver_name"]
-    )
-    .expect("failed to define a metric")
-});
-
-static CERT_RELOAD_STARTED_COUNTER: Lazy<IntCounterVec> = Lazy::new(|| {
-    register_int_counter_vec!(
-        "tls_certs_reload_started_total",
-        "Number of certificate reload loop iterations started",
-        &["resolver_name"]
-    )
-    .expect("failed to define a metric")
-});
-
-static CERT_RELOAD_UPDATED_COUNTER: Lazy<IntCounterVec> = Lazy::new(|| {
-    register_int_counter_vec!(
-        "tls_certs_reload_updated_total",
-        "Number of times the certificate was updated to the new one",
-        &["resolver_name"]
-    )
-    .expect("failed to define a metric")
-});
-
-static CERT_RELOAD_FAILED_COUNTER: Lazy<IntCounterVec> = Lazy::new(|| {
-    register_int_counter_vec!(
-        "tls_certs_reload_failed_total",
-        "Number of times the certificate reload failed",
-        &["resolver_name"]
-    )
-    .expect("failed to define a metric")
-});
-
 /// Implementation of [`rustls::server::ResolvesServerCert`] which reloads certificates from
 /// the disk periodically.
 #[derive(Debug)]
@@ -136,28 +63,16 @@ pub struct ReloadingCertificateResolver {
 impl ReloadingCertificateResolver {
    /// Creates a new Resolver by loading certificate and private key from FS and
    /// creating tokio::task to reload them with provided reload_period.
-    /// resolver_name is used as metric's label.
    pub async fn new(
-        resolver_name: &str,
        key_filename: &Utf8Path,
        cert_filename: &Utf8Path,
        reload_period: Duration,
    ) -> anyhow::Result<Arc<Self>> {
-        // Create metrics for current resolver.
-        let cert_expiration_time = CERT_EXPIRATION_TIME.with_label_values(&[resolver_name]);
-        let cert_reload_started_counter =
-            CERT_RELOAD_STARTED_COUNTER.with_label_values(&[resolver_name]);
-        let cert_reload_updated_counter =
-            CERT_RELOAD_UPDATED_COUNTER.with_label_values(&[resolver_name]);
-        let cert_reload_failed_counter =
-            CERT_RELOAD_FAILED_COUNTER.with_label_values(&[resolver_name]);
-
-        let parsed_key = load_and_parse_certified_key(key_filename, cert_filename).await?;
-
        let this = Arc::new(Self {
-            certified_key: ArcSwap::from_pointee(parsed_key.certified_key),
+            certified_key: ArcSwap::from_pointee(
+                load_certified_key(key_filename, cert_filename).await?,
+            ),
        });
-        cert_expiration_time.set(parsed_key.expiration_time.as_secs());

        tokio::spawn({
            let weak_this = Arc::downgrade(&this);
@@ -173,22 +88,17 @@ impl ReloadingCertificateResolver {
                        Some(this) => this,
                        None => break, // Resolver has been destroyed, exit.
                    };
-                    cert_reload_started_counter.inc();
-
-                    match load_and_parse_certified_key(&key_filename, &cert_filename).await {
-                        Ok(parsed_key) => {
-                            if parsed_key.certified_key.cert == this.certified_key.load().cert {
+                    match load_certified_key(&key_filename, &cert_filename).await {
+                        Ok(new_certified_key) => {
+                            if new_certified_key.cert == this.certified_key.load().cert {
                                tracing::debug!("Certificate has not changed since last reloading");
                            } else {
                                tracing::info!("Certificate has been reloaded");
-                                this.certified_key.store(Arc::new(parsed_key.certified_key));
-                                cert_expiration_time.set(parsed_key.expiration_time.as_secs());
-                                cert_reload_updated_counter.inc();
+                                this.certified_key.store(Arc::new(new_certified_key));
                            }
                            last_reload_failed = false;
                        }
                        Err(err) => {
-                            cert_reload_failed_counter.inc();
                            // Note: Reloading certs may fail if it conflicts with the script updating
                            // the files at the same time. Warn only if the error is persistent.
                            if last_reload_failed {
--- a/libs/pageserver_api/Cargo.toml
+++ b/libs/pageserver_api/Cargo.toml
@@ -34,7 +34,6 @@ postgres_backend.workspace = true
 nix = {workspace = true, optional = true}
 reqwest.workspace = true
 rand.workspace = true
-tracing-utils.workspace = true

 [dev-dependencies]
 bincode.workspace = true
--- a/libs/pageserver_api/src/config.rs
+++ b/libs/pageserver_api/src/config.rs
@@ -51,54 +51,9 @@ pub struct NodeMetadata {
 /// If there cannot be a static default value because we need to make runtime
 /// checks to determine the default, make it an `Option` (which defaults to None).
 /// The runtime check should be done in the consuming crate, i.e., `pageserver`.
-///
-/// Unknown fields are silently ignored during deserialization.
-/// The alternative, which we used in the past, was to set `deny_unknown_fields`,
-/// which fails deserialization, and hence pageserver startup, if there is an unknown field.
-/// The reason we don't do that anymore is that it complicates
-/// usage of config fields for feature flagging, which we commonly do for
-/// region-by-region rollouts.
-/// The complications mainly arise because the `pageserver.toml` contents on a
-/// prod server have a separate lifecycle from the pageserver binary.
-/// For instance, `pageserver.toml` contents today are defined in the internal
-/// infra repo, and thus introducing a new config field to pageserver and
-/// rolling it out to prod servers are separate commits in separate repos
-/// that can't be made or rolled back atomically.
-/// Rollbacks in particular pose a risk with deny_unknown_fields because
-/// the old pageserver binary may reject a new config field, resulting in
-/// an outage unless the person doing the pageserver rollback remembers
-/// to also revert the commit that added the config field in to the
-/// `pageserver.toml` templates in the internal infra repo.
-/// (A pre-deploy config check would eliminate this risk during rollbacks,
-///  cf [here](https://github.com/neondatabase/cloud/issues/24349).)
-/// In addition to this compatibility problem during emergency rollbacks,
-/// deny_unknown_fields adds further complications when decomissioning a feature
-/// flag: with deny_unknown_fields, we can't remove a flag from the [`ConfigToml`]
-/// until all prod servers' `pageserver.toml` files have been updated to a version
-/// that doesn't specify the flag. Otherwise new software would fail to start up.
-/// This adds the requirement for an intermediate step where the new config field
-/// is accepted but ignored, prolonging the decomissioning process by an entire
-/// release cycle.
-/// By contrast  with unknown fields silently ignored, decomissioning a feature
-/// flag is a one-step process: we can skip the intermediate step and straight
-/// remove the field from the [`ConfigToml`]. We leave the field in the
-/// `pageserver.toml` files on prod servers until we reach certainty that we
-/// will not roll back to old software whose behavior was dependent on config.
-/// Then we can remove the field from the templates in the internal infra repo.
-/// This process is [documented internally](
-/// https://docs.neon.build/storage/pageserver_configuration.html).
-///
-/// Note that above relaxed compatbility for the config format does NOT APPLY
-/// TO THE STORAGE FORMAT. As general guidance, when introducing storage format
-/// changes, ensure that the potential rollback target version will be compatible
-/// with the new format. This must hold regardless of what flags are set in in the `pageserver.toml`:
-/// any format version that exists in an environment must be compatible with the software that runs there.
-/// Use a pageserver.toml flag only to gate whether software _writes_ the new format.
-/// For more compatibility considerations, refer to [internal docs](
-/// https://docs.neon.build/storage/compat.html?highlight=compat#format-versions--compatibility)
 #[serde_as]
 #[derive(Clone, Debug, serde::Deserialize, serde::Serialize)]
-#[serde(default)]
+#[serde(default, deny_unknown_fields)]
 pub struct ConfigToml {
    // types mapped 1:1 into the runtime PageServerConfig type
    pub listen_pg_addr: String,
@@ -179,11 +134,10 @@ pub struct ConfigToml {
    pub load_previous_heatmap: Option<bool>,
    #[serde(skip_serializing_if = "Option::is_none")]
    pub generate_unarchival_heatmap: Option<bool>,
-    pub tracing: Option<Tracing>,
-    pub enable_tls_page_service_api: bool,
 }

 #[derive(Debug, Clone, PartialEq, Eq, serde::Serialize, serde::Deserialize)]
+#[serde(deny_unknown_fields)]
 pub struct DiskUsageEvictionTaskConfig {
    pub max_usage_pct: utils::serde_percent::Percent,
    pub min_avail_bytes: u64,
@@ -198,19 +152,17 @@ pub struct DiskUsageEvictionTaskConfig {

 #[derive(Debug, Clone, PartialEq, Eq, serde::Serialize, serde::Deserialize)]
 #[serde(tag = "mode", rename_all = "kebab-case")]
+#[serde(deny_unknown_fields)]
 pub enum PageServicePipeliningConfig {
    Serial,
    Pipelined(PageServicePipeliningConfigPipelined),
 }
 #[derive(Debug, Clone, PartialEq, Eq, serde::Serialize, serde::Deserialize)]
+#[serde(deny_unknown_fields)]
 pub struct PageServicePipeliningConfigPipelined {
    /// Causes runtime errors if larger than max get_vectored batch size.
    pub max_batch_size: NonZeroUsize,
    pub execution: PageServiceProtocolPipelinedExecutionStrategy,
-    // The default below is such that new versions of the software can start
-    // with the old configuration.
-    #[serde(default)]
-    pub batching: PageServiceProtocolPipelinedBatchingStrategy,
 }

 #[derive(Debug, Clone, Copy, PartialEq, Eq, serde::Serialize, serde::Deserialize)]
@@ -220,21 +172,9 @@ pub enum PageServiceProtocolPipelinedExecutionStrategy {
    Tasks,
 }

-#[derive(Default, Debug, Clone, Copy, PartialEq, Eq, serde::Serialize, serde::Deserialize)]
-#[serde(rename_all = "kebab-case")]
-pub enum PageServiceProtocolPipelinedBatchingStrategy {
-    /// All get page requests in a batch will be at the same LSN
-    #[default]
-    UniformLsn,
-    /// Get page requests in a batch may be at different LSN
-    ///
-    /// One key cannot be present more than once at different LSNs in
-    /// the same batch.
-    ScatteredLsn,
-}
-
 #[derive(Debug, Clone, PartialEq, Eq, serde::Serialize, serde::Deserialize)]
 #[serde(tag = "mode", rename_all = "kebab-case")]
+#[serde(deny_unknown_fields)]
 pub enum GetVectoredConcurrentIo {
    /// The read path is fully sequential: layers are visited
    /// one after the other and IOs are issued and waited upon
@@ -251,54 +191,6 @@ pub enum GetVectoredConcurrentIo {
    SidecarTask,
 }

-#[derive(Debug, Copy, Clone, PartialEq, Eq, serde::Serialize, serde::Deserialize)]
-pub struct Ratio {
-    pub numerator: usize,
-    pub denominator: usize,
-}
-
-#[derive(Debug, Clone, PartialEq, Eq, serde::Serialize, serde::Deserialize)]
-pub struct OtelExporterConfig {
-    pub endpoint: String,
-    pub protocol: OtelExporterProtocol,
-    #[serde(with = "humantime_serde")]
-    pub timeout: Duration,
-}
-
-#[derive(Debug, Copy, Clone, PartialEq, Eq, serde::Serialize, serde::Deserialize)]
-#[serde(rename_all = "kebab-case")]
-pub enum OtelExporterProtocol {
-    Grpc,
-    HttpBinary,
-    HttpJson,
-}
-
-#[derive(Debug, Clone, PartialEq, Eq, serde::Serialize, serde::Deserialize)]
-pub struct Tracing {
-    pub sampling_ratio: Ratio,
-    pub export_config: OtelExporterConfig,
-}
-
-impl From<&OtelExporterConfig> for tracing_utils::ExportConfig {
-    fn from(val: &OtelExporterConfig) -> Self {
-        tracing_utils::ExportConfig {
-            endpoint: Some(val.endpoint.clone()),
-            protocol: val.protocol.into(),
-            timeout: val.timeout,
-        }
-    }
-}
-
-impl From<OtelExporterProtocol> for tracing_utils::Protocol {
-    fn from(val: OtelExporterProtocol) -> Self {
-        match val {
-            OtelExporterProtocol::Grpc => tracing_utils::Protocol::Grpc,
-            OtelExporterProtocol::HttpJson => tracing_utils::Protocol::HttpJson,
-            OtelExporterProtocol::HttpBinary => tracing_utils::Protocol::HttpBinary,
-        }
-    }
-}
-
 pub mod statvfs {
    pub mod mock {
        #[derive(Debug, Clone, PartialEq, Eq, serde::Serialize, serde::Deserialize)]
@@ -353,7 +245,7 @@ pub struct MaxVectoredReadBytes(pub NonZeroUsize);

 /// Tenant-level configuration values, used for various purposes.
 #[derive(Debug, Clone, PartialEq, Eq, serde::Serialize, serde::Deserialize)]
-#[serde(default)]
+#[serde(deny_unknown_fields, default)]
 pub struct TenantConfigToml {
    // Flush out an inmemory layer, if it's holding WAL older than this
    // This puts a backstop on how much WAL needs to be re-digested if the
@@ -393,6 +285,12 @@ pub struct TenantConfigToml {
    /// Level0 delta layer threshold at which to stall layer flushes. Must be >compaction_threshold
    /// to avoid deadlock. 0 to disable. Disabled by default.
    pub l0_flush_stall_threshold: Option<usize>,
+    /// 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.
    // The unit is #of bytes of WAL.
@@ -469,17 +367,12 @@ pub struct TenantConfigToml {
    // gc-compaction related configs
    /// Enable automatic gc-compaction trigger on this tenant.
    pub gc_compaction_enabled: bool,
-    /// Enable verification of gc-compaction results.
-    pub gc_compaction_verification: bool,
    /// The initial threshold for gc-compaction in KB. Once the total size of layers below the gc-horizon is above this threshold,
    /// gc-compaction will be triggered.
    pub gc_compaction_initial_threshold_kb: u64,
    /// The ratio that triggers the auto gc-compaction. If (the total size of layers between L2 LSN and gc-horizon) / (size below the L2 LSN)
    /// is above this ratio, gc-compaction will be triggered.
    pub gc_compaction_ratio_percent: u64,
-    /// Tenant level performance sampling ratio override. Controls the ratio of get page requests
-    /// that will get perf sampling for the tenant.
-    pub sampling_ratio: Option<Ratio>,
 }

 pub mod defaults {
@@ -632,12 +525,9 @@ impl Default for ConfigToml {
            page_service_pipelining: if !cfg!(test) {
                PageServicePipeliningConfig::Serial
            } else {
-                // Do not turn this into the default until scattered reads have been
-                // validated and rolled-out fully.
                PageServicePipeliningConfig::Pipelined(PageServicePipeliningConfigPipelined {
                    max_batch_size: NonZeroUsize::new(32).unwrap(),
                    execution: PageServiceProtocolPipelinedExecutionStrategy::ConcurrentFutures,
-                    batching: PageServiceProtocolPipelinedBatchingStrategy::ScatteredLsn,
                })
            },
            get_vectored_concurrent_io: if !cfg!(test) {
@@ -653,8 +543,6 @@ impl Default for ConfigToml {
            validate_wal_contiguity: None,
            load_previous_heatmap: None,
            generate_unarchival_heatmap: None,
-            tracing: None,
-            enable_tls_page_service_api: false,
        }
    }
 }
@@ -691,6 +579,8 @@ pub mod tenant_conf_defaults {
    pub const DEFAULT_COMPACTION_ALGORITHM: crate::models::CompactionAlgorithm =
        crate::models::CompactionAlgorithm::Legacy;

+    pub const DEFAULT_L0_FLUSH_WAIT_UPLOAD: bool = false;
+
    pub const DEFAULT_GC_HORIZON: u64 = 64 * 1024 * 1024;

    // Large DEFAULT_GC_PERIOD is fine as long as PITR_INTERVAL is larger.
@@ -714,7 +604,6 @@ pub mod tenant_conf_defaults {
    // image layers should be created.
    pub const DEFAULT_IMAGE_LAYER_CREATION_CHECK_THRESHOLD: u8 = 2;
    pub const DEFAULT_GC_COMPACTION_ENABLED: bool = false;
-    pub const DEFAULT_GC_COMPACTION_VERIFICATION: bool = true;
    pub const DEFAULT_GC_COMPACTION_INITIAL_THRESHOLD_KB: u64 = 5 * 1024 * 1024; // 5GB
    pub const DEFAULT_GC_COMPACTION_RATIO_PERCENT: u64 = 100;
 }
@@ -738,6 +627,7 @@ impl Default for TenantConfigToml {
            compaction_l0_semaphore: DEFAULT_COMPACTION_L0_SEMAPHORE,
            l0_flush_delay_threshold: None,
            l0_flush_stall_threshold: None,
+            l0_flush_wait_upload: DEFAULT_L0_FLUSH_WAIT_UPLOAD,
            gc_horizon: DEFAULT_GC_HORIZON,
            gc_period: humantime::parse_duration(DEFAULT_GC_PERIOD)
                .expect("cannot parse default gc period"),
@@ -769,10 +659,8 @@ impl Default for TenantConfigToml {
            wal_receiver_protocol_override: None,
            rel_size_v2_enabled: false,
            gc_compaction_enabled: DEFAULT_GC_COMPACTION_ENABLED,
-            gc_compaction_verification: DEFAULT_GC_COMPACTION_VERIFICATION,
            gc_compaction_initial_threshold_kb: DEFAULT_GC_COMPACTION_INITIAL_THRESHOLD_KB,
            gc_compaction_ratio_percent: DEFAULT_GC_COMPACTION_RATIO_PERCENT,
-            sampling_ratio: None,
        }
    }
 }
--- a/libs/pageserver_api/src/controller_api.rs
+++ b/libs/pageserver_api/src/controller_api.rs
@@ -7,8 +7,7 @@ use std::time::{Duration, Instant};
 /// API (`/control/v1` prefix).  Implemented by the server
 /// in [`storage_controller::http`]
 use serde::{Deserialize, Serialize};
-use utils::id::{NodeId, TenantId, TimelineId};
-use utils::lsn::Lsn;
+use utils::id::{NodeId, TenantId};

 use crate::models::{PageserverUtilization, ShardParameters, TenantConfig};
 use crate::shard::{ShardStripeSize, TenantShardId};
@@ -500,15 +499,6 @@ pub struct SafekeeperSchedulingPolicyRequest {
    pub scheduling_policy: SkSchedulingPolicy,
 }

-/// Import request for safekeeper timelines.
-#[derive(Serialize, Deserialize, Clone)]
-pub struct TimelineImportRequest {
-    pub tenant_id: TenantId,
-    pub timeline_id: TimelineId,
-    pub start_lsn: Lsn,
-    pub sk_set: Vec<NodeId>,
-}
-
 #[cfg(test)]
 mod test {
    use serde_json;
--- a/libs/pageserver_api/src/key.rs
+++ b/libs/pageserver_api/src/key.rs
@@ -927,7 +927,7 @@ impl Key {

    /// Guaranteed to return `Ok()` if [`Self::is_rel_block_key`] returns `true` for `key`.
    #[inline(always)]
-    pub fn to_rel_block(self) -> Result<(RelTag, BlockNumber), ToRelBlockError> {
+    pub fn to_rel_block(self) -> anyhow::Result<(RelTag, BlockNumber)> {
        Ok(match self.field1 {
            0x00 => (
                RelTag {
@@ -938,7 +938,7 @@ impl Key {
                },
                self.field6,
            ),
-            _ => return Err(ToRelBlockError(self.field1)),
+            _ => anyhow::bail!("unexpected value kind 0x{:02x}", self.field1),
        })
    }
 }
@@ -951,17 +951,6 @@ impl std::str::FromStr for Key {
    }
 }

-#[derive(Debug)]
-pub struct ToRelBlockError(u8);
-
-impl fmt::Display for ToRelBlockError {
-    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
-        write!(f, "unexpected value kind 0x{:02x}", self.0)
-    }
-}
-
-impl std::error::Error for ToRelBlockError {}
-
 #[cfg(test)]
 mod tests {
    use std::str::FromStr;
--- a/libs/pageserver_api/src/keyspace.rs
+++ b/libs/pageserver_api/src/keyspace.rs
@@ -613,7 +613,8 @@ mod tests {
    use rand::{RngCore, SeedableRng};

    use super::*;
-    use crate::shard::{DEFAULT_STRIPE_SIZE, ShardCount, ShardNumber, ShardStripeSize};
+    use crate::models::ShardParameters;
+    use crate::shard::{ShardCount, ShardNumber};

    // Helper function to create a key range.
    //
@@ -963,8 +964,12 @@ mod tests {
    }
    #[test]
    fn sharded_range_relation_gap() {
-        let shard_identity =
-            ShardIdentity::new(ShardNumber(0), ShardCount::new(4), DEFAULT_STRIPE_SIZE).unwrap();
+        let shard_identity = ShardIdentity::new(
+            ShardNumber(0),
+            ShardCount::new(4),
+            ShardParameters::DEFAULT_STRIPE_SIZE,
+        )
+        .unwrap();

        let range = ShardedRange::new(
            Range {
@@ -980,8 +985,12 @@ mod tests {

    #[test]
    fn shard_identity_keyspaces_single_key() {
-        let shard_identity =
-            ShardIdentity::new(ShardNumber(1), ShardCount::new(4), DEFAULT_STRIPE_SIZE).unwrap();
+        let shard_identity = ShardIdentity::new(
+            ShardNumber(1),
+            ShardCount::new(4),
+            ShardParameters::DEFAULT_STRIPE_SIZE,
+        )
+        .unwrap();

        let range = ShardedRange::new(
            Range {
@@ -1025,8 +1034,12 @@ mod tests {

    #[test]
    fn shard_identity_keyspaces_forkno_gap() {
-        let shard_identity =
-            ShardIdentity::new(ShardNumber(1), ShardCount::new(4), DEFAULT_STRIPE_SIZE).unwrap();
+        let shard_identity = ShardIdentity::new(
+            ShardNumber(1),
+            ShardCount::new(4),
+            ShardParameters::DEFAULT_STRIPE_SIZE,
+        )
+        .unwrap();

        let range = ShardedRange::new(
            Range {
@@ -1048,7 +1061,7 @@ mod tests {
            let shard_identity = ShardIdentity::new(
                ShardNumber(shard_number),
                ShardCount::new(4),
-                DEFAULT_STRIPE_SIZE,
+                ShardParameters::DEFAULT_STRIPE_SIZE,
            )
            .unwrap();

@@ -1131,44 +1144,37 @@ mod tests {
    /// for a single tenant.
    #[test]
    fn sharded_range_fragment_simple() {
-        const SHARD_COUNT: u8 = 4;
-        const STRIPE_SIZE: u32 = DEFAULT_STRIPE_SIZE.0;
-
        let shard_identity = ShardIdentity::new(
            ShardNumber(0),
-            ShardCount::new(SHARD_COUNT),
-            ShardStripeSize(STRIPE_SIZE),
+            ShardCount::new(4),
+            ShardParameters::DEFAULT_STRIPE_SIZE,
        )
        .unwrap();

        // A range which we happen to know covers exactly one stripe which belongs to this shard
        let input_start = Key::from_hex("000000067f00000001000000ae0000000000").unwrap();
-        let mut input_end = input_start;
-        input_end.field6 += STRIPE_SIZE; // field6 is block number
+        let input_end = Key::from_hex("000000067f00000001000000ae0000008000").unwrap();

        // Ask for stripe_size blocks, we get the whole stripe
        assert_eq!(
-            do_fragment(input_start, input_end, &shard_identity, STRIPE_SIZE),
-            (STRIPE_SIZE, vec![(STRIPE_SIZE, input_start..input_end)])
+            do_fragment(input_start, input_end, &shard_identity, 32768),
+            (32768, vec![(32768, input_start..input_end)])
        );

        // Ask for more, we still get the whole stripe
        assert_eq!(
-            do_fragment(input_start, input_end, &shard_identity, 10 * STRIPE_SIZE),
-            (STRIPE_SIZE, vec![(STRIPE_SIZE, input_start..input_end)])
+            do_fragment(input_start, input_end, &shard_identity, 10000000),
+            (32768, vec![(32768, input_start..input_end)])
        );

        // Ask for target_nblocks of half the stripe size, we get two halves
        assert_eq!(
-            do_fragment(input_start, input_end, &shard_identity, STRIPE_SIZE / 2),
+            do_fragment(input_start, input_end, &shard_identity, 16384),
            (
-                STRIPE_SIZE,
+                32768,
                vec![
-                    (
-                        STRIPE_SIZE / 2,
-                        input_start..input_start.add(STRIPE_SIZE / 2)
-                    ),
-                    (STRIPE_SIZE / 2, input_start.add(STRIPE_SIZE / 2)..input_end)
+                    (16384, input_start..input_start.add(16384)),
+                    (16384, input_start.add(16384)..input_end)
                ]
            )
        );
@@ -1176,53 +1182,40 @@ mod tests {

    #[test]
    fn sharded_range_fragment_multi_stripe() {
-        const SHARD_COUNT: u8 = 4;
-        const STRIPE_SIZE: u32 = DEFAULT_STRIPE_SIZE.0;
-        const RANGE_SIZE: u32 = SHARD_COUNT as u32 * STRIPE_SIZE;
-
        let shard_identity = ShardIdentity::new(
            ShardNumber(0),
-            ShardCount::new(SHARD_COUNT),
-            ShardStripeSize(STRIPE_SIZE),
+            ShardCount::new(4),
+            ShardParameters::DEFAULT_STRIPE_SIZE,
        )
        .unwrap();

        // A range which covers multiple stripes, exactly one of which belongs to the current shard.
        let input_start = Key::from_hex("000000067f00000001000000ae0000000000").unwrap();
-        let mut input_end = input_start;
-        input_end.field6 += RANGE_SIZE; // field6 is block number
-
+        let input_end = Key::from_hex("000000067f00000001000000ae0000020000").unwrap();
        // Ask for all the blocks, get a fragment that covers the whole range but reports
        // its size to be just the blocks belonging to our shard.
        assert_eq!(
-            do_fragment(input_start, input_end, &shard_identity, RANGE_SIZE),
-            (STRIPE_SIZE, vec![(STRIPE_SIZE, input_start..input_end)])
+            do_fragment(input_start, input_end, &shard_identity, 131072),
+            (32768, vec![(32768, input_start..input_end)])
        );

-        // Ask for a sub-stripe quantity that results in 3 fragments.
-        let limit = STRIPE_SIZE / 3 + 1;
+        // Ask for a sub-stripe quantity
        assert_eq!(
-            do_fragment(input_start, input_end, &shard_identity, limit),
+            do_fragment(input_start, input_end, &shard_identity, 16000),
            (
-                STRIPE_SIZE,
+                32768,
                vec![
-                    (limit, input_start..input_start.add(limit)),
-                    (limit, input_start.add(limit)..input_start.add(2 * limit)),
-                    (
-                        STRIPE_SIZE - 2 * limit,
-                        input_start.add(2 * limit)..input_end
-                    ),
+                    (16000, input_start..input_start.add(16000)),
+                    (16000, input_start.add(16000)..input_start.add(32000)),
+                    (768, input_start.add(32000)..input_end),
                ]
            )
        );

        // Try on a range that starts slightly after our owned stripe
        assert_eq!(
-            do_fragment(input_start.add(1), input_end, &shard_identity, RANGE_SIZE),
-            (
-                STRIPE_SIZE - 1,
-                vec![(STRIPE_SIZE - 1, input_start.add(1)..input_end)]
-            )
+            do_fragment(input_start.add(1), input_end, &shard_identity, 131072),
+            (32767, vec![(32767, input_start.add(1)..input_end)])
        );
    }

@@ -1230,40 +1223,32 @@ mod tests {
    /// a previous relation.
    #[test]
    fn sharded_range_fragment_starting_from_logical_size() {
-        const SHARD_COUNT: u8 = 4;
-        const STRIPE_SIZE: u32 = DEFAULT_STRIPE_SIZE.0;
-        const RANGE_SIZE: u32 = SHARD_COUNT as u32 * STRIPE_SIZE;
-
        let input_start = Key::from_hex("000000067f00000001000000ae00ffffffff").unwrap();
-        let mut input_end = Key::from_hex("000000067f00000001000000ae0100000000").unwrap();
-        input_end.field6 += RANGE_SIZE; // field6 is block number
+        let input_end = Key::from_hex("000000067f00000001000000ae0100008000").unwrap();

        // Shard 0 owns the first stripe in the relation, and the preceding logical size is shard local too
        let shard_identity = ShardIdentity::new(
            ShardNumber(0),
-            ShardCount::new(SHARD_COUNT),
-            ShardStripeSize(STRIPE_SIZE),
+            ShardCount::new(4),
+            ShardParameters::DEFAULT_STRIPE_SIZE,
        )
        .unwrap();
        assert_eq!(
-            do_fragment(input_start, input_end, &shard_identity, 2 * STRIPE_SIZE),
-            (
-                STRIPE_SIZE + 1,
-                vec![(STRIPE_SIZE + 1, input_start..input_end)]
-            )
+            do_fragment(input_start, input_end, &shard_identity, 0x10000),
+            (0x8001, vec![(0x8001, input_start..input_end)])
        );

        // Shard 1 does not own the first stripe in the relation, but it does own the logical size (all shards
        // store all logical sizes)
        let shard_identity = ShardIdentity::new(
            ShardNumber(1),
-            ShardCount::new(SHARD_COUNT),
-            ShardStripeSize(STRIPE_SIZE),
+            ShardCount::new(4),
+            ShardParameters::DEFAULT_STRIPE_SIZE,
        )
        .unwrap();
        assert_eq!(
-            do_fragment(input_start, input_end, &shard_identity, 2 * STRIPE_SIZE),
-            (1, vec![(1, input_start..input_end)])
+            do_fragment(input_start, input_end, &shard_identity, 0x10000),
+            (0x1, vec![(0x1, input_start..input_end)])
        );
    }

@@ -1299,8 +1284,12 @@ mod tests {
        );

        // Same, but using a sharded identity
-        let shard_identity =
-            ShardIdentity::new(ShardNumber(0), ShardCount::new(4), DEFAULT_STRIPE_SIZE).unwrap();
+        let shard_identity = ShardIdentity::new(
+            ShardNumber(0),
+            ShardCount::new(4),
+            ShardParameters::DEFAULT_STRIPE_SIZE,
+        )
+        .unwrap();
        assert_eq!(
            do_fragment(input_start, input_end, &shard_identity, 0x8000),
            (u32::MAX, vec![(u32::MAX, input_start..input_end),])
@@ -1342,7 +1331,7 @@ mod tests {
                ShardIdentity::new(
                    ShardNumber((prng.next_u32() % shard_count) as u8),
                    ShardCount::new(shard_count as u8),
-                    DEFAULT_STRIPE_SIZE,
+                    ShardParameters::DEFAULT_STRIPE_SIZE,
                )
                .unwrap()
            };
--- a/libs/pageserver_api/src/models.rs
+++ b/libs/pageserver_api/src/models.rs
@@ -23,10 +23,9 @@ use utils::lsn::Lsn;
 use utils::postgres_client::PostgresClientProtocol;
 use utils::{completion, serde_system_time};

-use crate::config::Ratio;
 use crate::key::{CompactKey, Key};
 use crate::reltag::RelTag;
-use crate::shard::{DEFAULT_STRIPE_SIZE, ShardCount, ShardStripeSize, TenantShardId};
+use crate::shard::{ShardCount, ShardStripeSize, TenantShardId};

 /// The state of a tenant in this pageserver.
 ///
@@ -80,22 +79,10 @@ pub enum TenantState {
    ///
    /// Transitions out of this state are possible through `set_broken()`.
    Stopping {
-        /// The barrier can be used to wait for shutdown to complete. The first caller to set
-        /// Some(Barrier) is responsible for driving shutdown to completion. Subsequent callers
-        /// will wait for the first caller's existing barrier.
-        ///
-        /// None is set when an attach is cancelled, to signal to shutdown that the attach has in
-        /// fact cancelled:
-        ///
-        /// 1. `shutdown` sees `TenantState::Attaching`, and cancels the tenant.
-        /// 2. `attach` sets `TenantState::Stopping(None)` and exits.
-        /// 3. `set_stopping` waits for `TenantState::Stopping(None)` and sets
-        ///    `TenantState::Stopping(Some)` to claim the barrier as the shutdown owner.
-        //
        // Because of https://github.com/serde-rs/serde/issues/2105 this has to be a named field,
        // otherwise it will not be skipped during deserialization
        #[serde(skip)]
-        progress: Option<completion::Barrier>,
+        progress: completion::Barrier,
    },
    /// The tenant is recognized by the pageserver, but can no longer be used for
    /// any operations.
@@ -438,6 +425,8 @@ pub struct ShardParameters {
 }

 impl ShardParameters {
+    pub const DEFAULT_STRIPE_SIZE: ShardStripeSize = ShardStripeSize(256 * 1024 / 8);
+
    pub fn is_unsharded(&self) -> bool {
        self.count.is_unsharded()
    }
@@ -447,7 +436,7 @@ impl Default for ShardParameters {
    fn default() -> Self {
        Self {
            count: ShardCount::new(0),
-            stripe_size: DEFAULT_STRIPE_SIZE,
+            stripe_size: Self::DEFAULT_STRIPE_SIZE,
        }
    }
 }
@@ -534,6 +523,8 @@ pub struct TenantConfigPatch {
    #[serde(skip_serializing_if = "FieldPatch::is_noop")]
    pub l0_flush_stall_threshold: FieldPatch<usize>,
    #[serde(skip_serializing_if = "FieldPatch::is_noop")]
+    pub l0_flush_wait_upload: FieldPatch<bool>,
+    #[serde(skip_serializing_if = "FieldPatch::is_noop")]
    pub gc_horizon: FieldPatch<u64>,
    #[serde(skip_serializing_if = "FieldPatch::is_noop")]
    pub gc_period: FieldPatch<String>,
@@ -576,13 +567,9 @@ pub struct TenantConfigPatch {
    #[serde(skip_serializing_if = "FieldPatch::is_noop")]
    pub gc_compaction_enabled: FieldPatch<bool>,
    #[serde(skip_serializing_if = "FieldPatch::is_noop")]
-    pub gc_compaction_verification: FieldPatch<bool>,
-    #[serde(skip_serializing_if = "FieldPatch::is_noop")]
    pub gc_compaction_initial_threshold_kb: FieldPatch<u64>,
    #[serde(skip_serializing_if = "FieldPatch::is_noop")]
    pub gc_compaction_ratio_percent: FieldPatch<u64>,
-    #[serde(skip_serializing_if = "FieldPatch::is_noop")]
-    pub sampling_ratio: FieldPatch<Option<Ratio>>,
 }

 /// Like [`crate::config::TenantConfigToml`], but preserves the information
@@ -627,6 +614,9 @@ pub struct TenantConfig {
    #[serde(skip_serializing_if = "Option::is_none")]
    pub l0_flush_stall_threshold: Option<usize>,

+    #[serde(skip_serializing_if = "Option::is_none")]
+    pub l0_flush_wait_upload: Option<bool>,
+
    #[serde(skip_serializing_if = "Option::is_none")]
    pub gc_horizon: Option<u64>,

@@ -698,17 +688,11 @@ pub struct TenantConfig {
    #[serde(skip_serializing_if = "Option::is_none")]
    pub gc_compaction_enabled: Option<bool>,

-    #[serde(skip_serializing_if = "Option::is_none")]
-    pub gc_compaction_verification: Option<bool>,
-
    #[serde(skip_serializing_if = "Option::is_none")]
    pub gc_compaction_initial_threshold_kb: Option<u64>,

    #[serde(skip_serializing_if = "Option::is_none")]
    pub gc_compaction_ratio_percent: Option<u64>,
-
-    #[serde(skip_serializing_if = "Option::is_none")]
-    pub sampling_ratio: Option<Option<Ratio>>,
 }

 impl TenantConfig {
@@ -728,6 +712,7 @@ impl TenantConfig {
            mut compaction_l0_semaphore,
            mut l0_flush_delay_threshold,
            mut l0_flush_stall_threshold,
+            mut l0_flush_wait_upload,
            mut gc_horizon,
            mut gc_period,
            mut image_creation_threshold,
@@ -749,10 +734,8 @@ impl TenantConfig {
            mut wal_receiver_protocol_override,
            mut rel_size_v2_enabled,
            mut gc_compaction_enabled,
-            mut gc_compaction_verification,
            mut gc_compaction_initial_threshold_kb,
            mut gc_compaction_ratio_percent,
-            mut sampling_ratio,
        } = self;

        patch.checkpoint_distance.apply(&mut checkpoint_distance);
@@ -782,6 +765,7 @@ impl TenantConfig {
        patch
            .l0_flush_stall_threshold
            .apply(&mut l0_flush_stall_threshold);
+        patch.l0_flush_wait_upload.apply(&mut l0_flush_wait_upload);
        patch.gc_horizon.apply(&mut gc_horizon);
        patch
            .gc_period
@@ -841,16 +825,12 @@ impl TenantConfig {
        patch
            .gc_compaction_enabled
            .apply(&mut gc_compaction_enabled);
-        patch
-            .gc_compaction_verification
-            .apply(&mut gc_compaction_verification);
        patch
            .gc_compaction_initial_threshold_kb
            .apply(&mut gc_compaction_initial_threshold_kb);
        patch
            .gc_compaction_ratio_percent
            .apply(&mut gc_compaction_ratio_percent);
-        patch.sampling_ratio.apply(&mut sampling_ratio);

        Ok(Self {
            checkpoint_distance,
@@ -864,6 +844,7 @@ impl TenantConfig {
            compaction_l0_semaphore,
            l0_flush_delay_threshold,
            l0_flush_stall_threshold,
+            l0_flush_wait_upload,
            gc_horizon,
            gc_period,
            image_creation_threshold,
@@ -885,10 +866,8 @@ impl TenantConfig {
            wal_receiver_protocol_override,
            rel_size_v2_enabled,
            gc_compaction_enabled,
-            gc_compaction_verification,
            gc_compaction_initial_threshold_kb,
            gc_compaction_ratio_percent,
-            sampling_ratio,
        })
    }

@@ -932,6 +911,9 @@ impl TenantConfig {
            l0_flush_stall_threshold: self
                .l0_flush_stall_threshold
                .or(global_conf.l0_flush_stall_threshold),
+            l0_flush_wait_upload: self
+                .l0_flush_wait_upload
+                .unwrap_or(global_conf.l0_flush_wait_upload),
            gc_horizon: self.gc_horizon.unwrap_or(global_conf.gc_horizon),
            gc_period: self.gc_period.unwrap_or(global_conf.gc_period),
            image_creation_threshold: self
@@ -984,16 +966,12 @@ impl TenantConfig {
            gc_compaction_enabled: self
                .gc_compaction_enabled
                .unwrap_or(global_conf.gc_compaction_enabled),
-            gc_compaction_verification: self
-                .gc_compaction_verification
-                .unwrap_or(global_conf.gc_compaction_verification),
            gc_compaction_initial_threshold_kb: self
                .gc_compaction_initial_threshold_kb
                .unwrap_or(global_conf.gc_compaction_initial_threshold_kb),
            gc_compaction_ratio_percent: self
                .gc_compaction_ratio_percent
                .unwrap_or(global_conf.gc_compaction_ratio_percent),
-            sampling_ratio: self.sampling_ratio.unwrap_or(global_conf.sampling_ratio),
        }
    }
 }
@@ -1127,7 +1105,7 @@ pub struct CompactionAlgorithmSettings {
 }

 #[derive(Debug, PartialEq, Eq, Clone, Deserialize, Serialize)]
-#[serde(tag = "mode", rename_all = "kebab-case")]
+#[serde(tag = "mode", rename_all = "kebab-case", deny_unknown_fields)]
 pub enum L0FlushConfig {
    #[serde(rename_all = "snake_case")]
    Direct { max_concurrency: NonZeroUsize },
@@ -1451,6 +1429,11 @@ pub struct TimelineInfo {
    pub last_record_lsn: Lsn,
    pub prev_record_lsn: Option<Lsn>,

+    /// Legacy field, retained for one version to enable old storage controller to
+    /// decode (it was a mandatory field).
+    #[serde(default, rename = "latest_gc_cutoff_lsn")]
+    pub _unused: Lsn,
+
    /// The LSN up to which GC has advanced: older data may still exist but it is not available for clients.
    /// This LSN is not suitable for deciding where to create branches etc: use [`TimelineInfo::min_readable_lsn`] instead,
    /// as it is easier to reason about.
@@ -1691,7 +1674,6 @@ pub struct SecondaryProgress {
 pub struct TenantScanRemoteStorageShard {
    pub tenant_shard_id: TenantShardId,
    pub generation: Option<u32>,
-    pub stripe_size: Option<ShardStripeSize>,
 }

 #[derive(Serialize, Deserialize, Debug, Default)]
@@ -2743,15 +2725,10 @@ mod tests {
                "Activating",
            ),
            (line!(), TenantState::Active, "Active"),
-            (
-                line!(),
-                TenantState::Stopping { progress: None },
-                "Stopping",
-            ),
            (
                line!(),
                TenantState::Stopping {
-                    progress: Some(completion::Barrier::default()),
+                    progress: utils::completion::Barrier::default(),
                },
                "Stopping",
            ),
--- a/libs/pageserver_api/src/record.rs
+++ b/libs/pageserver_api/src/record.rs
@@ -58,8 +58,6 @@ pub enum NeonWalRecord {
        /// to true. This record does not need the history WALs to reconstruct. See [`NeonWalRecord::will_init`] and
        /// its references in `timeline.rs`.
        will_init: bool,
-        /// Only append the record if the current image is the same as the one specified in this field.
-        only_if: Option<String>,
    },
 }

@@ -83,17 +81,6 @@ impl NeonWalRecord {
            append: s.as_ref().to_string(),
            clear: false,
            will_init: false,
-            only_if: None,
-        }
-    }
-
-    #[cfg(feature = "testing")]
-    pub fn wal_append_conditional(s: impl AsRef<str>, only_if: impl AsRef<str>) -> Self {
-        Self::Test {
-            append: s.as_ref().to_string(),
-            clear: false,
-            will_init: false,
-            only_if: Some(only_if.as_ref().to_string()),
        }
    }

@@ -103,7 +90,6 @@ impl NeonWalRecord {
            append: s.as_ref().to_string(),
            clear: true,
            will_init: false,
-            only_if: None,
        }
    }

@@ -113,7 +99,6 @@ impl NeonWalRecord {
            append: s.as_ref().to_string(),
            clear: true,
            will_init: true,
-            only_if: None,
        }
    }
 }
--- a/libs/pageserver_api/src/shard.rs
+++ b/libs/pageserver_api/src/shard.rs
@@ -78,12 +78,6 @@ impl Default for ShardStripeSize {
    }
 }

-impl std::fmt::Display for ShardStripeSize {
-    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
-        self.0.fmt(f)
-    }
-}
-
 /// Layout version: for future upgrades where we might change how the key->shard mapping works
 #[derive(Clone, Copy, Serialize, Deserialize, Eq, PartialEq, Hash, Debug)]
 pub struct ShardLayout(u8);
@@ -92,11 +86,8 @@ const LAYOUT_V1: ShardLayout = ShardLayout(1);
 /// ShardIdentity uses a magic layout value to indicate if it is unusable
 const LAYOUT_BROKEN: ShardLayout = ShardLayout(255);

-/// The default stripe size in pages. 16 MiB divided by 8 kiB page size.
-///
-/// A lower stripe size distributes ingest load better across shards, but reduces IO amortization.
-/// 16 MiB appears to be a reasonable balance: <https://github.com/neondatabase/neon/pull/10510>.
-pub const DEFAULT_STRIPE_SIZE: ShardStripeSize = ShardStripeSize(16 * 1024 / 8);
+/// Default stripe size in pages: 256MiB divided by 8kiB page size.
+const DEFAULT_STRIPE_SIZE: ShardStripeSize = ShardStripeSize(256 * 1024 / 8);

 #[derive(thiserror::Error, Debug, PartialEq, Eq)]
 pub enum ShardConfigError {
@@ -546,7 +537,7 @@ mod tests {
            field6: 0x7d06,
        };

-        let shard = key_to_shard_number(ShardCount(10), ShardStripeSize(32768), &key);
+        let shard = key_to_shard_number(ShardCount(10), DEFAULT_STRIPE_SIZE, &key);
        assert_eq!(shard, ShardNumber(8));
    }

--- a/libs/postgres_backend/src/lib.rs
+++ b/libs/postgres_backend/src/lib.rs
@@ -5,6 +5,7 @@
 #![deny(unsafe_code)]
 #![deny(clippy::undocumented_unsafe_blocks)]
 use std::future::Future;
+use std::io::ErrorKind;
 use std::net::SocketAddr;
 use std::os::fd::{AsRawFd, RawFd};
 use std::pin::Pin;
@@ -226,7 +227,7 @@ impl<IO: AsyncRead + AsyncWrite + Unpin> MaybeWriteOnly<IO> {
        match self {
            MaybeWriteOnly::Full(framed) => framed.read_startup_message().await,
            MaybeWriteOnly::WriteOnly(_) => {
-                Err(io::Error::other("reading from write only half").into())
+                Err(io::Error::new(ErrorKind::Other, "reading from write only half").into())
            }
            MaybeWriteOnly::Broken => panic!("IO on invalid MaybeWriteOnly"),
        }
@@ -236,7 +237,7 @@ impl<IO: AsyncRead + AsyncWrite + Unpin> MaybeWriteOnly<IO> {
        match self {
            MaybeWriteOnly::Full(framed) => framed.read_message().await,
            MaybeWriteOnly::WriteOnly(_) => {
-                Err(io::Error::other("reading from write only half").into())
+                Err(io::Error::new(ErrorKind::Other, "reading from write only half").into())
            }
            MaybeWriteOnly::Broken => panic!("IO on invalid MaybeWriteOnly"),
        }
@@ -974,7 +975,7 @@ impl<IO: AsyncRead + AsyncWrite + Unpin> AsyncWrite for CopyDataWriter<'_, IO> {
            .write_message_noflush(&BeMessage::CopyData(buf))
            // write_message only writes to the buffer, so it can fail iff the
            // message is invaid, but CopyData can't be invalid.
-            .map_err(|_| io::Error::other("failed to serialize CopyData"))?;
+            .map_err(|_| io::Error::new(ErrorKind::Other, "failed to serialize CopyData"))?;

        Poll::Ready(Ok(buf.len()))
    }
--- a/libs/postgres_backend/tests/simple_select.rs
+++ b/libs/postgres_backend/tests/simple_select.rs
@@ -85,8 +85,8 @@ static KEY: Lazy<rustls::pki_types::PrivateKeyDer<'static>> = Lazy::new(|| {

 static CERT: Lazy<rustls::pki_types::CertificateDer<'static>> = Lazy::new(|| {
    let mut cursor = Cursor::new(include_bytes!("cert.pem"));
-
-    rustls_pemfile::certs(&mut cursor).next().unwrap().unwrap()
+    let cert = rustls_pemfile::certs(&mut cursor).next().unwrap().unwrap();
+    cert
 });

 // test that basic select with ssl works
--- a/libs/pq_proto/src/framed.rs
+++ b/libs/pq_proto/src/framed.rs
@@ -35,7 +35,7 @@ impl ConnectionError {
    pub fn into_io_error(self) -> io::Error {
        match self {
            ConnectionError::Io(io) => io,
-            ConnectionError::Protocol(pe) => io::Error::other(pe.to_string()),
+            ConnectionError::Protocol(pe) => io::Error::new(io::ErrorKind::Other, pe.to_string()),
        }
    }
 }
--- a/libs/pq_proto/src/lib.rs
+++ b/libs/pq_proto/src/lib.rs
@@ -257,7 +257,7 @@ pub enum ProtocolError {
 impl ProtocolError {
    /// Proxy stream.rs uses only io::Error; provide it.
    pub fn into_io_error(self) -> io::Error {
-        io::Error::other(self.to_string())
+        io::Error::new(io::ErrorKind::Other, self.to_string())
    }
 }

--- a/libs/proxy/postgres-protocol2/src/authentication/sasl.rs
+++ b/libs/proxy/postgres-protocol2/src/authentication/sasl.rs
@@ -212,7 +212,7 @@ impl ScramSha256 {
                    password,
                    channel_binding,
                } => (nonce, password, channel_binding),
-                _ => return Err(io::Error::other("invalid SCRAM state")),
+                _ => return Err(io::Error::new(io::ErrorKind::Other, "invalid SCRAM state")),
            };

        let message =
@@ -291,7 +291,7 @@ impl ScramSha256 {
                server_key,
                auth_message,
            } => (server_key, auth_message),
-            _ => return Err(io::Error::other("invalid SCRAM state")),
+            _ => return Err(io::Error::new(io::ErrorKind::Other, "invalid SCRAM state")),
        };

        let message =
@@ -301,7 +301,10 @@ impl ScramSha256 {

        let verifier = match parsed {
            ServerFinalMessage::Error(e) => {
-                return Err(io::Error::other(format!("SCRAM error: {}", e)));
+                return Err(io::Error::new(
+                    io::ErrorKind::Other,
+                    format!("SCRAM error: {}", e),
+                ));
            }
            ServerFinalMessage::Verifier(verifier) => verifier,
        };
--- a/libs/proxy/tokio-postgres2/Cargo.toml
+++ b/libs/proxy/tokio-postgres2/Cargo.toml
@@ -8,9 +8,10 @@ license = "MIT/Apache-2.0"
 bytes.workspace = true
 fallible-iterator.workspace = true
 futures-util = { workspace = true, features = ["sink"] }
-tracing.workspace = true
+log = "0.4"
 parking_lot.workspace = true
 pin-project-lite.workspace = true
+phf = "0.11"
 postgres-protocol2 = { path = "../postgres-protocol2" }
 postgres-types2 = { path = "../postgres-types2" }
 tokio = { workspace = true, features = ["io-util", "time", "net"] }
--- a/libs/proxy/tokio-postgres2/src/connection.rs
+++ b/libs/proxy/tokio-postgres2/src/connection.rs
@@ -6,13 +6,13 @@ use std::task::{Context, Poll};
 use bytes::BytesMut;
 use fallible_iterator::FallibleIterator;
 use futures_util::{Sink, Stream, ready};
+use log::{info, trace};
 use postgres_protocol2::message::backend::Message;
 use postgres_protocol2::message::frontend;
 use tokio::io::{AsyncRead, AsyncWrite};
 use tokio::sync::mpsc;
 use tokio_util::codec::Framed;
 use tokio_util::sync::PollSender;
-use tracing::{info, trace};

 use crate::codec::{BackendMessage, BackendMessages, FrontendMessage, PostgresCodec};
 use crate::error::DbError;
--- a/libs/proxy/tokio-postgres2/src/error/sqlstate.rs
+++ b/libs/proxy/tokio-postgres2/src/error/sqlstate.rs
--- a/libs/proxy/tokio-postgres2/src/prepare.rs
+++ b/libs/proxy/tokio-postgres2/src/prepare.rs
@@ -5,9 +5,9 @@ use std::sync::Arc;
 use bytes::Bytes;
 use fallible_iterator::FallibleIterator;
 use futures_util::{TryStreamExt, pin_mut};
+use log::debug;
 use postgres_protocol2::message::backend::Message;
 use postgres_protocol2::message::frontend;
-use tracing::debug;

 use crate::client::{CachedTypeInfo, InnerClient};
 use crate::codec::FrontendMessage;
--- a/libs/proxy/tokio-postgres2/src/query.rs
+++ b/libs/proxy/tokio-postgres2/src/query.rs
@@ -7,11 +7,11 @@ use std::task::{Context, Poll};
 use bytes::{BufMut, Bytes, BytesMut};
 use fallible_iterator::FallibleIterator;
 use futures_util::{Stream, ready};
+use log::{Level, debug, log_enabled};
 use pin_project_lite::pin_project;
 use postgres_protocol2::message::backend::Message;
 use postgres_protocol2::message::frontend;
 use postgres_types2::{Format, ToSql, Type};
-use tracing::debug;

 use crate::client::{InnerClient, Responses};
 use crate::codec::FrontendMessage;
@@ -36,7 +36,7 @@ where
    I: IntoIterator<Item = &'a (dyn ToSql + Sync)>,
    I::IntoIter: ExactSizeIterator,
 {
-    let buf = if tracing::enabled!(tracing::Level::DEBUG) {
+    let buf = if log_enabled!(Level::Debug) {
        let params = params.into_iter().collect::<Vec<_>>();
        debug!(
            "executing statement {} with parameters: {:?}",
--- a/libs/proxy/tokio-postgres2/src/simple_query.rs
+++ b/libs/proxy/tokio-postgres2/src/simple_query.rs
@@ -6,10 +6,10 @@ use std::task::{Context, Poll};
 use bytes::Bytes;
 use fallible_iterator::FallibleIterator;
 use futures_util::{Stream, ready};
+use log::debug;
 use pin_project_lite::pin_project;
 use postgres_protocol2::message::backend::Message;
 use postgres_protocol2::message::frontend;
-use tracing::debug;

 use crate::client::{InnerClient, Responses};
 use crate::codec::FrontendMessage;
--- a/libs/remote_keys/Cargo.toml
+++ b/libs/remote_keys/Cargo.toml
@@ -1,12 +0,0 @@
-[package]
-name = "remote_keys"
-version = "0.1.0"
-edition = "2024"
-license.workspace = true
-
-[dependencies]
-anyhow.workspace = true
-aws-smithy-types.workspace = true
-aws-sdk-kms.workspace = true
-aws-config.workspace = true
-utils.workspace = true
--- a/libs/remote_keys/src/aws_keys.rs
+++ b/libs/remote_keys/src/aws_keys.rs
@@ -1,47 +0,0 @@
-use aws_config::BehaviorVersion;
-
-use crate::KeyId;
-
-pub struct AwsRemoteKeyClient {
-    client: aws_sdk_kms::Client,
-}
-
-impl AwsRemoteKeyClient {
-    pub async fn new() -> Self {
-        let sdk_config = aws_config::defaults(BehaviorVersion::v2024_03_28())
-            .retry_config(
-                aws_config::retry::RetryConfig::standard()
-                    .with_max_attempts(5) // Retry up to 5 times
-                    .with_initial_backoff(std::time::Duration::from_millis(200)) // Start with 200ms delay
-                    .with_max_backoff(std::time::Duration::from_secs(5)), // Cap at 5 seconds
-            )
-            .load()
-            .await;
-        let client = aws_sdk_kms::Client::new(&sdk_config);
-        Self { client }
-    }
-
-    pub async fn decrypt(&self, key_id: &KeyId, ciphertext: impl Into<Vec<u8>>) -> Vec<u8> {
-        let output = self
-            .client
-            .decrypt()
-            .key_id(&key_id.0)
-            .ciphertext_blob(aws_smithy_types::Blob::new(ciphertext.into()))
-            .send()
-            .await
-            .expect("decrypt");
-        output.plaintext.expect("plaintext").into_inner()
-    }
-
-    pub async fn encrypt(&self, key_id: &KeyId, ciphertext: impl Into<Vec<u8>>) -> Vec<u8> {
-        let output = self
-            .client
-            .encrypt()
-            .key_id(&key_id.0)
-            .plaintext(aws_smithy_types::Blob::new(ciphertext.into()))
-            .send()
-            .await
-            .expect("decrypt");
-        output.ciphertext_blob.expect("ciphertext").into_inner()
-    }
-}
--- a/libs/remote_keys/src/lib.rs
+++ b/libs/remote_keys/src/lib.rs
@@ -1,6 +0,0 @@
-mod aws_keys;
-pub use aws_keys::AwsRemoteKeyClient;
-
-/// A string uniquely identifying a key
-#[derive(Debug, PartialEq, Eq)]
-pub struct KeyId(pub String);
--- a/libs/remote_storage/Cargo.toml
+++ b/libs/remote_storage/Cargo.toml
@@ -28,7 +28,7 @@ toml_edit.workspace = true
 tracing.workspace = true
 scopeguard.workspace = true
 metrics.workspace = true
-utils = { path = "../utils", default-features = false }
+utils.workspace = true
 pin-project-lite.workspace = true

 azure_core.workspace = true
--- a/libs/remote_storage/src/azure_blob.rs
+++ b/libs/remote_storage/src/azure_blob.rs
@@ -801,7 +801,8 @@ where
            // that support needs to be hacked in.
            //
            // including {self:?} into the message would be useful, but unsure how to unproject.
-            _ => std::task::Poll::Ready(Err(std::io::Error::other(
+            _ => std::task::Poll::Ready(Err(std::io::Error::new(
+                std::io::ErrorKind::Other,
                "cloned or initial values cannot be read",
            ))),
        }
@@ -854,7 +855,7 @@ where
        };
        Err(azure_core::error::Error::new(
            azure_core::error::ErrorKind::Io,
-            std::io::Error::other(msg),
+            std::io::Error::new(std::io::ErrorKind::Other, msg),
        ))
    }

--- a/libs/remote_storage/tests/test_real_s3.rs
+++ b/libs/remote_storage/tests/test_real_s3.rs
@@ -558,7 +558,7 @@ async fn upload_large_enough_file(
 ) -> usize {
    let header = bytes::Bytes::from_static("remote blob data content".as_bytes());
    let body = bytes::Bytes::from(vec![0u8; 1024]);
-    let contents = std::iter::once(header).chain(std::iter::repeat_n(body, 128));
+    let contents = std::iter::once(header).chain(std::iter::repeat(body).take(128));

    let len = contents.clone().fold(0, |acc, next| acc + next.len());

--- a/libs/safekeeper_api/src/models.rs
+++ b/libs/safekeeper_api/src/models.rs
@@ -71,7 +71,6 @@ pub struct PeerInfo {
    pub ts: Instant,
    pub pg_connstr: String,
    pub http_connstr: String,
-    pub https_connstr: Option<String>,
 }

 pub type FullTransactionId = u64;
@@ -228,8 +227,6 @@ pub struct TimelineDeleteResult {
    pub dir_existed: bool,
 }

-pub type TenantDeleteResult = std::collections::HashMap<String, TimelineDeleteResult>;
-
 fn lsn_invalid() -> Lsn {
    Lsn::INVALID
 }
@@ -262,8 +259,6 @@ pub struct SkTimelineInfo {
    pub safekeeper_connstr: Option<String>,
    #[serde(default)]
    pub http_connstr: Option<String>,
-    #[serde(default)]
-    pub https_connstr: Option<String>,
    // Minimum of all active RO replicas flush LSN
    #[serde(default = "lsn_invalid")]
    pub standby_horizon: Lsn,
--- a/libs/tracing-utils/Cargo.toml
+++ b/libs/tracing-utils/Cargo.toml
@@ -14,7 +14,6 @@ tokio = { workspace = true, features = ["rt", "rt-multi-thread"] }
 tracing.workspace = true
 tracing-opentelemetry.workspace = true
 tracing-subscriber.workspace = true
-pin-project-lite.workspace = true

 [dev-dependencies]
 tracing-subscriber.workspace = true    # For examples in docs
--- a/libs/tracing-utils/src/lib.rs
+++ b/libs/tracing-utils/src/lib.rs
@@ -31,10 +31,10 @@
 //!         .init();
 //! }
 //! ```
+#![deny(unsafe_code)]
 #![deny(clippy::undocumented_unsafe_blocks)]

 pub mod http;
-pub mod perf_span;

 use opentelemetry::KeyValue;
 use opentelemetry::trace::TracerProvider;
--- a/libs/tracing-utils/src/perf_span.rs
+++ b/libs/tracing-utils/src/perf_span.rs
@@ -1,144 +0,0 @@
-//! Crutch module to work around tracing infrastructure deficiencies
-//!
-//! We wish to collect granular request spans without impacting performance
-//! by much. Ideally, we should have zero overhead for a sampling rate of 0.
-//!
-//! The approach taken by the pageserver crate is to use a completely different
-//! span hierarchy for the performance spans. Spans are explicitly stored in
-//! the request context and use a different [`tracing::Subscriber`] in order
-//! to avoid expensive filtering.
-//!
-//! [`tracing::Span`] instances record their [`tracing::Dispatch`] and, implcitly,
-//! their [`tracing::Subscriber`] at creation time. However, upon exiting the span,
-//! the global default [`tracing::Dispatch`] is used. This is problematic if one
-//! wishes to juggle different subscribers.
-//!
-//! In order to work around this, this module provides a [`PerfSpan`] type which
-//! wraps a [`Span`] and sets the default subscriber when exiting the span. This
-//! achieves the correct routing.
-//!
-//! There's also a modified version of [`tracing::Instrument`] which works with
-//! [`PerfSpan`].
-
-use core::{
-    future::Future,
-    marker::Sized,
-    mem::ManuallyDrop,
-    pin::Pin,
-    task::{Context, Poll},
-};
-use pin_project_lite::pin_project;
-use tracing::{Dispatch, span::Span};
-
-#[derive(Debug, Clone)]
-pub struct PerfSpan {
-    inner: ManuallyDrop<Span>,
-    dispatch: Dispatch,
-}
-
-#[must_use = "once a span has been entered, it should be exited"]
-pub struct PerfSpanEntered<'a> {
-    span: &'a PerfSpan,
-}
-
-impl PerfSpan {
-    pub fn new(span: Span, dispatch: Dispatch) -> Self {
-        Self {
-            inner: ManuallyDrop::new(span),
-            dispatch,
-        }
-    }
-
-    pub fn enter(&self) -> PerfSpanEntered {
-        if let Some(ref id) = self.inner.id() {
-            self.dispatch.enter(id);
-        }
-
-        PerfSpanEntered { span: self }
-    }
-
-    pub fn inner(&self) -> &Span {
-        &self.inner
-    }
-}
-
-impl Drop for PerfSpan {
-    fn drop(&mut self) {
-        // Bring the desired dispatch into scope before explicitly calling
-        // the span destructor. This routes the span exit to the correct
-        // [`tracing::Subscriber`].
-        let _dispatch_guard = tracing::dispatcher::set_default(&self.dispatch);
-        // SAFETY: ManuallyDrop in Drop implementation
-        unsafe { ManuallyDrop::drop(&mut self.inner) }
-    }
-}
-
-impl Drop for PerfSpanEntered<'_> {
-    fn drop(&mut self) {
-        assert!(self.span.inner.id().is_some());
-
-        let _dispatch_guard = tracing::dispatcher::set_default(&self.span.dispatch);
-        self.span.dispatch.exit(&self.span.inner.id().unwrap());
-    }
-}
-
-pub trait PerfInstrument: Sized {
-    fn instrument(self, span: PerfSpan) -> PerfInstrumented<Self> {
-        PerfInstrumented {
-            inner: ManuallyDrop::new(self),
-            span,
-        }
-    }
-}
-
-pin_project! {
-    #[project = PerfInstrumentedProj]
-    #[derive(Debug, Clone)]
-    #[must_use = "futures do nothing unless you `.await` or poll them"]
-    pub struct PerfInstrumented<T> {
-        // `ManuallyDrop` is used here to to enter instrument `Drop` by entering
-        // `Span` and executing `ManuallyDrop::drop`.
-        #[pin]
-        inner: ManuallyDrop<T>,
-        span: PerfSpan,
-    }
-
-    impl<T> PinnedDrop for PerfInstrumented<T> {
-        fn drop(this: Pin<&mut Self>) {
-            let this = this.project();
-            let _enter = this.span.enter();
-            // SAFETY: 1. `Pin::get_unchecked_mut()` is safe, because this isn't
-            //             different from wrapping `T` in `Option` and calling
-            //             `Pin::set(&mut this.inner, None)`, except avoiding
-            //             additional memory overhead.
-            //         2. `ManuallyDrop::drop()` is safe, because
-            //            `PinnedDrop::drop()` is guaranteed to be called only
-            //            once.
-            unsafe { ManuallyDrop::drop(this.inner.get_unchecked_mut()) }
-        }
-    }
-}
-
-impl<'a, T> PerfInstrumentedProj<'a, T> {
-    /// Get a mutable reference to the [`Span`] a pinned mutable reference to
-    /// the wrapped type.
-    fn span_and_inner_pin_mut(self) -> (&'a mut PerfSpan, Pin<&'a mut T>) {
-        // SAFETY: As long as `ManuallyDrop<T>` does not move, `T` won't move
-        //         and `inner` is valid, because `ManuallyDrop::drop` is called
-        //         only inside `Drop` of the `Instrumented`.
-        let inner = unsafe { self.inner.map_unchecked_mut(|v| &mut **v) };
-        (self.span, inner)
-    }
-}
-
-impl<T: Future> Future for PerfInstrumented<T> {
-    type Output = T::Output;
-
-    fn poll(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Self::Output> {
-        let (span, inner) = self.project().span_and_inner_pin_mut();
-        let _enter = span.enter();
-        inner.poll(cx)
-    }
-}
-
-impl<T: Sized> PerfInstrument for T {}
--- a/libs/utils/Cargo.toml
+++ b/libs/utils/Cargo.toml
@@ -5,8 +5,7 @@ edition.workspace = true
 license.workspace = true

 [features]
-default = ["rename_noreplace"]
-rename_noreplace = []
+default = []
 # Enables test-only APIs, incuding failpoints. In particular, enables the `fail_point!` macro,
 # which adds some runtime cost to run tests on outage conditions
 testing = ["fail/failpoints"]
@@ -36,7 +35,7 @@ serde_with.workspace = true
 serde_json.workspace = true
 signal-hook.workspace = true
 thiserror.workspace = true
-tokio = { workspace = true, features = ["signal"] }
+tokio.workspace = true
 tokio-tar.workspace = true
 tokio-util.workspace = true
 toml_edit = { workspace = true, features = ["serde"] }
--- a/libs/utils/src/auth.rs
+++ b/libs/utils/src/auth.rs
@@ -173,7 +173,7 @@ impl std::fmt::Debug for JwtAuth {
 }

 // this function is used only for testing purposes in CLI e g generate tokens during init
-pub fn encode_from_key_file<S: Serialize>(claims: &S, key_data: &[u8]) -> Result<String> {
+pub fn encode_from_key_file(claims: &Claims, key_data: &[u8]) -> Result<String> {
    let key = EncodingKey::from_ed_pem(key_data)?;
    Ok(encode(&Header::new(STORAGE_TOKEN_ALGORITHM), claims, &key)?)
 }
--- a/libs/utils/src/crashsafe.rs
+++ b/libs/utils/src/crashsafe.rs
@@ -81,9 +81,12 @@ pub fn path_with_suffix_extension(
 }

 pub fn fsync_file_and_parent(file_path: &Utf8Path) -> io::Result<()> {
-    let parent = file_path
-        .parent()
-        .ok_or_else(|| io::Error::other(format!("File {file_path:?} has no parent")))?;
+    let parent = file_path.parent().ok_or_else(|| {
+        io::Error::new(
+            io::ErrorKind::Other,
+            format!("File {file_path:?} has no parent"),
+        )
+    })?;

    fsync(file_path)?;
    fsync(parent)?;
--- a/libs/utils/src/elapsed_accum.rs
+++ b/libs/utils/src/elapsed_accum.rs
@@ -1,26 +0,0 @@
-use std::time::{Duration, Instant};
-
-#[derive(Default)]
-pub struct ElapsedAccum {
-    accum: Duration,
-}
-
-impl ElapsedAccum {
-    pub fn get(&self) -> Duration {
-        self.accum
-    }
-    pub fn guard(&mut self) -> impl Drop + '_ {
-        let start = Instant::now();
-        scopeguard::guard(start, |last_wait_at| {
-            self.accum += Instant::now() - last_wait_at;
-        })
-    }
-
-    pub async fn measure<Fut, O>(&mut self, fut: Fut) -> O
-    where
-        Fut: Future<Output = O>,
-    {
-        let _guard = self.guard();
-        fut.await
-    }
-}
--- a/libs/utils/src/fs_ext.rs
+++ b/libs/utils/src/fs_ext.rs
@@ -3,9 +3,7 @@ use std::{fs, io, path::Path};

 use anyhow::Context;

-#[cfg(feature = "rename_noreplace")]
 mod rename_noreplace;
-#[cfg(feature = "rename_noreplace")]
 pub use rename_noreplace::rename_noreplace;

 pub trait PathExt {
--- a/libs/utils/src/fs_ext/rename_noreplace.rs
+++ b/libs/utils/src/fs_ext/rename_noreplace.rs
@@ -8,7 +8,7 @@ pub fn rename_noreplace<P1: ?Sized + NixPath, P2: ?Sized + NixPath>(
    dst: &P2,
 ) -> nix::Result<()> {
    {
-        #[cfg(all(target_os = "linux", target_env = "gnu"))]
+        #[cfg(target_os = "linux")]
        {
            nix::fcntl::renameat2(
                None,
@@ -29,7 +29,7 @@ pub fn rename_noreplace<P1: ?Sized + NixPath, P2: ?Sized + NixPath>(
            })??;
            nix::errno::Errno::result(res).map(drop)
        }
-        #[cfg(not(any(all(target_os = "linux", target_env = "gnu"), target_os = "macos")))]
+        #[cfg(not(any(target_os = "linux", target_os = "macos")))]
        {
            std::compile_error!("OS does not support no-replace renames");
        }
--- a/Show More
+++ b/Show More
				`@@ -1 +0,0 @@`
				`SELECT lfc_value AS lfc_used_pages FROM neon.neon_lfc_stats WHERE lfc_key = 'file_cache_used_pages';`