Compute release 2025-02-27

.github/PULL_REQUEST_TEMPLATE/release-pr.md

@@ -0,0 +1,21 @@
+## Release 202Y-MM-DD
+
+**NB: this PR must be merged only by 'Create a merge commit'!**
+
+### Checklist when preparing for release
+- [ ] Read or refresh [the release flow guide](https://www.notion.so/neondatabase/Release-general-flow-61f2e39fd45d4d14a70c7749604bd70b)
+- [ ] Ask in the [cloud Slack channel](https://neondb.slack.com/archives/C033A2WE6BZ) that you are going to rollout the release. Any blockers?
+- [ ] Does this release contain any db migrations? Destructive ones? What is the rollback plan?
+
+<!-- List everything that should be done **before** release, any issues / setting changes / etc -->
+
+### Checklist after release
+- [ ] Make sure instructions from PRs included in this release and labeled `manual_release_instructions` are executed (either by you or by people who wrote them).
+- [ ] Based on the merged commits write release notes and open a PR into `website` repo ([example](https://github.com/neondatabase/website/pull/219/files))
+- [ ] Check [#dev-production-stream](https://neondb.slack.com/archives/C03F5SM1N02) Slack channel
+- [ ] Check [stuck projects page](https://console.neon.tech/admin/projects?sort=last_active&order=desc&stuck=true)
+- [ ] Check [recent operation failures](https://console.neon.tech/admin/operations?action=create_timeline%2Cstart_compute%2Cstop_compute%2Csuspend_compute%2Capply_config%2Cdelete_timeline%2Cdelete_tenant%2Ccreate_branch%2Ccheck_availability&sort=updated_at&order=desc&had_retries=some)
+- [ ] Check [cloud SLO dashboard](https://neonprod.grafana.net/d/_oWcBMJ7k/cloud-slos?orgId=1)
+- [ ] Check [compute startup metrics dashboard](https://neonprod.grafana.net/d/5OkYJEmVz/compute-startup-time)
+
+<!-- List everything that should be done **after** release, any admin UI configuration / Grafana dashboard / alert changes / setting changes / etc -->

.github/actionlint.yml

@@ -32,6 +32,3 @@ config-variables:
   - 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

.github/actions/neon-branch-create/action.yml

@@ -84,13 +84,7 @@ runs:
           --header "Authorization: Bearer ${API_KEY}"
           )
 
-        role_name=$(echo "$roles" | jq --raw-output '
-          (.roles | map(select(.protected == false))) as $roles |
-          if any($roles[]; .name == "neondb_owner")
-          then "neondb_owner"
-          else $roles[0].name
-          end
-        ')
+        role_name=$(echo $roles | jq --raw-output '.roles[] | select(.protected == false) | .name')
         echo "role_name=${role_name}" >> $GITHUB_OUTPUT
       env:
         API_HOST: ${{ inputs.api_host }}
@@ -113,13 +107,13 @@ runs:
             )
 
           if [ -z "${reset_password}" ]; then
-            sleep $i
+            sleep 1
             continue
           fi
 
           password=$(echo $reset_password | jq --raw-output '.role.password')
           if [ "${password}" == "null" ]; then
-            sleep $i # increasing backoff
+            sleep 1
             continue
           fi
 

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

@@ -44,11 +44,6 @@ inputs:
     description: 'Postgres version to use for tests'
     required: false
     default: 'v16'
-  sanitizers:
-    description: 'enabled or disabled'
-    required: false
-    default: 'disabled'
-    type: string
   benchmark_durations:
     description: 'benchmark durations JSON'
     required: false
@@ -64,7 +59,7 @@ runs:
       if: inputs.build_type != 'remote'
       uses: ./.github/actions/download
       with:
-        name: neon-${{ runner.os }}-${{ runner.arch }}-${{ inputs.build_type }}${{ inputs.sanitizers == 'enabled' && '-sanitized' || '' }}-artifact
+        name: neon-${{ runner.os }}-${{ runner.arch }}-${{ inputs.build_type }}-artifact
         path: /tmp/neon
         aws-oicd-role-arn: ${{ inputs.aws-oicd-role-arn }}
 
@@ -117,7 +112,6 @@ runs:
         ALLOW_FORWARD_COMPATIBILITY_BREAKAGE: contains(github.event.pull_request.labels.*.name, 'forward compatibility breakage')
         RERUN_FAILED: ${{ inputs.rerun_failed }}
         PG_VERSION: ${{ inputs.pg_version }}
-        SANITIZERS: ${{ inputs.sanitizers }}
       shell: bash -euxo pipefail {0}
       run: |
         # PLATFORM will be embedded in the perf test report

.github/scripts/generate_image_maps.py

@@ -1,16 +1,14 @@
 import itertools
 import json
 import os
-import sys
 
-source_tag = os.getenv("SOURCE_TAG")
-target_tag = os.getenv("TARGET_TAG")
-branch = os.getenv("BRANCH")
-dev_acr = os.getenv("DEV_ACR")
-prod_acr = os.getenv("PROD_ACR")
-dev_aws = os.getenv("DEV_AWS")
-prod_aws = os.getenv("PROD_AWS")
-aws_region = os.getenv("AWS_REGION")
+build_tag = os.environ["BUILD_TAG"]
+branch = os.environ["BRANCH"]
+dev_acr = os.environ["DEV_ACR"]
+prod_acr = os.environ["PROD_ACR"]
+dev_aws = os.environ["DEV_AWS"]
+prod_aws = os.environ["PROD_AWS"]
+aws_region = os.environ["AWS_REGION"]
 
 components = {
     "neon": ["neon"],
@@ -41,23 +39,24 @@ registries = {
 
 outputs: dict[str, dict[str, list[str]]] = {}
 
-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_tags = [build_tag, "latest"] if branch == "main" else [build_tag]
+target_stages = ["dev", "prod"] if branch.startswith("release") else ["dev"]
 
 for component_name, component_images in components.items():
     for stage in target_stages:
-        outputs[f"{component_name}-{stage}"] = {
-            f"docker.io/neondatabase/{component_image}:{source_tag}": [
-                f"{registry}/{component_image}:{tag}"
-                for registry, tag in itertools.product(registries[stage], target_tags)
-                if not (registry == "docker.io/neondatabase" and tag == source_tag)
+        outputs[f"{component_name}-{stage}"] = dict(
+            [
+                (
+                    f"docker.io/neondatabase/{component_image}:{build_tag}",
+                    [
+                        f"{combo[0]}/{component_image}:{combo[1]}"
+                        for combo in itertools.product(registries[stage], target_tags)
+                    ],
+                )
+                for component_image in component_images
             ]
-            for component_image in component_images
-        }
+        )
 
-with open(os.getenv("GITHUB_OUTPUT", "/dev/null"), "a") as f:
+with open(os.environ["GITHUB_OUTPUT"], "a") as f:
     for key, value in outputs.items():
         f.write(f"{key}={json.dumps(value)}\n")
-        print(f"Image map for {key}:\n{json.dumps(value, indent=2)}\n\n", file=sys.stderr)

.github/scripts/lint-release-pr.sh

@@ -1,110 +0,0 @@
-#!/usr/bin/env bash
-
-set -euo pipefail
-
-DOCS_URL="https://docs.neon.build/overview/repositories/neon.html"
-
-message() {
-  if [[ -n "${GITHUB_PR_NUMBER:-}" ]]; then
-    gh pr comment --repo "${GITHUB_REPOSITORY}" "${GITHUB_PR_NUMBER}" --edit-last --body "$1" \
-      || gh pr comment --repo "${GITHUB_REPOSITORY}" "${GITHUB_PR_NUMBER}" --body "$1"
-  fi
-  echo "$1"
-}
-
-report_error() {
-  message "❌ $1
-  For more details, see the documentation: ${DOCS_URL}"
-
-  exit 1
-}
-
-case "$RELEASE_BRANCH" in
-  "release") COMPONENT="Storage" ;;
-  "release-proxy") COMPONENT="Proxy" ;;
-  "release-compute") COMPONENT="Compute" ;;
-  *)
-    report_error "Unknown release branch: ${RELEASE_BRANCH}"
-    ;;
-esac
-
-
-# Identify main and release branches
-MAIN_BRANCH="origin/main"
-REMOTE_RELEASE_BRANCH="origin/${RELEASE_BRANCH}"
-
-# Find merge base
-MERGE_BASE=$(git merge-base "${MAIN_BRANCH}" "${REMOTE_RELEASE_BRANCH}")
-echo "Merge base of ${MAIN_BRANCH} and ${RELEASE_BRANCH}: ${MERGE_BASE}"
-
-# Get the HEAD commit (last commit in PR, expected to be the merge commit)
-LAST_COMMIT=$(git rev-parse HEAD)
-
-MERGE_COMMIT_MESSAGE=$(git log -1 --format=%s "${LAST_COMMIT}")
-EXPECTED_MESSAGE_REGEX="^$COMPONENT release [0-9]{4}-[0-9]{2}-[0-9]{2}$"
-
-if ! [[ "${MERGE_COMMIT_MESSAGE}" =~ ${EXPECTED_MESSAGE_REGEX} ]]; then
-  report_error "Merge commit message does not match expected pattern: '<component> release YYYY-MM-DD'
-  Expected component: ${COMPONENT}
-  Found: '${MERGE_COMMIT_MESSAGE}'"
-fi
-echo "✅ Merge commit message is correctly formatted: '${MERGE_COMMIT_MESSAGE}'"
-
-LAST_COMMIT_PARENTS=$(git cat-file -p "${LAST_COMMIT}" | jq -sR '[capture("parent (?<parent>[0-9a-f]{40})"; "g") | .parent]')
-
-if [[ "$(echo "${LAST_COMMIT_PARENTS}" | jq 'length')" -ne 2 ]]; then
-  report_error "Last commit must be a merge commit with exactly two parents"
-fi
-
-EXPECTED_RELEASE_HEAD=$(git rev-parse "${REMOTE_RELEASE_BRANCH}")
-if echo "${LAST_COMMIT_PARENTS}" | jq -e --arg rel "${EXPECTED_RELEASE_HEAD}" 'index($rel) != null' > /dev/null; then
-  LINEAR_HEAD=$(echo "${LAST_COMMIT_PARENTS}" | jq -r '[.[] | select(. != $rel)][0]' --arg rel "${EXPECTED_RELEASE_HEAD}")
-else
-  report_error "Last commit must merge the release branch (${RELEASE_BRANCH})"
-fi
-echo "✅ Last commit correctly merges the previous commit and the release branch"
-echo "Top commit of linear history: ${LINEAR_HEAD}"
-
-MERGE_COMMIT_TREE=$(git rev-parse "${LAST_COMMIT}^{tree}")
-LINEAR_HEAD_TREE=$(git rev-parse "${LINEAR_HEAD}^{tree}")
-
-if [[ "${MERGE_COMMIT_TREE}" != "${LINEAR_HEAD_TREE}" ]]; then
-  report_error "Tree of merge commit (${MERGE_COMMIT_TREE}) does not match tree of linear history head (${LINEAR_HEAD_TREE})
-  This indicates that the merge of ${RELEASE_BRANCH} into this branch was not performed using the merge strategy 'ours'"
-fi
-echo "✅ Merge commit tree matches the linear history head"
-
-EXPECTED_PREVIOUS_COMMIT="${LINEAR_HEAD}"
-
-# Now traverse down the history, ensuring each commit has exactly one parent
-CURRENT_COMMIT="${EXPECTED_PREVIOUS_COMMIT}"
-while [[ "${CURRENT_COMMIT}" != "${MERGE_BASE}" && "${CURRENT_COMMIT}" != "${EXPECTED_RELEASE_HEAD}" ]]; do
-  CURRENT_COMMIT_PARENTS=$(git cat-file -p "${CURRENT_COMMIT}" | jq -sR '[capture("parent (?<parent>[0-9a-f]{40})"; "g") | .parent]')
-
-  if [[ "$(echo "${CURRENT_COMMIT_PARENTS}" | jq 'length')" -ne 1 ]]; then
-    report_error "Commit ${CURRENT_COMMIT} must have exactly one parent"
-  fi
-
-  NEXT_COMMIT=$(echo "${CURRENT_COMMIT_PARENTS}" | jq -r '.[0]')
-
-  if [[ "${NEXT_COMMIT}" == "${MERGE_BASE}" ]]; then
-    echo "✅ Reached merge base (${MERGE_BASE})"
-    PR_BASE="${MERGE_BASE}"
-  elif [[ "${NEXT_COMMIT}" == "${EXPECTED_RELEASE_HEAD}" ]]; then
-    echo "✅ Reached release branch (${EXPECTED_RELEASE_HEAD})"
-    PR_BASE="${EXPECTED_RELEASE_HEAD}"
-  elif [[ -z "${NEXT_COMMIT}" ]]; then
-    report_error "Unexpected end of commit history before reaching merge base"
-  fi
-
-  # Move to the next commit in the chain
-  CURRENT_COMMIT="${NEXT_COMMIT}"
-done
-
-echo "✅ All commits are properly ordered and linear"
-echo "✅ Release PR structure is valid"
-
-echo
-
-message "Commits that are part of this release:
-$(git log --oneline "${PR_BASE}..${LINEAR_HEAD}")"

.github/scripts/previous-releases.jq

@@ -17,12 +17,6 @@
 ({};
  .[$entry.component] |= (if . == null or $entry.version > .version then $entry else . end))
 
-# Ensure that each component exists, or fail
-| (["storage", "compute", "proxy"] - (keys)) as $missing
-| if ($missing | length) > 0 then
-    "Error: Found no release for \($missing | join(", "))!\n" | halt_error(1)
-  else . end
-
 # Convert the resulting object into an array of formatted strings
 | to_entries
 | map("\(.key)=\(.value.full)")

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

@@ -280,7 +280,7 @@ jobs:
       - name: Upload Neon artifact
         uses: ./.github/actions/upload
         with:
-          name: neon-${{ runner.os }}-${{ runner.arch }}-${{ inputs.build-type }}${{ inputs.sanitizers == 'enabled' && '-sanitized' || '' }}-artifact
+          name: neon-${{ runner.os }}-${{ runner.arch }}-${{ inputs.build-type }}-artifact
           path: /tmp/neon
           aws-oicd-role-arn: ${{ vars.DEV_AWS_OIDC_ROLE_ARN }}
 
@@ -347,7 +347,6 @@ jobs:
           real_s3_region: eu-central-1
           rerun_failed: true
           pg_version: ${{ matrix.pg_version }}
-          sanitizers: ${{ inputs.sanitizers }}
           aws-oicd-role-arn: ${{ vars.DEV_AWS_OIDC_ROLE_ARN }}
           # `--session-timeout` is equal to (timeout-minutes - 10 minutes) * 60 seconds.
           # Attempt to stop tests gracefully to generate test reports
@@ -360,6 +359,7 @@ jobs:
           PAGESERVER_VIRTUAL_FILE_IO_ENGINE: tokio-epoll-uring
           PAGESERVER_GET_VECTORED_CONCURRENT_IO: sidecar-task
           USE_LFC: ${{ matrix.lfc_state == 'with-lfc' && 'true' || 'false' }}
+          SANITIZERS: ${{ inputs.sanitizers }}
 
       # Temporary disable this step until we figure out why it's so flaky
       # Ref https://github.com/neondatabase/neon/issues/4540

.github/workflows/_create-release-pr.yml

@@ -7,8 +7,8 @@ on:
         description: 'Component name'
         required: true
         type: string
-      source-branch:
-        description: 'Source branch'
+      release-branch:
+        description: 'Release branch'
         required: true
         type: string
     secrets:
@@ -30,25 +30,17 @@ jobs:
     steps:
     - uses: actions/checkout@v4
       with:
-        ref: ${{ inputs.source-branch }}
-        fetch-depth: 0
+        ref: main
 
     - name: Set variables
       id: vars
       env:
         COMPONENT_NAME: ${{ inputs.component-name }}
-        RELEASE_BRANCH: >-
-          ${{
-            false
-            || inputs.component-name == 'Storage' && 'release'
-            || inputs.component-name == 'Proxy' && 'release-proxy'
-            || inputs.component-name == 'Compute' && 'release-compute'
-          }}
+        RELEASE_BRANCH: ${{ inputs.release-branch }}
       run: |
         today=$(date +'%Y-%m-%d')
         echo "title=${COMPONENT_NAME} release ${today}" | tee -a ${GITHUB_OUTPUT}
         echo "rc-branch=rc/${RELEASE_BRANCH}/${today}"  | tee -a ${GITHUB_OUTPUT}
-        echo "release-branch=${RELEASE_BRANCH}"         | tee -a ${GITHUB_OUTPUT}
 
     - name: Configure git
       run: |
@@ -57,36 +49,31 @@ jobs:
 
     - name: Create RC branch
       env:
-        RELEASE_BRANCH: ${{ steps.vars.outputs.release-branch }}
         RC_BRANCH: ${{ steps.vars.outputs.rc-branch }}
         TITLE: ${{ steps.vars.outputs.title }}
       run: |
-        git switch -c "${RC_BRANCH}"
+        git checkout -b "${RC_BRANCH}"
 
-        # Manually create a merge commit on the current branch, keeping the
-        # tree and setting the parents to the current HEAD and the HEAD of the
-        # release branch. This commit is what we'll fast-forward the release
-        # branch to when merging the release branch.
-        # For details on why, look at
-        # https://docs.neon.build/overview/repositories/neon.html#background-on-commit-history-of-release-prs
-        current_tree=$(git rev-parse 'HEAD^{tree}')
-        release_head=$(git rev-parse "origin/${RELEASE_BRANCH}")
-        current_head=$(git rev-parse HEAD)
-        merge_commit=$(git commit-tree -p "${current_head}" -p "${release_head}" -m "${TITLE}" "${current_tree}")
-
-        # Fast-forward the current branch to the newly created merge_commit
-        git merge --ff-only ${merge_commit}
+        # create an empty commit to distinguish workflow runs
+        # from other possible releases from the same commit
+        git commit --allow-empty -m "${TITLE}"
 
         git push origin "${RC_BRANCH}"
 
-    - name: Create a PR into ${{ steps.vars.outputs.release-branch }}
+    - name: Create a PR into ${{ inputs.release-branch }}
       env:
         GH_TOKEN: ${{ secrets.ci-access-token }}
         RC_BRANCH: ${{ steps.vars.outputs.rc-branch }}
-        RELEASE_BRANCH: ${{ steps.vars.outputs.release-branch }}
+        RELEASE_BRANCH: ${{ inputs.release-branch }}
         TITLE: ${{ steps.vars.outputs.title }}
       run: |
+        cat << EOF > body.md
+          ## ${TITLE}
+
+          **Please merge this Pull Request using 'Create a merge commit' button**
+        EOF
+
         gh pr create --title "${TITLE}" \
-                     --body "" \
+                     --body-file "body.md" \
                      --head "${RC_BRANCH}" \
                      --base "${RELEASE_BRANCH}"

.github/workflows/_meta.yml

@@ -19,18 +19,11 @@ on:
         description: "Tag of the last compute release"
         value: ${{ jobs.tags.outputs.compute }}
       run-kind:
-        description: "The kind of run we're currently in. Will be one of `push-main`, `storage-release`, `compute-release`, `proxy-release`, `storage-rc-pr`, `compute-rc-pr`,  `proxy-rc-pr`, `pr`, or `workflow-dispatch`"
+        description: "The kind of run we're currently in. Will be one of `pr-main`, `push-main`, `storage-rc`, `storage-release`, `proxy-rc`, `proxy-release`, `compute-rc`, `compute-release` or `merge_queue`"
         value: ${{ jobs.tags.outputs.run-kind }}
-      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 }}
 
 permissions: {}
 
-defaults:
-  run:
-    shell: bash -euo pipefail {0}
-
 jobs:
   tags:
     runs-on: ubuntu-22.04
@@ -40,7 +33,6 @@ jobs:
       proxy: ${{ steps.previous-releases.outputs.proxy }}
       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 }}
     permissions:
       contents: read
     steps:
@@ -59,11 +51,10 @@ jobs:
               || (inputs.github-event-name == 'push'         && github.ref_name == 'release')         && 'storage-release'
               || (inputs.github-event-name == 'push'         && github.ref_name == 'release-compute') && 'compute-release'
               || (inputs.github-event-name == 'push'         && github.ref_name == 'release-proxy')   && 'proxy-release'
+              || (inputs.github-event-name == 'pull_request' && github.base_ref == 'main')            && 'pr-main'
               || (inputs.github-event-name == 'pull_request' && github.base_ref == 'release')         && 'storage-rc-pr'
               || (inputs.github-event-name == 'pull_request' && github.base_ref == 'release-compute') && 'compute-rc-pr'
               || (inputs.github-event-name == 'pull_request' && github.base_ref == 'release-proxy')   && 'proxy-rc-pr'
-              || (inputs.github-event-name == 'pull_request')                                         && 'pr'
-              || (inputs.github-event-name == 'workflow_dispatch')                                    && 'workflow-dispatch'
               || 'unknown'
             }}
         run: |
@@ -90,17 +81,10 @@ jobs:
           compute-release)
             echo "tag=release-compute-$(git rev-list --count HEAD)" | tee -a $GITHUB_OUTPUT
             ;;
-          pr|storage-rc-pr|compute-rc-pr|proxy-rc-pr)
-            BUILD_AND_TEST_RUN_ID=$(gh api --paginate \
-              -H "Accept: application/vnd.github+json" \
-              -H "X-GitHub-Api-Version: 2022-11-28" \
-              "/repos/${GITHUB_REPOSITORY}/actions/runs?head_sha=${CURRENT_SHA}&branch=${CURRENT_BRANCH}" \
-              | jq '[.workflow_runs[] | select(.name == "Build and Test")][0].id // ("Error: No matching workflow run found." | halt_error(1))')
+          pr-main|storage-rc-pr|compute-rc-pr|proxy-rc-pr)
+            BUILD_AND_TEST_RUN_ID=$(gh run list -b $CURRENT_BRANCH -c $CURRENT_SHA -w 'Build and Test' -L 1 --json databaseId --jq '.[].databaseId')
             echo "tag=$BUILD_AND_TEST_RUN_ID" | tee -a $GITHUB_OUTPUT
             ;;
-          workflow-dispatch)
-            echo "tag=$GITHUB_RUN_ID" | tee -a $GITHUB_OUTPUT
-            ;;
           *)
             echo "Unexpected RUN_KIND ('${RUN_KIND}'), failing to assign build-tag!"
             exit 1
@@ -117,13 +101,3 @@ jobs:
             "/repos/${GITHUB_REPOSITORY}/releases" \
           | jq -f .github/scripts/previous-releases.jq -r \
           | tee -a "${GITHUB_OUTPUT}"
-
-      - name: Get the release PR run ID
-        id: release-pr-run-id
-        if: ${{ contains(fromJson('["storage-release", "compute-release", "proxy-release"]'), steps.run-kind.outputs.run-kind) }}
-        env:
-          GH_TOKEN: ${{ secrets.GITHUB_TOKEN }}
-          CURRENT_SHA: ${{ github.event.pull_request.head.sha || github.sha }}
-        run: |
-          RELEASE_PR_RUN_ID=$(gh api "/repos/${GITHUB_REPOSITORY}/actions/runs?head_sha=$CURRENT_SHA" | jq '[.workflow_runs[] | select(.name == "Build and Test") | select(.head_branch | test("^rc/release(-(proxy)|(compute))?/[0-9]{4}-[0-9]{2}-[0-9]{2}$"; "s"))] | first | .id // ("Faied to find Build and Test run from  RC PR!" | halt_error(1))')
-          echo "release-pr-run-id=$RELEASE_PR_RUN_ID" | tee -a $GITHUB_OUTPUT

.github/workflows/benchmarking.yml

@@ -140,9 +140,6 @@ jobs:
           --ignore test_runner/performance/test_logical_replication.py
           --ignore test_runner/performance/test_physical_replication.py
           --ignore test_runner/performance/test_perf_ingest_using_pgcopydb.py
-          --ignore test_runner/performance/test_cumulative_statistics_persistence.py
-          --ignore test_runner/performance/test_perf_many_relations.py
-          --ignore test_runner/performance/test_perf_oltp_large_tenant.py
       env:
         BENCHMARK_CONNSTR: ${{ steps.create-neon-project.outputs.dsn }}
         VIP_VAP_ACCESS_TOKEN: "${{ secrets.VIP_VAP_ACCESS_TOKEN }}"
@@ -174,61 +171,6 @@ jobs:
       env:
         SLACK_BOT_TOKEN: ${{ secrets.SLACK_BOT_TOKEN }}
 
-  cumstats-test:
-    if: ${{ github.event.inputs.run_only_pgvector_tests == 'false' || github.event.inputs.run_only_pgvector_tests == null }}
-    permissions:
-      contents: write
-      statuses: write
-      id-token: write # aws-actions/configure-aws-credentials
-    env:
-      POSTGRES_DISTRIB_DIR: /tmp/neon/pg_install
-      DEFAULT_PG_VERSION: 17
-      TEST_OUTPUT: /tmp/test_output
-      BUILD_TYPE: remote
-      SAVE_PERF_REPORT: ${{ github.event.inputs.save_perf_report || ( github.ref_name == 'main' ) }}
-      PLATFORM: "neon-staging"
-
-    runs-on: [ self-hosted, us-east-2, x64 ]
-    container:
-      image: neondatabase/build-tools:pinned-bookworm
-      credentials:
-        username: ${{ secrets.NEON_DOCKERHUB_USERNAME }}
-        password: ${{ secrets.NEON_DOCKERHUB_PASSWORD }}
-      options: --init
-
-    steps:
-    - uses: actions/checkout@v4
-
-    - name: Configure AWS credentials
-      uses: aws-actions/configure-aws-credentials@v4
-      with:
-        aws-region: eu-central-1
-        role-to-assume: ${{ vars.DEV_AWS_OIDC_ROLE_ARN }}
-        role-duration-seconds: 18000 # 5 hours
-
-    - name: Download Neon artifact
-      uses: ./.github/actions/download
-      with:
-        name: neon-${{ runner.os }}-${{ runner.arch }}-release-artifact
-        path: /tmp/neon/
-        prefix: latest
-        aws-oicd-role-arn: ${{ vars.DEV_AWS_OIDC_ROLE_ARN }}
-    
-    - name: Verify that cumulative statistics are preserved
-      uses: ./.github/actions/run-python-test-set
-      with:
-        build_type: ${{ env.BUILD_TYPE }}
-        test_selection: performance/test_cumulative_statistics_persistence.py
-        run_in_parallel: false
-        save_perf_report: ${{ env.SAVE_PERF_REPORT }}
-        extra_params: -m remote_cluster --timeout 3600
-        pg_version: ${{ env.DEFAULT_PG_VERSION }}
-        aws-oicd-role-arn: ${{ vars.DEV_AWS_OIDC_ROLE_ARN }}
-      env:
-        VIP_VAP_ACCESS_TOKEN: "${{ secrets.VIP_VAP_ACCESS_TOKEN }}"
-        PERF_TEST_RESULT_CONNSTR: "${{ secrets.PERF_TEST_RESULT_CONNSTR }}"
-        NEON_API_KEY: ${{ secrets.NEON_STAGING_API_KEY }}
-
   replication-tests:
     if: ${{ github.event.inputs.run_only_pgvector_tests == 'false' || github.event.inputs.run_only_pgvector_tests == null }}
     permissions:

.github/workflows/build_and_test.yml

@@ -476,7 +476,7 @@ jobs:
         (
           !github.event.pull_request.draft
           || contains( github.event.pull_request.labels.*.name, 'run-e2e-tests-in-draft')
-          || needs.meta.outputs.run-kind == 'push-main'
+          || contains(fromJSON('["push-main", "storage-release", "proxy-release", "compute-release"]'), needs.meta.outputs.run-kind)
         ) && !failure() && !cancelled()
       }}
     needs: [ check-permissions, push-neon-image-dev, push-compute-image-dev, meta ]
@@ -487,7 +487,7 @@ jobs:
 
   neon-image-arch:
     needs: [ check-permissions, build-build-tools-image, meta ]
-    if: ${{ contains(fromJSON('["push-main", "pr", "storage-rc-pr", "proxy-rc-pr"]'), needs.meta.outputs.run-kind) }}
+    if: ${{ contains(fromJSON('["push-main", "pr-main", "storage-release", "storage-rc-pr", "proxy-release", "proxy-rc-pr"]'), needs.meta.outputs.run-kind) }}
     strategy:
       matrix:
         arch: [ x64, arm64 ]
@@ -537,7 +537,7 @@ jobs:
 
   neon-image:
     needs: [ neon-image-arch, meta ]
-    if: ${{ contains(fromJSON('["push-main", "pr", "storage-rc-pr", "proxy-rc-pr"]'), needs.meta.outputs.run-kind) }}
+    if: ${{ contains(fromJSON('["push-main", "pr-main", "storage-release", "storage-rc-pr", "proxy-release", "proxy-rc-pr"]'), needs.meta.outputs.run-kind) }}
     runs-on: ubuntu-22.04
     permissions:
       id-token: write # aws-actions/configure-aws-credentials
@@ -559,7 +559,7 @@ jobs:
 
   compute-node-image-arch:
     needs: [ check-permissions, build-build-tools-image, meta ]
-    if: ${{ contains(fromJSON('["push-main", "pr", "compute-rc-pr"]'), needs.meta.outputs.run-kind) }}
+    if: ${{ contains(fromJSON('["push-main", "pr-main", "compute-release", "compute-rc-pr"]'), needs.meta.outputs.run-kind) }}
     permissions:
       id-token: write # aws-actions/configure-aws-credentials
       statuses: write
@@ -651,7 +651,7 @@ jobs:
 
   compute-node-image:
     needs: [ compute-node-image-arch, meta ]
-    if: ${{ contains(fromJSON('["push-main", "pr", "compute-rc-pr"]'), needs.meta.outputs.run-kind) }}
+    if: ${{ contains(fromJSON('["push-main", "pr-main", "compute-release", "compute-rc-pr"]'), needs.meta.outputs.run-kind) }}
     permissions:
       id-token: write # aws-actions/configure-aws-credentials
       statuses: write
@@ -692,15 +692,15 @@ jobs:
                                              neondatabase/neon-test-extensions-${{ matrix.version.pg }}:${{ needs.meta.outputs.build-tag }}-${{ matrix.version.debian }}-x64 \
                                              neondatabase/neon-test-extensions-${{ matrix.version.pg }}:${{ needs.meta.outputs.build-tag }}-${{ matrix.version.debian }}-arm64
 
-  vm-compute-node-image-arch:
+  vm-compute-node-image:
     needs: [ check-permissions, meta, compute-node-image ]
-    if: ${{ contains(fromJSON('["push-main", "pr", "compute-rc-pr"]'), needs.meta.outputs.run-kind) }}
-    runs-on: ${{ fromJson(format('["self-hosted", "{0}"]', matrix.arch == 'arm64' && 'large-arm64' || 'large')) }}
+    if: ${{ contains(fromJSON('["push-main", "pr-main", "compute-release", "compute-rc-pr"]'), needs.meta.outputs.run-kind) }}
+    runs-on: [ self-hosted, large ]
     strategy:
       fail-fast: false
       matrix:
-        arch: [ amd64, arm64 ]
         version:
+          # see the comment for `compute-node-image-arch` job
           - pg: v14
             debian: bullseye
           - pg: v15
@@ -717,7 +717,7 @@ jobs:
 
       - name: Downloading vm-builder
         run: |
-          curl -fL https://github.com/neondatabase/autoscaling/releases/download/$VM_BUILDER_VERSION/vm-builder-${{ matrix.arch }} -o vm-builder
+          curl -fL https://github.com/neondatabase/autoscaling/releases/download/$VM_BUILDER_VERSION/vm-builder-amd64 -o vm-builder
           chmod +x vm-builder
 
       - uses: neondatabase/dev-actions/set-docker-config-dir@6094485bf440001c94a94a3f9e221e81ff6b6193
@@ -738,47 +738,17 @@ jobs:
             -size=2G \
             -spec=compute/vm-image-spec-${{ matrix.version.debian }}.yaml \
             -src=neondatabase/compute-node-${{ matrix.version.pg }}:${{ needs.meta.outputs.build-tag }} \
-            -dst=neondatabase/vm-compute-node-${{ matrix.version.pg }}:${{ needs.meta.outputs.build-tag }}-${{ matrix.arch }} \
-            -target-arch=linux/${{ matrix.arch }}
+            -dst=neondatabase/vm-compute-node-${{ matrix.version.pg }}:${{ needs.meta.outputs.build-tag }} \
+            -target-arch=linux/amd64
 
       - name: Pushing vm-compute-node image
         run: |
-          docker push neondatabase/vm-compute-node-${{ matrix.version.pg }}:${{ needs.meta.outputs.build-tag }}-${{ matrix.arch }}
-
-  vm-compute-node-image:
-    needs: [ vm-compute-node-image-arch, meta ]
-    if: ${{ contains(fromJSON('["push-main", "pr", "compute-rc-pr"]'), needs.meta.outputs.run-kind) }}
-    runs-on: ubuntu-22.04
-    strategy:
-      matrix:
-        version:
-          # see the comment for `compute-node-image-arch` job
-          - pg: v14
-          - pg: v15
-          - pg: v16
-          - pg: v17
-    steps:
-      - uses: docker/login-action@v3
-        with:
-          username: ${{ secrets.NEON_DOCKERHUB_USERNAME }}
-          password: ${{ secrets.NEON_DOCKERHUB_PASSWORD }}
-
-      - name: Create multi-arch compute-node image
-        run: |
-          docker buildx imagetools create -t neondatabase/vm-compute-node-${{ matrix.version.pg }}:${{ needs.meta.outputs.build-tag }} \
-                                             neondatabase/vm-compute-node-${{ matrix.version.pg }}:${{ needs.meta.outputs.build-tag }}-amd64 \
-                                             neondatabase/vm-compute-node-${{ matrix.version.pg }}:${{ needs.meta.outputs.build-tag }}-arm64
-
+          docker push neondatabase/vm-compute-node-${{ matrix.version.pg }}:${{ needs.meta.outputs.build-tag }}
 
   test-images:
     needs: [ check-permissions, meta, neon-image, compute-node-image ]
     # Depends on jobs that can get skipped
-    if: >-
-      ${{
-        !failure()
-        && !cancelled()
-        && contains(fromJSON('["push-main", "pr", "storage-rc-pr", "proxy-rc-pr", "compute-rc-pr"]'), needs.meta.outputs.run-kind)
-      }}
+    if: "!failure() && !cancelled()"
     strategy:
       fail-fast: false
       matrix:
@@ -805,7 +775,7 @@ jobs:
       # Ensure that we don't have bad versions.
       - name: Verify image versions
         shell: bash # ensure no set -e for better error messages
-        if: ${{ contains(fromJSON('["push-main", "pr", "storage-rc-pr", "proxy-rc-pr"]'), needs.meta.outputs.run-kind) }}
+        if: ${{ contains(fromJSON('["push-main", "pr-main", "storage-release", "storage-rc-pr", "proxy-release", "proxy-rc-pr"]'), needs.meta.outputs.run-kind) }}
         run: |
           pageserver_version=$(docker run --rm neondatabase/neon:${{ needs.meta.outputs.build-tag }} "/bin/sh" "-c" "/usr/local/bin/pageserver --version")
 
@@ -826,19 +796,19 @@ jobs:
         env:
           TAG: >-
             ${{
-              needs.meta.outputs.run-kind == 'compute-rc-pr'
+              contains(fromJSON('["compute-release", "compute-rc-pr"]'), needs.meta.outputs.run-kind)
               && needs.meta.outputs.previous-storage-release
               || needs.meta.outputs.build-tag
             }}
           COMPUTE_TAG: >-
             ${{
-              contains(fromJSON('["storage-rc-pr", "proxy-rc-pr"]'), needs.meta.outputs.run-kind)
+              contains(fromJSON('["storage-release", "storage-rc-pr", "proxy-release", "proxy-rc-pr"]'), needs.meta.outputs.run-kind)
               && needs.meta.outputs.previous-compute-release
               || needs.meta.outputs.build-tag
             }}
           TEST_EXTENSIONS_TAG: >-
             ${{
-              contains(fromJSON('["storage-rc-pr", "proxy-rc-pr"]'), needs.meta.outputs.run-kind)
+              contains(fromJSON('["storage-release", "storage-rc-pr", "proxy-release", "proxy-rc-pr"]'), needs.meta.outputs.run-kind)
               && 'latest'
               || needs.meta.outputs.build-tag
             }}
@@ -853,15 +823,15 @@ jobs:
 
       - name: Test extension upgrade
         timeout-minutes: 20
-        if: ${{ contains(fromJSON('["pr", "compute-rc-pr"]'), needs.meta.outputs.run-kind) }}
+        if: ${{ contains(fromJSON('["pr-main", "compute-rc-pr"]'), needs.meta.outputs.run-kind) }}
         env:
           TAG: >-
             ${{
               false
-              || needs.meta.outputs.run-kind == 'pr' && needs.meta.outputs.build-tag
+              || needs.meta.outputs.run-kind == 'pr-main' && needs.meta.outputs.build-tag
               || needs.meta.outputs.run-kind == 'compute-rc-pr' && needs.meta.outputs.previous-storage-release
             }}
-          TEST_EXTENSIONS_TAG: ${{ needs.meta.outputs.previous-compute-release }}
+          TEST_EXTENSIONS_TAG: latest
           NEW_COMPUTE_TAG: ${{ needs.meta.outputs.build-tag }}
           OLD_COMPUTE_TAG: ${{ needs.meta.outputs.previous-compute-release }}
         run: ./docker-compose/test_extensions_upgrade.sh
@@ -890,13 +860,7 @@ jobs:
         id: generate
         run: python3 .github/scripts/generate_image_maps.py
         env:
-          SOURCE_TAG: >-
-            ${{
-              contains(fromJson('["storage-release", "compute-release", "proxy-release"]'), needs.meta.outputs.run-kind)
-              && needs.meta.outputs.release-pr-run-id
-              || needs.meta.outputs.build-tag
-            }}
-          TARGET_TAG: ${{ needs.meta.outputs.build-tag }}
+          BUILD_TAG: "${{ needs.meta.outputs.build-tag }}"
           BRANCH: "${{ github.ref_name }}"
           DEV_ACR: "${{ vars.AZURE_DEV_REGISTRY_NAME }}"
           PROD_ACR: "${{ vars.AZURE_PROD_REGISTRY_NAME }}"
@@ -906,7 +870,7 @@ jobs:
 
   push-neon-image-dev:
     needs: [ meta, generate-image-maps, neon-image ]
-    if: ${{ !failure() && !cancelled() && contains(fromJSON('["push-main", "pr", "storage-release", "storage-rc-pr", "proxy-release", "proxy-rc-pr"]'), needs.meta.outputs.run-kind) }}
+    if: ${{ contains(fromJSON('["push-main", "pr-main", "storage-release", "storage-rc-pr", "proxy-release", "proxy-rc-pr"]'), needs.meta.outputs.run-kind) }}
     uses: ./.github/workflows/_push-to-container-registry.yml
     permissions:
       id-token: write  # Required for aws/azure login
@@ -924,7 +888,7 @@ jobs:
 
   push-compute-image-dev:
     needs: [ meta, generate-image-maps, vm-compute-node-image ]
-    if: ${{ !failure() && !cancelled() && contains(fromJSON('["push-main", "pr", "compute-release", "compute-rc-pr"]'), needs.meta.outputs.run-kind) }}
+    if: ${{ contains(fromJSON('["push-main", "pr-main", "compute-release", "compute-rc-pr"]'), needs.meta.outputs.run-kind) }}
     uses: ./.github/workflows/_push-to-container-registry.yml
     permissions:
       id-token: write  # Required for aws/azure login
@@ -942,8 +906,7 @@ jobs:
 
   push-neon-image-prod:
     needs: [ meta, generate-image-maps, neon-image, test-images ]
-    # Depends on jobs that can get skipped
-    if: ${{ !failure() && !cancelled() && contains(fromJSON('["storage-release", "proxy-release"]'), needs.meta.outputs.run-kind) }}
+    if: ${{ contains(fromJSON('["storage-release", "proxy-release"]'), needs.meta.outputs.run-kind) }}
     uses: ./.github/workflows/_push-to-container-registry.yml
     permissions:
       id-token: write  # Required for aws/azure login
@@ -961,8 +924,7 @@ jobs:
 
   push-compute-image-prod:
     needs: [ meta, generate-image-maps, vm-compute-node-image, test-images ]
-    # Depends on jobs that can get skipped
-    if: ${{ !failure() && !cancelled() && needs.meta.outputs.run-kind == 'compute-release' }}
+    if: ${{ needs.meta.outputs.run-kind == 'compute-release' }}
     uses: ./.github/workflows/_push-to-container-registry.yml
     permissions:
       id-token: write  # Required for aws/azure login
@@ -993,7 +955,7 @@ jobs:
 
   trigger-custom-extensions-build-and-wait:
     needs: [ check-permissions, meta ]
-    if: ${{ contains(fromJSON('["push-main", "pr", "compute-release", "compute-rc-pr"]'), needs.meta.outputs.run-kind) }}
+    if: ${{ contains(fromJSON('["push-main", "pr-main", "compute-release", "compute-rc-pr"]'), needs.meta.outputs.run-kind) }}
     runs-on: ubuntu-22.04
     permissions:
       id-token: write # aws-actions/configure-aws-credentials
@@ -1072,7 +1034,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, build-and-test-locally, trigger-custom-extensions-build-and-wait ]
+    needs: [ check-permissions, 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:
@@ -1186,7 +1148,7 @@ jobs:
               -f deployPgSniRouter=false \
               -f deployProxy=false \
               -f deployStorage=true \
-              -f deployStorageBroker=false \
+              -f deployStorageBroker=true \
               -f deployStorageController=true \
               -f branch=main \
               -f dockerTag=${{needs.meta.outputs.build-tag}} \
@@ -1194,7 +1156,7 @@ jobs:
 
             gh workflow --repo neondatabase/infra run deploy-prod.yml --ref main \
               -f deployStorage=true \
-              -f deployStorageBroker=false \
+              -f deployStorageBroker=true \
               -f deployStorageController=true \
               -f branch=main \
               -f dockerTag=${{needs.meta.outputs.build-tag}}
@@ -1242,11 +1204,11 @@ jobs:
           payload: |
             channel: ${{ vars.SLACK_STORAGE_CHANNEL_ID }}
             text: |
-              🔴 <!subteam^S06CJ87UMNY|@oncall-storage>: deploy job on release branch had unexpected status "${{ needs.deploy.result }}" <${{ github.server_url }}/${{ github.repository }}/actions/runs/${{ github.run_id }}|GitHub Run>.
+              🔴 @oncall-storage: deploy job on release branch had unexpected status "${{ needs.deploy.result }}" <${{ github.server_url }}/${{ github.repository }}/actions/runs/${{ github.run_id }}|GitHub Run>.
 
   # The job runs on `release` branch and copies compatibility data and Neon artifact from the last *release PR* to the latest directory
   promote-compatibility-data:
-    needs: [ meta, deploy ]
+    needs: [ deploy ]
     permissions:
       id-token: write # aws-actions/configure-aws-credentials
       statuses: write
@@ -1256,6 +1218,37 @@ jobs:
 
     runs-on: ubuntu-22.04
     steps:
+      - name: Fetch GITHUB_RUN_ID and COMMIT_SHA for the last merged release PR
+        id: fetch-last-release-pr-info
+        env:
+          GH_TOKEN: ${{ secrets.GITHUB_TOKEN }}
+        run: |
+          branch_name_and_pr_number=$(gh pr list \
+            --repo "${GITHUB_REPOSITORY}" \
+            --base release \
+            --state merged \
+            --limit 10 \
+            --json mergeCommit,headRefName,number \
+            --jq ".[] | select(.mergeCommit.oid==\"${GITHUB_SHA}\") | { branch_name: .headRefName, pr_number: .number }")
+          branch_name=$(echo "${branch_name_and_pr_number}" | jq -r '.branch_name')
+          pr_number=$(echo "${branch_name_and_pr_number}" | jq -r '.pr_number')
+
+          run_id=$(gh run list \
+            --repo "${GITHUB_REPOSITORY}" \
+            --workflow build_and_test.yml \
+            --branch "${branch_name}" \
+            --json databaseId \
+            --limit 1 \
+            --jq '.[].databaseId')
+
+          last_commit_sha=$(gh pr view "${pr_number}" \
+            --repo "${GITHUB_REPOSITORY}" \
+            --json commits \
+            --jq '.commits[-1].oid')
+
+          echo "run-id=${run_id}" | tee -a ${GITHUB_OUTPUT}
+          echo "commit-sha=${last_commit_sha}" | tee -a ${GITHUB_OUTPUT}
+
       - uses: aws-actions/configure-aws-credentials@v4
         with:
           aws-region: eu-central-1
@@ -1266,8 +1259,8 @@ jobs:
         env:
           BUCKET: neon-github-public-dev
           AWS_REGION: eu-central-1
-          COMMIT_SHA: ${{ github.sha }}
-          RUN_ID: ${{ needs.meta.outputs.release-pr-run-id }}
+          COMMIT_SHA: ${{ steps.fetch-last-release-pr-info.outputs.commit-sha }}
+          RUN_ID: ${{ steps.fetch-last-release-pr-info.outputs.run-id }}
         run: |
           old_prefix="artifacts/${COMMIT_SHA}/${RUN_ID}"
           new_prefix="artifacts/latest"
@@ -1354,7 +1347,7 @@ jobs:
           || needs.check-codestyle-python.result == 'skipped'
           || needs.check-codestyle-rust.result == 'skipped'
           || 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))
-          || (needs.test-images.result == 'skipped' && contains(fromJSON('["push-main", "pr", "storage-rc-pr", "proxy-rc-pr", "compute-rc-pr"]'), needs.meta.outputs.run-kind))
-          || (needs.trigger-custom-extensions-build-and-wait.result == 'skipped' && contains(fromJSON('["push-main", "pr", "compute-release", "compute-rc-pr"]'), needs.meta.outputs.run-kind))
+          || (needs.push-compute-image-dev.result == 'skipped' && contains(fromJSON('["push-main", "pr-main", "compute-release", "compute-rc-pr"]'), needs.meta.outputs.run-kind))
+          || (needs.push-neon-image-dev.result == 'skipped' && contains(fromJSON('["push-main", "pr-main", "storage-release", "storage-rc-pr", "proxy-release", "proxy-rc-pr"]'), needs.meta.outputs.run-kind))
+          || needs.test-images.result == 'skipped'
+          || (needs.trigger-custom-extensions-build-and-wait.result == 'skipped' && contains(fromJSON('["push-main", "pr-main", "compute-release", "compute-rc-pr"]'), needs.meta.outputs.run-kind))

.github/workflows/cargo-deny.yml

@@ -7,7 +7,7 @@ on:
         required: false
         type: string
   schedule:
-    - cron: '0 10 * * *'
+    - cron: '0 0 * * *'
 
 jobs:
   cargo-deny:
@@ -50,9 +50,8 @@ jobs:
           method: chat.postMessage
           token: ${{ secrets.SLACK_BOT_TOKEN }}
           payload: |
-            channel: ${{ vars.SLACK_ON_CALL_DEVPROD_STREAM }}
+            channel: ${{ vars.SLACK_CICD_CHANNEL_ID }}
             text: |
               Periodic cargo-deny on ${{ matrix.ref }}: ${{ job.status }}
               <${{ github.server_url }}/${{ github.repository }}/actions/runs/${{ github.run_id }}|GitHub Run>
-              Fixing the problem should be fairly straight forward from the logs. If not, <#${{ vars.SLACK_RUST_CHANNEL_ID }}> is there to help.
-              Pinging <!subteam^S0838JPSH32|@oncall-devprod>.
+              Pinging @oncall-devprod.

.github/workflows/fast-forward.yml

@@ -1,36 +0,0 @@
-name: Fast forward merge
-on:
-  pull_request:
-    types: [labeled]
-    branches:
-      - release
-      - release-proxy
-      - release-compute
-
-jobs:
-  fast-forward:
-    if: ${{ github.event.label.name == 'fast-forward' }}
-    runs-on: ubuntu-22.04
-
-    steps:
-      - name: Remove fast-forward label to PR
-        env:
-          GH_TOKEN: ${{ secrets.CI_ACCESS_TOKEN }}
-        run: |
-          gh pr edit ${{ github.event.pull_request.number }} --repo "${GITHUB_REPOSITORY}" --remove-label "fast-forward"
-
-      - name: Fast forwarding
-        uses: sequoia-pgp/fast-forward@ea7628bedcb0b0b96e94383ada458d812fca4979
-        # See https://docs.github.com/en/graphql/reference/enums#mergestatestatus
-        if: ${{ github.event.pull_request.mergeable_state  == 'clean' }}
-        with:
-          merge: true
-          comment: on-error
-          github_token: ${{ secrets.CI_ACCESS_TOKEN }}
-
-      - name: Comment if mergeable_state is not clean
-        if: ${{ github.event.pull_request.mergeable_state  != 'clean' }}
-        run: |
-          gh pr comment ${{ github.event.pull_request.number }} \
-            --repo "${GITHUB_REPOSITORY}" \
-            --body "Not trying to forward pull-request, because \`mergeable_state\` is \`${{ github.event.pull_request.mergeable_state }}\`, not \`clean\`."

.github/workflows/force-test-extensions-upgrade.yml

@@ -52,9 +52,8 @@ jobs:
       - name: Test extension upgrade
         timeout-minutes: 20
         env:
-          NEW_COMPUTE_TAG: latest
-          OLD_COMPUTE_TAG: ${{ steps.get-last-compute-release-tag.outputs.tag }}
-          TEST_EXTENSIONS_TAG: ${{ steps.get-last-compute-release-tag.outputs.tag }}
+          NEWTAG: latest
+          OLDTAG: ${{ steps.get-last-compute-release-tag.outputs.tag }}
           PG_VERSION: ${{ matrix.pg-version }}
           FORCE_ALL_UPGRADE_TESTS: true
         run: ./docker-compose/test_extensions_upgrade.sh

.github/workflows/large_oltp_benchmark.yml

@@ -1,147 +0,0 @@
-name: large oltp benchmark
-
-on:
-  # uncomment to run on push for debugging your PR
-  push:
-    branches: [ bodobolero/synthetic_oltp_workload ]
-
-  schedule:
-    # * is a special character in YAML so you have to quote this string
-    #          ┌───────────── minute (0 - 59)
-    #          │ ┌───────────── hour (0 - 23)
-    #          │ │  ┌───────────── day of the month (1 - 31)
-    #          │ │  │ ┌───────────── month (1 - 12 or JAN-DEC)
-    #          │ │  │ │ ┌───────────── day of the week (0 - 6 or SUN-SAT)
-    - cron:   '0 15 * * *' # run once a day, timezone is utc, avoid conflict with other benchmarks
-  workflow_dispatch: # adds ability to run this manually
-
-defaults:
-  run:
-    shell: bash -euxo pipefail {0}
-
-concurrency:
-  # Allow only one workflow globally because we need dedicated resources which only exist once
-  group: large-oltp-bench-workflow
-  cancel-in-progress: true
-
-jobs:
-  oltp:
-    strategy:
-      fail-fast: false # allow other variants to continue even if one fails
-      matrix:
-        include:
-          - target: new_branch 
-            custom_scripts: insert_webhooks.sql@2 select_any_webhook_with_skew.sql@4 select_recent_webhook.sql@4 
-          - target: reuse_branch 
-            custom_scripts: insert_webhooks.sql@2 select_any_webhook_with_skew.sql@4 select_recent_webhook.sql@4 
-      max-parallel: 1 # we want to run each stripe size sequentially to be able to compare the results
-    permissions:
-      contents: write
-      statuses: write
-      id-token: write # aws-actions/configure-aws-credentials
-    env:
-      TEST_PG_BENCH_DURATIONS_MATRIX: "1h" # todo update to > 1 h 
-      TEST_PGBENCH_CUSTOM_SCRIPTS: ${{ matrix.custom_scripts }}
-      POSTGRES_DISTRIB_DIR: /tmp/neon/pg_install
-      PG_VERSION: 16 # pre-determined by pre-determined project
-      TEST_OUTPUT: /tmp/test_output
-      BUILD_TYPE: remote
-      SAVE_PERF_REPORT: ${{ github.ref_name == 'main' }}
-      PLATFORM: ${{ matrix.target }}
-
-    runs-on: [ self-hosted, us-east-2, x64 ]
-    container:
-      image: neondatabase/build-tools:pinned-bookworm
-      credentials:
-        username: ${{ secrets.NEON_DOCKERHUB_USERNAME }}
-        password: ${{ secrets.NEON_DOCKERHUB_PASSWORD }}
-      options: --init
-
-    # Increase timeout to 8h, default timeout is 6h
-    timeout-minutes: 480
-
-    steps:
-    - uses: actions/checkout@v4
-
-    - name: Configure AWS credentials # necessary to download artefacts
-      uses: aws-actions/configure-aws-credentials@v4
-      with:
-        aws-region: eu-central-1
-        role-to-assume: ${{ vars.DEV_AWS_OIDC_ROLE_ARN }}
-        role-duration-seconds: 18000 # 5 hours is currently max associated with IAM role
-
-    - name: Download Neon artifact
-      uses: ./.github/actions/download
-      with:
-        name: neon-${{ runner.os }}-${{ runner.arch }}-release-artifact
-        path: /tmp/neon/
-        prefix: latest
-        aws-oicd-role-arn: ${{ vars.DEV_AWS_OIDC_ROLE_ARN }}
-
-    - name: Create Neon Branch for large tenant
-      if: ${{ matrix.target == 'new_branch' }}
-      id: create-neon-branch-oltp-target
-      uses: ./.github/actions/neon-branch-create
-      with:
-          project_id: ${{ vars.BENCHMARK_LARGE_OLTP_PROJECTID }}
-          api_key: ${{ secrets.NEON_STAGING_API_KEY }}
-
-    - name: Set up Connection String
-      id: set-up-connstr
-      run: |
-          case "${{ matrix.target }}" in
-              new_branch)
-              CONNSTR=${{ steps.create-neon-branch-oltp-target.outputs.dsn }}
-              ;;
-              reuse_branch)
-              CONNSTR=${{ secrets.BENCHMARK_LARGE_OLTP_REUSE_CONNSTR }}
-              ;;
-              *)
-              echo >&2 "Unknown target=${{ matrix.target }}"
-              exit 1
-              ;;
-          esac
-
-          echo "connstr=${CONNSTR}" >> $GITHUB_OUTPUT
-
-    - name: Benchmark pgbench with custom-scripts
-      uses: ./.github/actions/run-python-test-set
-      with:
-        build_type: ${{ env.BUILD_TYPE }}
-        test_selection: performance
-        run_in_parallel: false
-        save_perf_report: ${{ env.SAVE_PERF_REPORT }}
-        extra_params: -m remote_cluster --timeout 21600 -k test_perf_oltp_large_tenant
-        pg_version: ${{ env.PG_VERSION }}
-        aws-oicd-role-arn: ${{ vars.DEV_AWS_OIDC_ROLE_ARN }}
-      env:
-        BENCHMARK_CONNSTR: ${{ steps.set-up-connstr.outputs.connstr }}
-        VIP_VAP_ACCESS_TOKEN: "${{ secrets.VIP_VAP_ACCESS_TOKEN }}"
-        PERF_TEST_RESULT_CONNSTR: "${{ secrets.PERF_TEST_RESULT_CONNSTR }}"
-
-    - name: Delete Neon Branch for large tenant
-      if: ${{ always() && matrix.target == 'new_branch' }}
-      uses: ./.github/actions/neon-branch-delete
-      with:
-        project_id: ${{ vars.BENCHMARK_LARGE_OLTP_PROJECTID }}
-        branch_id: ${{ steps.create-neon-branch-oltp-target.outputs.branch_id }}
-        api_key: ${{ secrets.NEON_STAGING_API_KEY }}
-
-    - name: Create Allure report
-      id: create-allure-report
-      if: ${{ !cancelled() }}
-      uses: ./.github/actions/allure-report-generate
-      with:
-        aws-oicd-role-arn: ${{ vars.DEV_AWS_OIDC_ROLE_ARN }}
-  
-    - name: Post to a Slack channel
-      if: ${{ github.event.schedule && failure() }}
-      uses: slackapi/slack-github-action@v1
-      with:
-        channel-id: "C06KHQVQ7U3" # on-call-qa-staging-stream
-        slack-message: |
-          Periodic large oltp perf testing: ${{ job.status }}
-          <${{ github.server_url }}/${{ github.repository }}/actions/runs/${{ github.run_id }}|GitHub Run>
-          <${{ steps.create-allure-report.outputs.report-url }}|Allure report>
-      env:
-        SLACK_BOT_TOKEN: ${{ secrets.SLACK_BOT_TOKEN }}

.github/workflows/lint-release-pr.yml

@@ -1,24 +0,0 @@
-name: Lint Release PR
-
-on:
-  pull_request:
-    branches:
-      - release
-      - release-proxy
-      - release-compute
-
-jobs:
-  lint-release-pr:
-    runs-on: ubuntu-22.04
-    steps:
-      - name: Checkout PR branch
-        uses: actions/checkout@v4
-        with:
-          fetch-depth: 0  # Fetch full history for git operations
-          ref: ${{ github.event.pull_request.head.ref }}
-
-      - name: Run lint script
-        env:
-          RELEASE_BRANCH: ${{ github.base_ref }}
-        run: |
-          ./.github/scripts/lint-release-pr.sh

.github/workflows/periodic_pagebench.yml

@@ -3,12 +3,12 @@ name: Periodic pagebench performance test on dedicated EC2 machine in eu-central
 on:
   schedule:
     # * is a special character in YAML so you have to quote this string
-    #        ┌───────────── minute (0 - 59)
-    #        │   ┌───────────── hour (0 - 23)
-    #        │   │ ┌───────────── day of the month (1 - 31)
-    #        │   │ │ ┌───────────── month (1 - 12 or JAN-DEC)
-    #        │   │ │ │ ┌───────────── day of the week (0 - 6 or SUN-SAT)
-    - cron: '0 */3 * * *' # Runs every 3 hours
+    #          ┌───────────── minute (0 - 59)
+    #          │ ┌───────────── hour (0 - 23)
+    #          │ │ ┌───────────── day of the month (1 - 31)
+    #          │ │ │ ┌───────────── month (1 - 12 or JAN-DEC)
+    #          │ │ │ │ ┌───────────── day of the week (0 - 6 or SUN-SAT)
+    - cron:  '0 18 * * *' # Runs at 6 PM UTC every day
   workflow_dispatch: # Allows manual triggering of the workflow
     inputs:
       commit_hash:
@@ -78,10 +78,8 @@ jobs:
       run: |
         if [ -z "$INPUT_COMMIT_HASH" ]; then
           echo "COMMIT_HASH=$(curl -s https://api.github.com/repos/neondatabase/neon/commits/main | jq -r '.sha')" >> $GITHUB_ENV
-          echo "COMMIT_HASH_TYPE=latest" >> $GITHUB_ENV
         else
           echo "COMMIT_HASH=$INPUT_COMMIT_HASH" >> $GITHUB_ENV
-          echo "COMMIT_HASH_TYPE=manual" >> $GITHUB_ENV
         fi
 
     - name: Start Bench with run_id
@@ -91,7 +89,7 @@ jobs:
         -H 'accept: application/json' \
         -H 'Content-Type: application/json' \
         -H "Authorization: Bearer $API_KEY" \
-        -d "{\"neonRepoCommitHash\": \"${COMMIT_HASH}\", \"neonRepoCommitHashType\": \"${COMMIT_HASH_TYPE}\"}"
+        -d "{\"neonRepoCommitHash\": \"${COMMIT_HASH}\"}"
 
     - name: Poll Test Status
       id: poll_step

.github/workflows/pre-merge-checks.yml

@@ -8,6 +8,8 @@ on:
       - .github/workflows/build-build-tools-image.yml
       - .github/workflows/pre-merge-checks.yml
   merge_group:
+    branches:
+      - main
 
 defaults:
   run:
@@ -17,13 +19,11 @@ defaults:
 permissions: {}
 
 jobs:
-  meta:
+  get-changed-files:
     runs-on: ubuntu-22.04
     outputs:
       python-changed: ${{ steps.python-src.outputs.any_changed }}
       rust-changed: ${{ steps.rust-src.outputs.any_changed }}
-      branch: ${{ steps.group-metadata.outputs.branch }}
-      pr-number: ${{ steps.group-metadata.outputs.pr-number }}
     steps:
       - uses: actions/checkout@v4
 
@@ -58,20 +58,12 @@ jobs:
           echo "${PYTHON_CHANGED_FILES}"
           echo "${RUST_CHANGED_FILES}"
 
-      - name: Merge group metadata
-        if: ${{ github.event_name == 'merge_group' }}
-        id: group-metadata
-        env:
-          MERGE_QUEUE_REF: ${{ github.event.merge_group.head_ref }}
-        run: |
-          echo $MERGE_QUEUE_REF | jq -Rr 'capture("refs/heads/gh-readonly-queue/(?<branch>.*)/pr-(?<pr_number>[0-9]+)-[0-9a-f]{40}") | ["branch=" + .branch, "pr-number=" + .pr_number] | .[]' | tee -a "${GITHUB_OUTPUT}"
-
   build-build-tools-image:
     if: |
       false
-      || needs.meta.outputs.python-changed == 'true'
-      || needs.meta.outputs.rust-changed == 'true'
-    needs: [ meta ]
+      || needs.get-changed-files.outputs.python-changed == 'true'
+      || needs.get-changed-files.outputs.rust-changed == 'true'
+    needs: [ get-changed-files ]
     uses: ./.github/workflows/build-build-tools-image.yml
     with:
       # Build only one combination to save time
@@ -80,8 +72,8 @@ jobs:
     secrets: inherit
 
   check-codestyle-python:
-    if: needs.meta.outputs.python-changed == 'true'
-    needs: [ meta, build-build-tools-image ]
+    if: needs.get-changed-files.outputs.python-changed == 'true'
+    needs: [ get-changed-files, build-build-tools-image ]
     uses: ./.github/workflows/_check-codestyle-python.yml
     with:
       # `-bookworm-x64` suffix should match the combination in `build-build-tools-image`
@@ -89,8 +81,8 @@ jobs:
     secrets: inherit
 
   check-codestyle-rust:
-    if: needs.meta.outputs.rust-changed == 'true'
-    needs: [ meta, build-build-tools-image ]
+    if: needs.get-changed-files.outputs.rust-changed == 'true'
+    needs: [ get-changed-files, build-build-tools-image ]
     uses: ./.github/workflows/_check-codestyle-rust.yml
     with:
       # `-bookworm-x64` suffix should match the combination in `build-build-tools-image`
@@ -109,7 +101,7 @@ jobs:
       statuses: write # for `github.repos.createCommitStatus(...)`
       contents: write
     needs:
-      - meta
+      - get-changed-files
       - check-codestyle-python
       - check-codestyle-rust
     runs-on: ubuntu-22.04
@@ -137,20 +129,7 @@ jobs:
         run: exit 1
         if: |
           false
-          || (github.event_name == 'merge_group' && needs.meta.outputs.branch != 'main')
-          || (needs.check-codestyle-python.result == 'skipped' && needs.meta.outputs.python-changed == 'true')
-          || (needs.check-codestyle-rust.result   == 'skipped' && needs.meta.outputs.rust-changed   == 'true')
+          || (needs.check-codestyle-python.result == 'skipped' && needs.get-changed-files.outputs.python-changed == 'true')
+          || (needs.check-codestyle-rust.result   == 'skipped' && needs.get-changed-files.outputs.rust-changed   == 'true')
           || contains(needs.*.result, 'failure')
           || contains(needs.*.result, 'cancelled')
-
-      - name: Add fast-forward label to PR to trigger fast-forward merge
-        if: >-
-          ${{
-            always()
-            && github.event_name == 'merge_group'
-            && contains(fromJson('["release", "release-proxy", "release-compute"]'), github.base_ref)
-          }}
-        env:
-          GH_TOKEN: ${{ secrets.CI_ACCESS_TOKEN }}
-        run: >-
-          gh pr edit ${{ needs.meta.outputs.pr-number }} --repo "${GITHUB_REPOSITORY}" --add-label "fast-forward"

.github/workflows/release.yml

@@ -38,7 +38,7 @@ jobs:
     uses: ./.github/workflows/_create-release-pr.yml
     with:
       component-name: 'Storage'
-      source-branch: ${{ github.ref_name }}
+      release-branch: 'release'
     secrets:
       ci-access-token: ${{ secrets.CI_ACCESS_TOKEN }}
 
@@ -51,7 +51,7 @@ jobs:
     uses: ./.github/workflows/_create-release-pr.yml
     with:
       component-name: 'Proxy'
-      source-branch: ${{ github.ref_name }}
+      release-branch: 'release-proxy'
     secrets:
       ci-access-token: ${{ secrets.CI_ACCESS_TOKEN }}
 
@@ -64,6 +64,6 @@ jobs:
     uses: ./.github/workflows/_create-release-pr.yml
     with:
       component-name: 'Compute'
-      source-branch: ${{ github.ref_name }}
+      release-branch: 'release-compute'
     secrets:
       ci-access-token: ${{ secrets.CI_ACCESS_TOKEN }}

CODEOWNERS

@@ -1,8 +1,8 @@
 # Autoscaling
 /libs/vm_monitor/ @neondatabase/autoscaling
 
-# DevProd & PerfCorr
-/.github/ @neondatabase/developer-productivity @neondatabase/performance-correctness
+# DevProd
+/.github/ @neondatabase/developer-productivity
 
 # Compute
 /pgxn/ @neondatabase/compute

Cargo.toml

@@ -43,7 +43,7 @@ members = [
 ]
 
 [workspace.package]
-edition = "2024"
+edition = "2021"
 license = "Apache-2.0"
 
 ## All dependency versions, used in the project
@@ -53,6 +53,7 @@ anyhow = { version = "1.0", features = ["backtrace"] }
 arc-swap = "1.6"
 async-compression = { version = "0.4.0", features = ["tokio", "gzip", "zstd"] }
 atomic-take = "1.1.0"
+backtrace = "0.3.74"
 flate2 = "1.0.26"
 assert-json-diff = "2"
 async-stream = "0.3"
@@ -67,10 +68,9 @@ aws-credential-types = "1.2.0"
 aws-sigv4 = { version = "1.2", features = ["sign-http"] }
 aws-types = "1.3"
 axum = { version = "0.8.1", features = ["ws"] }
-axum-extra = { version = "0.10.0", features = ["typed-header"] }
 base64 = "0.13.0"
 bincode = "1.3"
-bindgen = "0.71"
+bindgen = "0.70"
 bit_field = "0.10.2"
 bstr = "1.0"
 byteorder = "1.4"
@@ -95,7 +95,6 @@ futures = "0.3"
 futures-core = "0.3"
 futures-util = "0.3"
 git-version = "0.3"
-governor = "0.8"
 hashbrown = "0.14"
 hashlink = "0.9.1"
 hdrhistogram = "7.5.2"
@@ -114,10 +113,11 @@ hyper-util = "0.1"
 tokio-tungstenite = "0.21.0"
 indexmap = "2"
 indoc = "2"
+inferno = "0.12.0"
 ipnet = "2.10.0"
 itertools = "0.10"
 itoa = "1.0.11"
-jemalloc_pprof = { version = "0.7", features = ["symbolize", "flamegraph"] }
+jemalloc_pprof = "0.6"
 jsonwebtoken = "9"
 lasso = "0.7"
 libc = "0.2"
@@ -126,9 +126,7 @@ measured = { version = "0.0.22", features=["lasso"] }
 measured-process = { version = "0.0.22" }
 memoffset = "0.9"
 nix = { version = "0.27", features = ["dir", "fs", "process", "socket", "signal", "poll"] }
-# Do not update to >= 7.0.0, at least. The update will have a significant impact
-# on compute startup metrics (start_postgres_ms), >= 25% degradation.
-notify = "6.0.0"
+notify = "8.0.0"
 num_cpus = "1.15"
 num-traits = "0.2.15"
 once_cell = "1.13"
@@ -141,7 +139,7 @@ parquet = { version = "53", default-features = false, features = ["zstd"] }
 parquet_derive = "53"
 pbkdf2 = { version = "0.12.1", features = ["simple", "std"] }
 pin-project-lite = "0.2"
-pprof = { version = "0.14", features = ["criterion", "flamegraph", "frame-pointer", "prost-codec"] }
+pprof = { version = "0.14", features = ["criterion", "flamegraph", "frame-pointer", "protobuf", "protobuf-codec"] }
 procfs = "0.16"
 prometheus = {version = "0.13", default-features=false, features = ["process"]} # removes protobuf dependency
 prost = "0.13"
@@ -157,7 +155,6 @@ rpds = "0.13"
 rustc-hash = "1.1.0"
 rustls = { version = "0.23.16", default-features = false }
 rustls-pemfile = "2"
-rustls-pki-types = "1.11"
 scopeguard = "1.1"
 sysinfo = "0.29.2"
 sd-notify = "0.4.1"
@@ -195,7 +192,7 @@ toml = "0.8"
 toml_edit = "0.22"
 tonic = {version = "0.12.3", default-features = false, features = ["channel", "tls", "tls-roots"]}
 tower = { version = "0.5.2", default-features = false }
-tower-http = { version = "0.6.2", features = ["auth", "request-id", "trace"] }
+tower-http = { version = "0.6.2", features = ["request-id", "trace"] }
 
 # This revision uses opentelemetry 0.27. There's no tag for it.
 tower-otel = { git = "https://github.com/mattiapenati/tower-otel", rev = "56a7321053bcb72443888257b622ba0d43a11fcd" }
@@ -221,7 +218,7 @@ zerocopy = { version = "0.7", features = ["derive"] }
 json-structural-diff = { version = "0.2.0" }
 
 ## TODO replace this with tracing
-env_logger = "0.11"
+env_logger = "0.10"
 log = "0.4"
 
 ## Libraries from neondatabase/ git forks, ideally with changes to be upstreamed

Makefile

@@ -11,16 +11,15 @@ ICU_PREFIX_DIR := /usr/local/icu
 #
 BUILD_TYPE ?= debug
 WITH_SANITIZERS ?= no
-PG_CFLAGS = -fsigned-char
 ifeq ($(BUILD_TYPE),release)
 	PG_CONFIGURE_OPTS = --enable-debug --with-openssl
-	PG_CFLAGS += -O2 -g3 $(CFLAGS)
+	PG_CFLAGS = -O2 -g3 $(CFLAGS)
 	PG_LDFLAGS = $(LDFLAGS)
 	# Unfortunately, `--profile=...` is a nightly feature
 	CARGO_BUILD_FLAGS += --release
 else ifeq ($(BUILD_TYPE),debug)
 	PG_CONFIGURE_OPTS = --enable-debug --with-openssl --enable-cassert --enable-depend
-	PG_CFLAGS += -O0 -g3 $(CFLAGS)
+	PG_CFLAGS = -O0 -g3 $(CFLAGS)
 	PG_LDFLAGS = $(LDFLAGS)
 else
 	$(error Bad build type '$(BUILD_TYPE)', see Makefile for options)
@@ -160,8 +159,6 @@ postgres-%: postgres-configure-% \
 	$(MAKE) -C $(POSTGRES_INSTALL_DIR)/build/$*/contrib/pg_visibility install
 	+@echo "Compiling pageinspect $*"
 	$(MAKE) -C $(POSTGRES_INSTALL_DIR)/build/$*/contrib/pageinspect install
-	+@echo "Compiling pg_trgm $*"
-	$(MAKE) -C $(POSTGRES_INSTALL_DIR)/build/$*/contrib/pg_trgm install
 	+@echo "Compiling amcheck $*"
 	$(MAKE) -C $(POSTGRES_INSTALL_DIR)/build/$*/contrib/amcheck install
 	+@echo "Compiling test_decoding $*"

compute/compute-node.Dockerfile

@@ -162,7 +162,7 @@ FROM build-deps AS pg-build
 ARG PG_VERSION
 COPY vendor/postgres-${PG_VERSION:?} postgres
 RUN cd postgres && \
-    export CONFIGURE_CMD="./configure CFLAGS='-O2 -g3 -fsigned-char' --enable-debug --with-openssl --with-uuid=ossp \
+    export CONFIGURE_CMD="./configure CFLAGS='-O2 -g3' --enable-debug --with-openssl --with-uuid=ossp \
     --with-icu --with-libxml --with-libxslt --with-lz4" && \
     if [ "${PG_VERSION:?}" != "v14" ]; then \
         # zstd is available only from PG15
@@ -1055,6 +1055,34 @@ RUN  if [ -d pg_embedding-src ]; then \
         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"
@@ -1352,27 +1380,6 @@ COPY --from=pg_session_jwt-src /ext-src/ /ext-src/
 WORKDIR /ext-src/pg_session_jwt-src
 RUN cargo pgrx install --release
 
-#########################################################################################
-#
-# Layer "pg-anon-pg-build"
-# compile anon extension
-#
-#########################################################################################
-FROM rust-extensions-build-pgrx12 AS pg-anon-pg-build
-ARG PG_VERSION
-COPY --from=pg-build /usr/local/pgsql/ /usr/local/pgsql/
-
-# 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.
-ENV PATH="/usr/local/pgsql/bin/:$PATH"
-RUN wget https://gitlab.com/dalibo/postgresql_anonymizer/-/archive/latest/postgresql_anonymizer-latest.tar.gz -O pg_anon.tar.gz && \
-    mkdir pg_anon-src && cd pg_anon-src && tar xzf ../pg_anon.tar.gz --strip-components=1 -C . && \
-    find /usr/local/pgsql -type f | sed 's|^/usr/local/pgsql/||' > /before.txt && \
-    sed -i 's/pgrx = "0.12.9"/pgrx = { version = "=0.12.9", features = [ "unsafe-postgres" ] }/g' Cargo.toml && \
-    make -j $(getconf _NPROCESSORS_ONLN) extension PG_CONFIG=/usr/local/pgsql/bin/pg_config PGVER=pg$(echo "$PG_VERSION" | sed 's/^v//') && \
-    make -j $(getconf _NPROCESSORS_ONLN) install PG_CONFIG=/usr/local/pgsql/bin/pg_config PGVER=pg$(echo "$PG_VERSION" | sed 's/^v//') && \
-    echo 'trusted = true' >> /usr/local/pgsql/share/extension/anon.control
-
 #########################################################################################
 #
 # Layer "wal2json-build"
@@ -1477,7 +1484,7 @@ WORKDIR /ext-src
 COPY compute/patches/pg_duckdb_v031.patch .
 COPY compute/patches/duckdb_v120.patch .
 # pg_duckdb build requires source dir to be a git repo to get submodules
-# allow neon_superuser to execute some functions that in pg_duckdb are available to superuser only:
+# allow neon_superuser to execute some functions that in pg_duckdb are available to superuser only: 
 # - extension management function duckdb.install_extension()
 # - access to duckdb.extensions table and its sequence
 RUN git clone --depth 1 --branch v0.3.1 https://github.com/duckdb/pg_duckdb.git pg_duckdb-src && \
@@ -1492,8 +1499,8 @@ ARG PG_VERSION
 COPY --from=pg_duckdb-src /ext-src/ /ext-src/
 WORKDIR /ext-src/pg_duckdb-src
 RUN make install -j $(getconf _NPROCESSORS_ONLN) && \
-    echo 'trusted = true' >> /usr/local/pgsql/share/extension/pg_duckdb.control
-
+    echo 'trusted = true' >> /usr/local/pgsql/share/extension/pg_duckdb.control 
+        
 #########################################################################################
 #
 # Layer "pg_repack"
@@ -1670,6 +1677,7 @@ 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/
@@ -1750,15 +1758,15 @@ ARG TARGETARCH
 # test_runner/regress/test_compute_metrics.py
 # See comment on the top of the file regading `echo`, `-e` and `\n`
 RUN if [ "$TARGETARCH" = "amd64" ]; then\
-        postgres_exporter_sha256='59aa4a7bb0f7d361f5e05732f5ed8c03cc08f78449cef5856eadec33a627694b';\
+        postgres_exporter_sha256='027e75dda7af621237ff8f5ac66b78a40b0093595f06768612b92b1374bd3105';\
         pgbouncer_exporter_sha256='c9f7cf8dcff44f0472057e9bf52613d93f3ffbc381ad7547a959daa63c5e84ac';\
         sql_exporter_sha256='38e439732bbf6e28ca4a94d7bc3686d3fa1abdb0050773d5617a9efdb9e64d08';\
     else\
-        postgres_exporter_sha256='d1dedea97f56c6d965837bfd1fbb3e35a3b4a4556f8cccee8bd513d8ee086124';\
+        postgres_exporter_sha256='131a376d25778ff9701a4c81f703f179e0b58db5c2c496e66fa43f8179484786';\
         pgbouncer_exporter_sha256='217c4afd7e6492ae904055bc14fe603552cf9bac458c063407e991d68c519da3';\
         sql_exporter_sha256='11918b00be6e2c3a67564adfdb2414fdcbb15a5db76ea17d1d1a944237a893c6';\
     fi\
-    && curl -sL https://github.com/prometheus-community/postgres_exporter/releases/download/v0.17.1/postgres_exporter-0.17.1.linux-${TARGETARCH}.tar.gz\
+    && curl -sL https://github.com/prometheus-community/postgres_exporter/releases/download/v0.16.0/postgres_exporter-0.16.0.linux-${TARGETARCH}.tar.gz\
      | tar xzf - --strip-components=1 -C.\
     && curl -sL https://github.com/prometheus-community/pgbouncer_exporter/releases/download/v0.10.2/pgbouncer_exporter-0.10.2.linux-${TARGETARCH}.tar.gz\
      | tar xzf - --strip-components=1 -C.\
@@ -1925,7 +1933,6 @@ RUN apt update && \
         locales \
         procps \
         ca-certificates \
-        rsyslog \
         $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
@@ -1971,13 +1978,6 @@ COPY --from=sql_exporter_preprocessor --chmod=0644 /home/nonroot/compute/etc/neo
 # Make the libraries we built available
 RUN echo '/usr/local/lib' >> /etc/ld.so.conf && /sbin/ldconfig
 
-# rsyslog config permissions
-# directory for rsyslogd pid file
-RUN mkdir /var/run/rsyslogd && \
-    chown -R postgres:postgres /var/run/rsyslogd && \
-    chown -R postgres:postgres /etc/rsyslog.d/
-
-
 ENV LANG=en_US.utf8
 USER postgres
 ENTRYPOINT ["/usr/local/bin/compute_ctl"]

compute/etc/neon_collector.jsonnet

@@ -29,7 +29,6 @@
     import 'sql_exporter/lfc_approximate_working_set_size.libsonnet',
     import 'sql_exporter/lfc_approximate_working_set_size_windows.libsonnet',
     import 'sql_exporter/lfc_cache_size_limit.libsonnet',
-    import 'sql_exporter/lfc_chunk_size.libsonnet',
     import 'sql_exporter/lfc_hits.libsonnet',
     import 'sql_exporter/lfc_misses.libsonnet',
     import 'sql_exporter/lfc_used.libsonnet',

compute/etc/sql_exporter/db_total_size.sql

@@ -1,5 +1 @@
-SELECT sum(pg_database_size(datname)) AS total
-FROM pg_database
--- Ignore invalid databases, as we will likely have problems with
--- getting their size from the Pageserver.
-WHERE datconnlimit != -2;
+SELECT sum(pg_database_size(datname)) AS total FROM pg_database;

compute/etc/sql_exporter/lfc_chunk_size.libsonnet

@@ -1,10 +0,0 @@
-{
-  metric_name: 'lfc_chunk_size',
-  type: 'gauge',
-  help: 'LFC chunk size, measured in 8KiB pages',
-  key_labels: null,
-  values: [
-    'lfc_chunk_size_pages',
-  ],
-  query: importstr 'sql_exporter/lfc_chunk_size.sql',
-}

compute/etc/sql_exporter/lfc_chunk_size.sql

@@ -1 +0,0 @@
-SELECT lfc_value AS lfc_chunk_size_pages FROM neon.neon_lfc_stats WHERE lfc_key = 'file_cache_chunk_size_pages';

compute/etc/sql_exporter/pg_stats_userdb.sql

@@ -1,20 +1,10 @@
 -- We export stats for 10 non-system databases. Without this limit it is too
 -- easy to abuse the system by creating lots of databases.
 
-SELECT pg_database_size(datname) AS db_size,
-  deadlocks,
-  tup_inserted AS inserted,
-  tup_updated AS updated,
-  tup_deleted AS deleted,
-  datname
+SELECT pg_database_size(datname) AS db_size, deadlocks, tup_inserted AS inserted,
+  tup_updated AS updated, tup_deleted AS deleted, datname
 FROM pg_stat_database
 WHERE datname IN (
   SELECT datname FROM pg_database
-  -- Ignore invalid databases, as we will likely have problems with
-  -- getting their size from the Pageserver.
-  WHERE datconnlimit != -2
-    AND datname <> 'postgres'
-    AND NOT datistemplate
-  ORDER BY oid
-  LIMIT 10
+  WHERE datname <> 'postgres' AND NOT datistemplate ORDER BY oid LIMIT 10
 );

compute/vm-image-spec-bookworm.yaml

@@ -39,10 +39,6 @@ commands:
     user: nobody
     sysvInitAction: respawn
     shell: '/bin/sql_exporter -config.file=/etc/sql_exporter_autoscaling.yml -web.listen-address=:9499'
-  - name: rsyslogd
-    user: postgres
-    sysvInitAction: respawn
-    shell: '/usr/sbin/rsyslogd -n -i /var/run/rsyslogd/rsyslogd.pid -f /etc/compute_rsyslog.conf'
 shutdownHook: |
   su -p postgres --session-command '/usr/local/bin/pg_ctl stop -D /var/db/postgres/compute/pgdata -m fast --wait -t 10'
 files:
@@ -58,7 +54,7 @@ files:
       # regardless of hostname (ALL)
       #
       # Also allow it to shut down the VM. The fast_import job does that when it's finished.
-      postgres ALL=(root) NOPASSWD: /neonvm/bin/resize-swap, /neonvm/bin/set-disk-quota, /neonvm/bin/poweroff, /usr/sbin/rsyslogd
+      postgres ALL=(root) NOPASSWD: /neonvm/bin/resize-swap, /neonvm/bin/set-disk-quota, /neonvm/bin/poweroff
   - filename: cgconfig.conf
     content: |
       # Configuration for cgroups in VM compute nodes
@@ -73,12 +69,6 @@ files:
           }
           memory {}
       }
-# Create dummy rsyslog config, because it refuses to start without at least one action configured.
-# compute_ctl will rewrite this file with the actual configuration, if needed.
-  - filename: compute_rsyslog.conf
-    content: |
-      *.*    /dev/null
-      $IncludeConfig /etc/rsyslog.d/*.conf
 build: |
   # Build cgroup-tools
   #
@@ -142,12 +132,6 @@ merge: |
   RUN set -e \
       && chmod 0644 /etc/cgconfig.conf
 
-
-  COPY compute_rsyslog.conf /etc/compute_rsyslog.conf
-  RUN chmod 0666 /etc/compute_rsyslog.conf
-  RUN chmod 0666 /var/log/
-
-
   COPY --from=libcgroup-builder /libcgroup-install/bin/*  /usr/bin/
   COPY --from=libcgroup-builder /libcgroup-install/lib/*  /usr/lib/
   COPY --from=libcgroup-builder /libcgroup-install/sbin/* /usr/sbin/

compute/vm-image-spec-bullseye.yaml

@@ -39,10 +39,6 @@ commands:
     user: nobody
     sysvInitAction: respawn
     shell: '/bin/sql_exporter -config.file=/etc/sql_exporter_autoscaling.yml -web.listen-address=:9499'
-  - name: rsyslogd
-    user: postgres
-    sysvInitAction: respawn
-    shell: '/usr/sbin/rsyslogd -n -i /var/run/rsyslogd/rsyslogd.pid -f /etc/compute_rsyslog.conf'
 shutdownHook: |
   su -p postgres --session-command '/usr/local/bin/pg_ctl stop -D /var/db/postgres/compute/pgdata -m fast --wait -t 10'
 files:
@@ -58,7 +54,7 @@ files:
       # regardless of hostname (ALL)
       #
       # Also allow it to shut down the VM. The fast_import job does that when it's finished.
-      postgres ALL=(root) NOPASSWD: /neonvm/bin/resize-swap, /neonvm/bin/set-disk-quota, /neonvm/bin/poweroff, /usr/sbin/rsyslogd
+      postgres ALL=(root) NOPASSWD: /neonvm/bin/resize-swap, /neonvm/bin/set-disk-quota, /neonvm/bin/poweroff
   - filename: cgconfig.conf
     content: |
       # Configuration for cgroups in VM compute nodes
@@ -73,12 +69,6 @@ files:
           }
           memory {}
       }
-# Create dummy rsyslog config, because it refuses to start without at least one action configured.
-# compute_ctl will rewrite this file with the actual configuration, if needed.
-  - filename: compute_rsyslog.conf
-    content: |
-      *.*    /dev/null
-      $IncludeConfig /etc/rsyslog.d/*.conf
 build: |
   # Build cgroup-tools
   #
@@ -138,11 +128,6 @@ merge: |
   RUN set -e \
       && chmod 0644 /etc/cgconfig.conf
 
-  COPY compute_rsyslog.conf /etc/compute_rsyslog.conf
-  RUN chmod 0666 /etc/compute_rsyslog.conf
-  RUN chmod 0666 /var/log/
-
-
   COPY --from=libcgroup-builder /libcgroup-install/bin/*  /usr/bin/
   COPY --from=libcgroup-builder /libcgroup-install/lib/*  /usr/lib/
   COPY --from=libcgroup-builder /libcgroup-install/sbin/* /usr/sbin/

compute_tools/Cargo.toml

@@ -17,7 +17,6 @@ aws-sdk-kms.workspace = true
 aws-smithy-types.workspace = true
 anyhow.workspace = true
 axum = { workspace = true, features = [] }
-axum-extra.workspace = true
 camino.workspace = true
 chrono.workspace = true
 cfg-if.workspace = true
@@ -26,7 +25,6 @@ fail.workspace = true
 flate2.workspace = true
 futures.workspace = true
 http.workspace = true
-jsonwebtoken.workspace = true
 metrics.workspace = true
 nix.workspace = true
 notify.workspace = true

compute_tools/src/bin/compute_ctl.rs

@@ -33,27 +33,39 @@
 //!             -b /usr/local/bin/postgres \
 //!             -r http://pg-ext-s3-gateway \
 //! ```
+use std::collections::HashMap;
 use std::ffi::OsString;
 use std::fs::File;
 use std::path::Path;
 use std::process::exit;
-use std::sync::mpsc;
+use std::str::FromStr;
+use std::sync::atomic::Ordering;
+use std::sync::{Arc, Condvar, Mutex, RwLock, mpsc};
 use std::thread;
 use std::time::Duration;
 
 use anyhow::{Context, Result};
+use chrono::Utc;
 use clap::Parser;
-use compute_api::responses::ComputeCtlConfig;
+use compute_api::responses::{ComputeCtlConfig, ComputeStatus};
 use compute_api::spec::ComputeSpec;
-use compute_tools::compute::{ComputeNode, ComputeNodeParams, forward_termination_signal};
+use compute_tools::compute::{
+    ComputeNode, ComputeState, PG_PID, ParsedSpec, forward_termination_signal,
+};
+use compute_tools::configurator::launch_configurator;
+use compute_tools::disk_quota::set_disk_quota;
 use compute_tools::extension_server::get_pg_version_string;
+use compute_tools::http::server::Server;
 use compute_tools::logger::*;
+use compute_tools::lsn_lease::launch_lsn_lease_bg_task_for_static;
+use compute_tools::monitor::launch_monitor;
 use compute_tools::params::*;
 use compute_tools::spec::*;
+use compute_tools::swap::resize_swap;
 use rlimit::{Resource, setrlimit};
 use signal_hook::consts::{SIGINT, SIGQUIT, SIGTERM};
 use signal_hook::iterator::Signals;
-use tracing::{error, info};
+use tracing::{error, info, warn};
 use url::Url;
 use utils::failpoint_support;
 
@@ -152,41 +164,29 @@ fn main() -> Result<()> {
     // 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 (pg_handle, start_pg_result) = {
+        // Enter startup tracing context
+        let _startup_context_guard = startup_context_from_env();
 
-    let cli_spec = try_spec_from_cli(&cli)?;
+        let cli_spec = try_spec_from_cli(&cli)?;
 
-    let compute_node = ComputeNode::new(
-        ComputeNodeParams {
-            compute_id: cli.compute_id,
-            connstr,
-            pgdata: cli.pgdata.clone(),
-            pgbin: cli.pgbin.clone(),
-            pgversion: get_pg_version_string(&cli.pgbin),
-            external_http_port: cli.external_http_port,
-            internal_http_port: cli.internal_http_port,
-            ext_remote_storage: cli.remote_ext_config.clone(),
-            resize_swap_on_bind: cli.resize_swap_on_bind,
-            set_disk_quota_for_fs: cli.set_disk_quota_for_fs,
-            #[cfg(target_os = "linux")]
-            filecache_connstr: cli.filecache_connstr,
-            #[cfg(target_os = "linux")]
-            cgroup: cli.cgroup,
-            #[cfg(target_os = "linux")]
-            vm_monitor_addr: cli.vm_monitor_addr,
-            build_tag,
+        let compute = wait_spec(build_tag, &cli, cli_spec)?;
 
-            live_config_allowed: cli_spec.live_config_allowed,
-        },
-        cli_spec.spec,
-        cli_spec.compute_ctl_config,
-    )?;
+        start_postgres(&cli, compute)?
 
-    let exit_code = compute_node.run()?;
+        // Startup is finished, exit the startup tracing span
+    };
+
+    // PostgreSQL is now running, if startup was successful. Wait until it exits.
+    let wait_pg_result = wait_postgres(pg_handle)?;
+
+    let delay_exit = cleanup_after_postgres_exit(start_pg_result)?;
+
+    maybe_delay_exit(delay_exit);
 
     scenario.teardown();
 
-    deinit_and_exit(exit_code);
+    deinit_and_exit(wait_pg_result);
 }
 
 async fn init() -> Result<String> {
@@ -207,6 +207,56 @@ async fn init() -> Result<String> {
     Ok(build_tag)
 }
 
+fn startup_context_from_env() -> Option<opentelemetry::ContextGuard> {
+    // Extract OpenTelemetry context for the startup actions from the
+    // TRACEPARENT and TRACESTATE env variables, and attach it to the current
+    // tracing context.
+    //
+    // This is used to propagate the context for the 'start_compute' operation
+    // from the neon control plane. This allows linking together the wider
+    // 'start_compute' operation that creates the compute container, with the
+    // startup actions here within the container.
+    //
+    // There is no standard for passing context in env variables, but a lot of
+    // tools use TRACEPARENT/TRACESTATE, so we use that convention too. See
+    // https://github.com/open-telemetry/opentelemetry-specification/issues/740
+    //
+    // Switch to the startup context here, and exit it once the startup has
+    // completed and Postgres is up and running.
+    //
+    // If this pod is pre-created without binding it to any particular endpoint
+    // yet, this isn't the right place to enter the startup context. In that
+    // case, the control plane should pass the tracing context as part of the
+    // /configure API call.
+    //
+    // NOTE: This is supposed to only cover the *startup* actions. Once
+    // postgres is configured and up-and-running, we exit this span. Any other
+    // actions that are performed on incoming HTTP requests, for example, are
+    // performed in separate spans.
+    //
+    // XXX: If the pod is restarted, we perform the startup actions in the same
+    // context as the original startup actions, which probably doesn't make
+    // sense.
+    let mut startup_tracing_carrier: HashMap<String, String> = HashMap::new();
+    if let Ok(val) = std::env::var("TRACEPARENT") {
+        startup_tracing_carrier.insert("traceparent".to_string(), val);
+    }
+    if let Ok(val) = std::env::var("TRACESTATE") {
+        startup_tracing_carrier.insert("tracestate".to_string(), val);
+    }
+    if !startup_tracing_carrier.is_empty() {
+        use opentelemetry::propagation::TextMapPropagator;
+        use opentelemetry_sdk::propagation::TraceContextPropagator;
+        let guard = TraceContextPropagator::new()
+            .extract(&startup_tracing_carrier)
+            .attach();
+        info!("startup tracing context attached");
+        Some(guard)
+    } else {
+        None
+    }
+}
+
 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 {
@@ -257,7 +307,357 @@ struct CliSpecParams {
     live_config_allowed: bool,
 }
 
-fn deinit_and_exit(exit_code: Option<i32>) -> ! {
+fn wait_spec(
+    build_tag: String,
+    cli: &Cli,
+    CliSpecParams {
+        spec,
+        live_config_allowed,
+        compute_ctl_config: _,
+    }: CliSpecParams,
+) -> Result<Arc<ComputeNode>> {
+    let mut new_state = ComputeState::new();
+    let spec_set;
+
+    if let Some(spec) = spec {
+        let pspec = ParsedSpec::try_from(spec).map_err(|msg| anyhow::anyhow!(msg))?;
+        info!("new pspec.spec: {:?}", pspec.spec);
+        new_state.pspec = Some(pspec);
+        spec_set = true;
+    } else {
+        spec_set = false;
+    }
+    let connstr = Url::parse(&cli.connstr).context("cannot parse connstr as a URL")?;
+    let conn_conf = postgres::config::Config::from_str(connstr.as_str())
+        .context("cannot build postgres config from connstr")?;
+    let tokio_conn_conf = tokio_postgres::config::Config::from_str(connstr.as_str())
+        .context("cannot build tokio postgres config from connstr")?;
+    let compute_node = ComputeNode {
+        compute_id: cli.compute_id.clone(),
+        connstr,
+        conn_conf,
+        tokio_conn_conf,
+        pgdata: cli.pgdata.clone(),
+        pgbin: cli.pgbin.clone(),
+        pgversion: get_pg_version_string(&cli.pgbin),
+        external_http_port: cli.external_http_port,
+        internal_http_port: cli.internal_http_port,
+        live_config_allowed,
+        state: Mutex::new(new_state),
+        state_changed: Condvar::new(),
+        ext_remote_storage: cli.remote_ext_config.clone(),
+        ext_download_progress: RwLock::new(HashMap::new()),
+        build_tag,
+    };
+    let compute = Arc::new(compute_node);
+
+    // If this is a pooled VM, prewarm before starting HTTP server and becoming
+    // available for binding. Prewarming helps Postgres start quicker later,
+    // because QEMU will already have its memory allocated from the host, and
+    // the necessary binaries will already be cached.
+    if !spec_set {
+        compute.prewarm_postgres()?;
+    }
+
+    // Launch the external HTTP server first, so that we can serve control plane
+    // requests while configuration is still in progress.
+    Server::External(cli.external_http_port).launch(&compute);
+
+    // The internal HTTP server could be launched later, but there isn't much
+    // sense in waiting.
+    Server::Internal(cli.internal_http_port).launch(&compute);
+
+    if !spec_set {
+        // No spec provided, hang waiting for it.
+        info!("no compute spec provided, waiting");
+
+        let mut state = compute.state.lock().unwrap();
+        while state.status != ComputeStatus::ConfigurationPending {
+            state = compute.state_changed.wait(state).unwrap();
+
+            if state.status == ComputeStatus::ConfigurationPending {
+                info!("got spec, continue configuration");
+                // Spec is already set by the http server handler.
+                break;
+            }
+        }
+
+        // Record for how long we slept waiting for the spec.
+        let now = Utc::now();
+        state.metrics.wait_for_spec_ms = now
+            .signed_duration_since(state.start_time)
+            .to_std()
+            .unwrap()
+            .as_millis() as u64;
+
+        // Reset start time, so that the total startup time that is calculated later will
+        // not include the time that we waited for the spec.
+        state.start_time = now;
+    }
+
+    launch_lsn_lease_bg_task_for_static(&compute);
+
+    Ok(compute)
+}
+
+fn start_postgres(
+    cli: &Cli,
+    compute: Arc<ComputeNode>,
+) -> Result<(Option<PostgresHandle>, StartPostgresResult)> {
+    // We got all we need, update the state.
+    let mut state = compute.state.lock().unwrap();
+
+    // Create a tracing span for the startup operation.
+    //
+    // We could otherwise just annotate the function with #[instrument], but if
+    // we're being configured from a /configure HTTP request, we want the
+    // startup to be considered part of the /configure request.
+    let _this_entered = {
+        // Temporarily enter the /configure request's span, so that the new span
+        // becomes its child.
+        let _parent_entered = state.startup_span.take().map(|p| p.entered());
+
+        tracing::info_span!("start_postgres")
+    }
+    .entered();
+
+    state.set_status(ComputeStatus::Init, &compute.state_changed);
+
+    info!(
+        "running compute with features: {:?}",
+        state.pspec.as_ref().unwrap().spec.features
+    );
+    // before we release the mutex, fetch some parameters for later.
+    let &ComputeSpec {
+        swap_size_bytes,
+        disk_quota_bytes,
+        #[cfg(target_os = "linux")]
+        disable_lfc_resizing,
+        ..
+    } = &state.pspec.as_ref().unwrap().spec;
+    drop(state);
+
+    // Launch remaining service threads
+    let _monitor_handle = launch_monitor(&compute);
+    let _configurator_handle = launch_configurator(&compute);
+
+    let mut prestartup_failed = false;
+    let mut delay_exit = false;
+
+    // Resize swap to the desired size if the compute spec says so
+    if let (Some(size_bytes), true) = (swap_size_bytes, cli.resize_swap_on_bind) {
+        // To avoid 'swapoff' hitting postgres startup, we need to run resize-swap to completion
+        // *before* starting postgres.
+        //
+        // In theory, we could do this asynchronously if SkipSwapon was enabled for VMs, but this
+        // carries a risk of introducing hard-to-debug issues - e.g. if postgres sometimes gets
+        // OOM-killed during startup because swap wasn't available yet.
+        match resize_swap(size_bytes) {
+            Ok(()) => {
+                let size_mib = size_bytes as f32 / (1 << 20) as f32; // just for more coherent display.
+                info!(%size_bytes, %size_mib, "resized swap");
+            }
+            Err(err) => {
+                let err = err.context("failed to resize swap");
+                error!("{err:#}");
+
+                // Mark compute startup as failed; don't try to start postgres, and report this
+                // error to the control plane when it next asks.
+                prestartup_failed = true;
+                compute.set_failed_status(err);
+                delay_exit = true;
+            }
+        }
+    }
+
+    // Set disk quota if the compute spec says so
+    if let (Some(disk_quota_bytes), Some(disk_quota_fs_mountpoint)) =
+        (disk_quota_bytes, cli.set_disk_quota_for_fs.as_ref())
+    {
+        match set_disk_quota(disk_quota_bytes, disk_quota_fs_mountpoint) {
+            Ok(()) => {
+                let size_mib = disk_quota_bytes as f32 / (1 << 20) as f32; // just for more coherent display.
+                info!(%disk_quota_bytes, %size_mib, "set disk quota");
+            }
+            Err(err) => {
+                let err = err.context("failed to set disk quota");
+                error!("{err:#}");
+
+                // Mark compute startup as failed; don't try to start postgres, and report this
+                // error to the control plane when it next asks.
+                prestartup_failed = true;
+                compute.set_failed_status(err);
+                delay_exit = true;
+            }
+        }
+    }
+
+    // Start Postgres
+    let mut pg = None;
+    if !prestartup_failed {
+        pg = match compute.start_compute() {
+            Ok(pg) => {
+                info!(postmaster_pid = %pg.0.id(), "Postgres was started");
+                Some(pg)
+            }
+            Err(err) => {
+                error!("could not start the compute node: {:#}", err);
+                compute.set_failed_status(err);
+                delay_exit = true;
+                None
+            }
+        };
+    } else {
+        warn!("skipping postgres startup because pre-startup step failed");
+    }
+
+    // Start the vm-monitor if directed to. The vm-monitor only runs on linux
+    // because it requires cgroups.
+    cfg_if::cfg_if! {
+        if #[cfg(target_os = "linux")] {
+            use std::env;
+            use tokio_util::sync::CancellationToken;
+
+            // This token is used internally by the monitor to clean up all threads
+            let token = CancellationToken::new();
+
+            // don't pass postgres connection string to vm-monitor if we don't want it to resize LFC
+            let pgconnstr = if disable_lfc_resizing.unwrap_or(false) {
+                None
+            } else {
+                Some(cli.filecache_connstr.clone())
+            };
+
+            let vm_monitor = if env::var_os("AUTOSCALING").is_some() {
+                let vm_monitor = tokio::spawn(vm_monitor::start(
+                    Box::leak(Box::new(vm_monitor::Args {
+                        cgroup: Some(cli.cgroup.clone()),
+                        pgconnstr,
+                        addr: cli.vm_monitor_addr.clone(),
+                    })),
+                    token.clone(),
+                ));
+                Some(vm_monitor)
+            } else {
+                None
+            };
+        }
+    }
+
+    Ok((
+        pg,
+        StartPostgresResult {
+            delay_exit,
+            compute,
+            #[cfg(target_os = "linux")]
+            token,
+            #[cfg(target_os = "linux")]
+            vm_monitor,
+        },
+    ))
+}
+
+type PostgresHandle = (std::process::Child, tokio::task::JoinHandle<Result<()>>);
+
+struct StartPostgresResult {
+    delay_exit: bool,
+    // passed through from WaitSpecResult
+    compute: Arc<ComputeNode>,
+
+    #[cfg(target_os = "linux")]
+    token: tokio_util::sync::CancellationToken,
+    #[cfg(target_os = "linux")]
+    vm_monitor: Option<tokio::task::JoinHandle<Result<()>>>,
+}
+
+fn wait_postgres(pg: Option<PostgresHandle>) -> Result<WaitPostgresResult> {
+    // Wait for the child Postgres process forever. In this state Ctrl+C will
+    // propagate to Postgres and it will be shut down as well.
+    let mut exit_code = None;
+    if let Some((mut pg, logs_handle)) = pg {
+        info!(postmaster_pid = %pg.id(), "Waiting for Postgres to exit");
+
+        let ecode = pg
+            .wait()
+            .expect("failed to start waiting on Postgres process");
+        PG_PID.store(0, Ordering::SeqCst);
+
+        // Process has exited. Wait for the log collecting task to finish.
+        let _ = tokio::runtime::Handle::current()
+            .block_on(logs_handle)
+            .map_err(|e| tracing::error!("log task panicked: {:?}", e));
+
+        info!("Postgres exited with code {}, shutting down", ecode);
+        exit_code = ecode.code()
+    }
+
+    Ok(WaitPostgresResult { exit_code })
+}
+
+struct WaitPostgresResult {
+    exit_code: Option<i32>,
+}
+
+fn cleanup_after_postgres_exit(
+    StartPostgresResult {
+        mut delay_exit,
+        compute,
+        #[cfg(target_os = "linux")]
+        vm_monitor,
+        #[cfg(target_os = "linux")]
+        token,
+    }: StartPostgresResult,
+) -> Result<bool> {
+    // Terminate the vm_monitor so it releases the file watcher on
+    // /sys/fs/cgroup/neon-postgres.
+    // Note: the vm-monitor only runs on linux because it requires cgroups.
+    cfg_if::cfg_if! {
+        if #[cfg(target_os = "linux")] {
+            if let Some(handle) = vm_monitor {
+                // Kills all threads spawned by the monitor
+                token.cancel();
+                // Kills the actual task running the monitor
+                handle.abort();
+            }
+        }
+    }
+
+    // Maybe sync safekeepers again, to speed up next startup
+    let compute_state = compute.state.lock().unwrap().clone();
+    let pspec = compute_state.pspec.as_ref().expect("spec must be set");
+    if matches!(pspec.spec.mode, compute_api::spec::ComputeMode::Primary) {
+        info!("syncing safekeepers on shutdown");
+        let storage_auth_token = pspec.storage_auth_token.clone();
+        let lsn = compute.sync_safekeepers(storage_auth_token)?;
+        info!("synced safekeepers at lsn {lsn}");
+    }
+
+    let mut state = compute.state.lock().unwrap();
+    if state.status == ComputeStatus::TerminationPending {
+        state.status = ComputeStatus::Terminated;
+        compute.state_changed.notify_all();
+        // we were asked to terminate gracefully, don't exit to avoid restart
+        delay_exit = true
+    }
+    drop(state);
+
+    if let Err(err) = compute.check_for_core_dumps() {
+        error!("error while checking for core dumps: {err:?}");
+    }
+
+    Ok(delay_exit)
+}
+
+fn maybe_delay_exit(delay_exit: bool) {
+    // If launch failed, keep serving HTTP requests for a while, so the cloud
+    // control plane can get the actual error.
+    if delay_exit {
+        info!("giving control plane 30s to collect the error before shutdown");
+        thread::sleep(Duration::from_secs(30));
+    }
+}
+
+fn deinit_and_exit(WaitPostgresResult { exit_code }: WaitPostgresResult) -> ! {
     // Shutdown trace pipeline gracefully, so that it has a chance to send any
     // pending traces before we exit. Shutting down OTEL tracing provider may
     // hang for quite some time, see, for example:

compute_tools/src/catalog.rs

@@ -58,14 +58,14 @@ pub async fn get_database_schema(
     compute: &Arc<ComputeNode>,
     dbname: &str,
 ) -> Result<impl Stream<Item = Result<bytes::Bytes, std::io::Error>> + use<>, SchemaDumpError> {
-    let pgbin = &compute.params.pgbin;
+    let pgbin = &compute.pgbin;
     let basepath = Path::new(pgbin).parent().unwrap();
     let pgdump = basepath.join("pg_dump");
 
     // Replace the DB in the connection string and disable it to parts.
     // This is the only option to handle DBs with special characters.
-    let conf = postgres_conf_for_db(&compute.params.connstr, dbname)
-        .map_err(|_| SchemaDumpError::Unexpected)?;
+    let conf =
+        postgres_conf_for_db(&compute.connstr, dbname).map_err(|_| SchemaDumpError::Unexpected)?;
     let host = conf
         .get_hosts()
         .first()

compute_tools/src/config.rs

@@ -1,16 +1,12 @@
-use anyhow::Result;
-use std::fmt::Write as FmtWrite;
 use std::fs::{File, OpenOptions};
 use std::io;
-use std::io::Write;
 use std::io::prelude::*;
 use std::path::Path;
 
-use compute_api::spec::{ComputeAudit, ComputeMode, ComputeSpec, GenericOption};
+use anyhow::Result;
+use compute_api::spec::{ComputeMode, ComputeSpec, GenericOption};
 
-use crate::pg_helpers::{
-    GenericOptionExt, GenericOptionsSearch, PgOptionsSerialize, escape_conf_value,
-};
+use crate::pg_helpers::{GenericOptionExt, PgOptionsSerialize, escape_conf_value};
 
 /// Check that `line` is inside a text file and put it there if it is not.
 /// Create file if it doesn't exist.
@@ -59,20 +55,10 @@ pub fn write_postgres_conf(
         writeln!(file, "neon.stripe_size={stripe_size}")?;
     }
     if !spec.safekeeper_connstrings.is_empty() {
-        let mut neon_safekeepers_value = String::new();
-        tracing::info!(
-            "safekeepers_connstrings is not zero, gen: {:?}",
-            spec.safekeepers_generation
-        );
-        // If generation is given, prepend sk list with g#number:
-        if let Some(generation) = spec.safekeepers_generation {
-            write!(neon_safekeepers_value, "g#{}:", generation)?;
-        }
-        neon_safekeepers_value.push_str(&spec.safekeeper_connstrings.join(","));
         writeln!(
             file,
             "neon.safekeepers={}",
-            escape_conf_value(&neon_safekeepers_value)
+            escape_conf_value(&spec.safekeeper_connstrings.join(","))
         )?;
     }
     if let Some(s) = &spec.tenant_id {
@@ -140,54 +126,6 @@ pub fn write_postgres_conf(
         writeln!(file, "# Managed by compute_ctl: end")?;
     }
 
-    // If audit logging is enabled, configure pgaudit.
-    //
-    // Note, that this is called after the settings from spec are written.
-    // This way we always override the settings from the spec
-    // and don't allow the user or the control plane admin to change them.
-    if let ComputeAudit::Hipaa = spec.audit_log_level {
-        writeln!(file, "# Managed by compute_ctl audit settings: begin")?;
-        // This log level is very verbose
-        // but this is necessary for HIPAA compliance.
-        writeln!(file, "pgaudit.log='all'")?;
-        writeln!(file, "pgaudit.log_parameter=on")?;
-        // Disable logging of catalog queries
-        // The catalog doesn't contain sensitive data, so we don't need to audit it.
-        writeln!(file, "pgaudit.log_catalog=off")?;
-        // Set log rotation to 5 minutes
-        // TODO: tune this after performance testing
-        writeln!(file, "pgaudit.log_rotation_age=5")?;
-
-        // Add audit shared_preload_libraries, if they are not present.
-        //
-        // The caller who sets the flag is responsible for ensuring that the necessary
-        // shared_preload_libraries are present in the compute image,
-        // otherwise the compute start will fail.
-        if let Some(libs) = spec.cluster.settings.find("shared_preload_libraries") {
-            let mut extra_shared_preload_libraries = String::new();
-            if !libs.contains("pgaudit") {
-                extra_shared_preload_libraries.push_str(",pgaudit");
-            }
-            if !libs.contains("pgauditlogtofile") {
-                extra_shared_preload_libraries.push_str(",pgauditlogtofile");
-            }
-            writeln!(
-                file,
-                "shared_preload_libraries='{}{}'",
-                libs, extra_shared_preload_libraries
-            )?;
-        } else {
-            // Typically, this should be unreacheable,
-            // because we always set at least some shared_preload_libraries in the spec
-            // but let's handle it explicitly anyway.
-            writeln!(
-                file,
-                "shared_preload_libraries='neon,pgaudit,pgauditlogtofile'"
-            )?;
-        }
-        writeln!(file, "# Managed by compute_ctl audit settings: end")?;
-    }
-
     writeln!(file, "neon.extension_server_port={}", extension_server_port)?;
 
     if spec.drop_subscriptions_before_start {

compute_tools/src/config_template/compute_audit_rsyslog_template.conf

@@ -1,10 +0,0 @@
-# Load imfile module to read log files
-module(load="imfile")
-
-# Input configuration for log files in the specified directory
-# Replace {log_directory} with the directory containing the log files
-input(type="imfile" File="{log_directory}/*.log" Tag="{tag}" Severity="info" Facility="local0")
-global(workDirectory="/var/log")
-
-# Forward logs to remote syslog server
-*.* @@{remote_endpoint}

compute_tools/src/extension_server.rs

@@ -202,24 +202,8 @@ pub async fn download_extension(
     // move contents of the libdir / sharedir in unzipped archive to the correct local paths
     for paths in [sharedir_paths, libdir_paths] {
         let (zip_dir, real_dir) = paths;
-
-        let dir = match std::fs::read_dir(&zip_dir) {
-            Ok(dir) => dir,
-            Err(e) => match e.kind() {
-                // In the event of a SQL-only extension, there would be nothing
-                // to move from the lib/ directory, so note that in the log and
-                // move on.
-                std::io::ErrorKind::NotFound => {
-                    info!("nothing to move from {}", zip_dir);
-                    continue;
-                }
-                _ => return Err(anyhow::anyhow!(e)),
-            },
-        };
-
         info!("mv {zip_dir:?}/*  {real_dir:?}");
-
-        for file in dir {
+        for file in std::fs::read_dir(zip_dir)? {
             let old_file = file?.path();
             let new_file =
                 Path::new(&real_dir).join(old_file.file_name().context("error parsing file")?);
@@ -269,31 +253,27 @@ pub fn create_control_files(remote_extensions: &RemoteExtSpec, pgbin: &str) {
     }
 }
 
-// Do request to extension storage proxy, e.g.,
+// Do request to extension storage proxy, i.e.
 // curl http://pg-ext-s3-gateway/latest/v15/extensions/anon.tar.zst
-// using HTTP GET and return the response body as bytes.
+// using HHTP GET
+// and return the response body as bytes
+//
 async fn download_extension_tar(ext_remote_storage: &str, ext_path: &str) -> Result<Bytes> {
     let uri = format!("{}/{}", ext_remote_storage, ext_path);
-    let filename = Path::new(ext_path)
-        .file_name()
-        .unwrap_or_else(|| std::ffi::OsStr::new("unknown"))
-        .to_str()
-        .unwrap_or("unknown")
-        .to_string();
 
-    info!("Downloading extension file '{}' from uri {}", filename, uri);
+    info!("Download extension {} from uri {}", ext_path, uri);
 
     match do_extension_server_request(&uri).await {
         Ok(resp) => {
             info!("Successfully downloaded remote extension data {}", ext_path);
             REMOTE_EXT_REQUESTS_TOTAL
-                .with_label_values(&[&StatusCode::OK.to_string(), &filename])
+                .with_label_values(&[&StatusCode::OK.to_string()])
                 .inc();
             Ok(resp)
         }
         Err((msg, status)) => {
             REMOTE_EXT_REQUESTS_TOTAL
-                .with_label_values(&[&status, &filename])
+                .with_label_values(&[&status])
                 .inc();
             bail!(msg);
         }

compute_tools/src/http/extract/mod.rs

@@ -1,9 +1,7 @@
 pub(crate) mod json;
 pub(crate) mod path;
 pub(crate) mod query;
-pub(crate) mod request_id;
 
 pub(crate) use json::Json;
 pub(crate) use path::Path;
 pub(crate) use query::Query;
-pub(crate) use request_id::RequestId;

compute_tools/src/http/extract/request_id.rs

@@ -1,86 +0,0 @@
-use std::{
-    fmt::Display,
-    ops::{Deref, DerefMut},
-};
-
-use axum::{extract::FromRequestParts, response::IntoResponse};
-use http::{StatusCode, request::Parts};
-
-use crate::http::{JsonResponse, headers::X_REQUEST_ID};
-
-/// Extract the request ID from the `X-Request-Id` header.
-#[derive(Debug, Clone, Default)]
-pub(crate) struct RequestId(pub String);
-
-#[derive(Debug)]
-/// Rejection used for [`RequestId`].
-///
-/// Contains one variant for each way the [`RequestId`] extractor can
-/// fail.
-pub(crate) enum RequestIdRejection {
-    /// The request is missing the header.
-    MissingRequestId,
-
-    /// The value of the header is invalid UTF-8.
-    InvalidUtf8,
-}
-
-impl RequestIdRejection {
-    pub fn status(&self) -> StatusCode {
-        match self {
-            RequestIdRejection::MissingRequestId => StatusCode::INTERNAL_SERVER_ERROR,
-            RequestIdRejection::InvalidUtf8 => StatusCode::BAD_REQUEST,
-        }
-    }
-
-    pub fn message(&self) -> String {
-        match self {
-            RequestIdRejection::MissingRequestId => "request ID is missing",
-            RequestIdRejection::InvalidUtf8 => "request ID is invalid UTF-8",
-        }
-        .to_string()
-    }
-}
-
-impl IntoResponse for RequestIdRejection {
-    fn into_response(self) -> axum::response::Response {
-        JsonResponse::error(self.status(), self.message())
-    }
-}
-
-impl<S> FromRequestParts<S> for RequestId
-where
-    S: Send + Sync,
-{
-    type Rejection = RequestIdRejection;
-
-    async fn from_request_parts(parts: &mut Parts, _state: &S) -> Result<Self, Self::Rejection> {
-        match parts.headers.get(X_REQUEST_ID) {
-            Some(value) => match value.to_str() {
-                Ok(request_id) => Ok(Self(request_id.to_string())),
-                Err(_) => Err(RequestIdRejection::InvalidUtf8),
-            },
-            None => Err(RequestIdRejection::MissingRequestId),
-        }
-    }
-}
-
-impl Deref for RequestId {
-    type Target = String;
-
-    fn deref(&self) -> &Self::Target {
-        &self.0
-    }
-}
-
-impl DerefMut for RequestId {
-    fn deref_mut(&mut self) -> &mut Self::Target {
-        &mut self.0
-    }
-}
-
-impl Display for RequestId {
-    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
-        f.write_str(&self.0)
-    }
-}

compute_tools/src/http/headers.rs

@@ -1,2 +0,0 @@
-/// Constant for `X-Request-Id` header.
-pub const X_REQUEST_ID: &str = "x-request-id";

compute_tools/src/http/middleware/authorize.rs

@@ -1,145 +0,0 @@
-use std::{collections::HashSet, net::SocketAddr};
-
-use anyhow::{Result, anyhow};
-use axum::{RequestExt, body::Body, extract::ConnectInfo};
-use axum_extra::{
-    TypedHeader,
-    headers::{Authorization, authorization::Bearer},
-};
-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::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,
-    jwks: JwkSet,
-    validation: Validation,
-}
-
-impl Authorize {
-    pub fn new(compute_id: String, jwks: JwkSet) -> Self {
-        let mut validation = Validation::new(Algorithm::EdDSA);
-        // Nothing is currently required
-        validation.required_spec_claims = HashSet::new();
-        validation.validate_exp = true;
-        // Unused by the control plane
-        validation.validate_aud = false;
-        // Unused by the control plane
-        validation.validate_nbf = false;
-
-        Self {
-            compute_id,
-            jwks,
-            validation,
-        }
-    }
-}
-
-impl AsyncAuthorizeRequest<Body> for Authorize {
-    type RequestBody = Body;
-    type ResponseBody = Body;
-    type Future = BoxFuture<'static, Result<Request<Body>, Response<Self::ResponseBody>>>;
-
-    fn authorize(&mut self, mut request: Request<Body>) -> Self::Future {
-        let compute_id = self.compute_id.clone();
-        let jwks = self.jwks.clone();
-        let validation = self.validation.clone();
-
-        Box::pin(async move {
-            let request_id = request.extract_parts::<RequestId>().await.unwrap();
-
-            // TODO: Remove this check after a successful rollout
-            if jwks.keys.is_empty() {
-                warn!(%request_id, "Authorization has not been configured");
-
-                return Ok(request);
-            }
-
-            let connect_info = request
-                .extract_parts::<ConnectInfo<SocketAddr>>()
-                .await
-                .unwrap();
-
-            // In the event the request is coming from the loopback interface,
-            // allow all requests
-            if connect_info.ip().is_loopback() {
-                warn!(%request_id, "Bypassed authorization because request is coming from the loopback interface");
-
-                return Ok(request);
-            }
-
-            let TypedHeader(Authorization(bearer)) = request
-                .extract_parts::<TypedHeader<Authorization<Bearer>>>()
-                .await
-                .map_err(|_| {
-                    JsonResponse::error(StatusCode::BAD_REQUEST, "invalid authorization token")
-                })?;
-
-            let data = match Self::verify(&jwks, bearer.token(), &validation) {
-                Ok(claims) => claims,
-                Err(e) => return Err(JsonResponse::error(StatusCode::UNAUTHORIZED, e)),
-            };
-
-            if data.claims.compute_id != compute_id {
-                return Err(JsonResponse::error(
-                    StatusCode::UNAUTHORIZED,
-                    "invalid claims in authorization token",
-                ));
-            }
-
-            // Make claims available to any subsequent middleware or request
-            // handlers
-            request.extensions_mut().insert(data.claims);
-
-            Ok(request)
-        })
-    }
-}
-
-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<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 {}: {}",
-                        jwk.common.key_id.as_ref().unwrap(),
-                        e
-                    );
-
-                    continue;
-                }
-            };
-
-            match jsonwebtoken::decode::<Claims>(token, &decoding_key, validation) {
-                Ok(data) => return Ok(data),
-                Err(e) => {
-                    warn!(
-                        "Failed to decode authorization token using {}: {}",
-                        jwk.common.key_id.as_ref().unwrap(),
-                        e
-                    );
-
-                    continue;
-                }
-            }
-        }
-
-        Err(anyhow!("Failed to verify authorization token"))
-    }
-}

compute_tools/src/http/middleware/mod.rs

@@ -1,2 +0,0 @@
-pub(in crate::http) mod authorize;
-pub(in crate::http) mod request_id;

compute_tools/src/http/middleware/request_id.rs

@@ -1,16 +0,0 @@
-use axum::{extract::Request, middleware::Next, response::Response};
-use uuid::Uuid;
-
-use crate::http::headers::X_REQUEST_ID;
-
-/// This middleware function allows compute_ctl to generate its own request ID
-/// if one isn't supplied. The control plane will always send one as a UUID. The
-/// neon Postgres extension on the other hand does not send one.
-pub async fn maybe_add_request_id_header(mut request: Request, next: Next) -> Response {
-    let headers = request.headers_mut();
-    if !headers.contains_key(X_REQUEST_ID) {
-        headers.append(X_REQUEST_ID, Uuid::new_v4().to_string().parse().unwrap());
-    }
-
-    next.run(request).await
-}

compute_tools/src/http/mod.rs

@@ -7,8 +7,6 @@ use serde::Serialize;
 use tracing::error;
 
 mod extract;
-mod headers;
-mod middleware;
 mod routes;
 pub mod server;
 

compute_tools/src/http/routes/configure.rs

@@ -22,7 +22,7 @@ pub(in crate::http) async fn configure(
     State(compute): State<Arc<ComputeNode>>,
     request: Json<ConfigurationRequest>,
 ) -> Response {
-    if !compute.params.live_config_allowed {
+    if !compute.live_config_allowed {
         return JsonResponse::error(
             StatusCode::PRECONDITION_FAILED,
             "live configuration is not allowed for this compute node".to_string(),

compute_tools/src/http/routes/extension_server.rs

@@ -18,11 +18,11 @@ pub(in crate::http) struct ExtensionServerParams {
 /// Download a remote extension.
 pub(in crate::http) async fn download_extension(
     Path(filename): Path<String>,
-    ext_server_params: Query<ExtensionServerParams>,
+    params: Query<ExtensionServerParams>,
     State(compute): State<Arc<ComputeNode>>,
 ) -> Response {
     // Don't even try to download extensions if no remote storage is configured
-    if compute.params.ext_remote_storage.is_none() {
+    if compute.ext_remote_storage.is_none() {
         return JsonResponse::error(
             StatusCode::PRECONDITION_FAILED,
             "remote storage is not configured",
@@ -46,9 +46,9 @@ pub(in crate::http) async fn download_extension(
 
         remote_extensions.get_ext(
             &filename,
-            ext_server_params.is_library,
-            &compute.params.build_tag,
-            &compute.params.pgversion,
+            params.is_library,
+            &compute.build_tag,
+            &compute.pgversion,
         )
     };
 

compute_tools/src/http/server.rs

@@ -5,62 +5,53 @@ use std::time::Duration;
 
 use anyhow::Result;
 use axum::Router;
-use axum::middleware::{self};
-use axum::response::IntoResponse;
+use axum::extract::Request;
+use axum::middleware::{self, Next};
+use axum::response::{IntoResponse, Response};
 use axum::routing::{get, post};
 use http::StatusCode;
-use jsonwebtoken::jwk::JwkSet;
 use tokio::net::TcpListener;
 use tower::ServiceBuilder;
-use tower_http::{
-    auth::AsyncRequireAuthorizationLayer, request_id::PropagateRequestIdLayer, trace::TraceLayer,
-};
-use tracing::{Span, error, info};
+use tower_http::request_id::PropagateRequestIdLayer;
+use tower_http::trace::TraceLayer;
+use tracing::{Span, debug, error, info};
+use uuid::Uuid;
 
-use super::middleware::request_id::maybe_add_request_id_header;
-use super::{
-    headers::X_REQUEST_ID,
-    middleware::authorize::Authorize,
-    routes::{
-        check_writability, configure, database_schema, dbs_and_roles, extension_server, extensions,
-        grants, insights, metrics, metrics_json, status, terminate,
-    },
+use super::routes::{
+    check_writability, configure, database_schema, dbs_and_roles, extension_server, extensions,
+    grants, insights, metrics, metrics_json, status, terminate,
 };
 use crate::compute::ComputeNode;
 
+const X_REQUEST_ID: &str = "x-request-id";
+
 /// `compute_ctl` has two servers: internal and external. The internal server
 /// binds to the loopback interface and handles communication from clients on
 /// the compute. The external server is what receives communication from the
 /// control plane, the metrics scraper, etc. We make the distinction because
 /// certain routes in `compute_ctl` only need to be exposed to local processes
 /// like Postgres via the neon extension and local_proxy.
-#[derive(Clone, Debug)]
+#[derive(Clone, Copy, Debug)]
 pub enum Server {
-    Internal {
-        port: u16,
-    },
-    External {
-        port: u16,
-        jwks: JwkSet,
-        compute_id: String,
-    },
+    Internal(u16),
+    External(u16),
 }
 
 impl Display for Server {
     fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
         match self {
-            Server::Internal { .. } => f.write_str("internal"),
-            Server::External { .. } => f.write_str("external"),
+            Server::Internal(_) => f.write_str("internal"),
+            Server::External(_) => f.write_str("external"),
         }
     }
 }
 
-impl From<&Server> for Router<Arc<ComputeNode>> {
-    fn from(server: &Server) -> Self {
+impl From<Server> for Router<Arc<ComputeNode>> {
+    fn from(server: Server) -> Self {
         let mut router = Router::<Arc<ComputeNode>>::new();
 
         router = match server {
-            Server::Internal { .. } => {
+            Server::Internal(_) => {
                 router = router
                     .route(
                         "/extension_server/{*filename}",
@@ -78,71 +69,59 @@ impl From<&Server> for Router<Arc<ComputeNode>> {
 
                 router
             }
-            Server::External {
-                jwks, compute_id, ..
-            } => {
-                let unauthenticated_router =
-                    Router::<Arc<ComputeNode>>::new().route("/metrics", get(metrics::get_metrics));
-
-                let authenticated_router = Router::<Arc<ComputeNode>>::new()
-                    .route("/check_writability", post(check_writability::is_writable))
-                    .route("/configure", post(configure::configure))
-                    .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))
-                    .route("/metrics.json", get(metrics_json::get_metrics))
-                    .route("/status", get(status::get_status))
-                    .route("/terminate", post(terminate::terminate))
-                    .layer(AsyncRequireAuthorizationLayer::new(Authorize::new(
-                        compute_id.clone(),
-                        jwks.clone(),
-                    )));
-
-                router
-                    .merge(unauthenticated_router)
-                    .merge(authenticated_router)
-            }
+            Server::External(_) => router
+                .route("/check_writability", post(check_writability::is_writable))
+                .route("/configure", post(configure::configure))
+                .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))
+                .route("/metrics", get(metrics::get_metrics))
+                .route("/metrics.json", get(metrics_json::get_metrics))
+                .route("/status", get(status::get_status))
+                .route("/terminate", post(terminate::terminate)),
         };
 
-        router
-            .fallback(Server::handle_404)
-            .method_not_allowed_fallback(Server::handle_405)
-            .layer(
-                ServiceBuilder::new()
-                    .layer(tower_otel::trace::HttpLayer::server(tracing::Level::INFO))
-                    // Add this middleware since we assume the request ID exists
-                    .layer(middleware::from_fn(maybe_add_request_id_header))
-                    .layer(
-                        TraceLayer::new_for_http()
-                            .on_request(|request: &http::Request<_>, _span: &Span| {
-                                let request_id = request
+        router.fallback(Server::handle_404).method_not_allowed_fallback(Server::handle_405).layer(
+            ServiceBuilder::new()
+                // Add this middleware since we assume the request ID exists
+                .layer(middleware::from_fn(maybe_add_request_id_header))
+                .layer(
+                    TraceLayer::new_for_http()
+                        .on_request(|request: &http::Request<_>, _span: &Span| {
+                            let request_id = request
+                                .headers()
+                                .get(X_REQUEST_ID)
+                                .unwrap()
+                                .to_str()
+                                .unwrap();
+
+                            match request.uri().path() {
+                                "/metrics" => {
+                                    debug!(%request_id, "{} {}", request.method(), request.uri())
+                                }
+                                _ => info!(%request_id, "{} {}", request.method(), request.uri()),
+                            };
+                        })
+                        .on_response(
+                            |response: &http::Response<_>, latency: Duration, _span: &Span| {
+                                let request_id = response
                                     .headers()
                                     .get(X_REQUEST_ID)
                                     .unwrap()
                                     .to_str()
                                     .unwrap();
 
-                                info!(%request_id, "{} {}", request.method(), request.uri());
-                            })
-                            .on_response(
-                                |response: &http::Response<_>, latency: Duration, _span: &Span| {
-                                    let request_id = response
-                                        .headers()
-                                        .get(X_REQUEST_ID)
-                                        .unwrap()
-                                        .to_str()
-                                        .unwrap();
-
-                                    info!(
-                                        %request_id,
-                                        code = response.status().as_u16(),
-                                        latency = latency.as_millis()
-                                    );
-                                },
-                            ),
-                    )
-                    .layer(PropagateRequestIdLayer::x_request_id()),
-            )
+                                info!(
+                                    %request_id,
+                                    code = response.status().as_u16(),
+                                    latency = latency.as_millis()
+                                )
+                            },
+                        ),
+                )
+                .layer(PropagateRequestIdLayer::x_request_id()),
+        )
+            .layer(tower_otel::trace::HttpLayer::server(tracing::Level::INFO))
     }
 }
 
@@ -166,15 +145,15 @@ impl Server {
         match self {
             // TODO: Change this to Ipv6Addr::LOCALHOST when the GitHub runners
             // allow binding to localhost
-            Server::Internal { .. } => IpAddr::from(Ipv6Addr::UNSPECIFIED),
-            Server::External { .. } => IpAddr::from(Ipv6Addr::UNSPECIFIED),
+            Server::Internal(_) => IpAddr::from(Ipv6Addr::UNSPECIFIED),
+            Server::External(_) => IpAddr::from(Ipv6Addr::UNSPECIFIED),
         }
     }
 
-    fn port(&self) -> u16 {
+    fn port(self) -> u16 {
         match self {
-            Server::Internal { port, .. } => *port,
-            Server::External { port, .. } => *port,
+            Server::Internal(port) => port,
+            Server::External(port) => port,
         }
     }
 
@@ -201,9 +180,7 @@ impl Server {
             );
         }
 
-        let router = Router::from(&self)
-            .with_state(compute)
-            .into_make_service_with_connect_info::<SocketAddr>();
+        let router = Router::from(self).with_state(compute);
 
         if let Err(e) = axum::serve(listener, router).await {
             error!("compute_ctl {} HTTP server error: {}", self, e);
@@ -218,3 +195,15 @@ impl Server {
         tokio::spawn(self.serve(state));
     }
 }
+
+/// This middleware function allows compute_ctl to generate its own request ID
+/// if one isn't supplied. The control plane will always send one as a UUID. The
+/// neon Postgres extension on the other hand does not send one.
+async fn maybe_add_request_id_header(mut request: Request, next: Next) -> Response {
+    let headers = request.headers_mut();
+    if headers.get(X_REQUEST_ID).is_none() {
+        headers.append(X_REQUEST_ID, Uuid::new_v4().to_string().parse().unwrap());
+    }
+
+    next.run(request).await
+}

compute_tools/src/installed_extensions.rs

@@ -2,7 +2,7 @@ use std::collections::HashMap;
 
 use anyhow::Result;
 use compute_api::responses::{InstalledExtension, InstalledExtensions};
-use tokio_postgres::{Client, Config, NoTls};
+use postgres::{Client, NoTls};
 
 use crate::metrics::INSTALLED_EXTENSIONS;
 
@@ -10,7 +10,7 @@ use crate::metrics::INSTALLED_EXTENSIONS;
 /// and to make database listing query here more explicit.
 ///
 /// Limit the number of databases to 500 to avoid excessive load.
-async fn list_dbs(client: &mut Client) -> Result<Vec<String>> {
+fn list_dbs(client: &mut Client) -> Result<Vec<String>> {
     // `pg_database.datconnlimit = -2` means that the database is in the
     // invalid state
     let databases = client
@@ -20,8 +20,7 @@ async fn list_dbs(client: &mut Client) -> Result<Vec<String>> {
                 AND datconnlimit <> - 2
                 LIMIT 500",
             &[],
-        )
-        .await?
+        )?
         .iter()
         .map(|row| {
             let db: String = row.get("datname");
@@ -37,36 +36,20 @@ async fn list_dbs(client: &mut Client) -> Result<Vec<String>> {
 /// Same extension can be installed in multiple databases with different versions,
 /// so we report a separate metric (number of databases where it is installed)
 /// for each extension version.
-pub async fn get_installed_extensions(mut conf: Config) -> Result<InstalledExtensions> {
+pub fn get_installed_extensions(mut conf: postgres::config::Config) -> Result<InstalledExtensions> {
     conf.application_name("compute_ctl:get_installed_extensions");
-    let databases: Vec<String> = {
-        let (mut client, connection) = conf.connect(NoTls).await?;
-        tokio::spawn(async move {
-            if let Err(e) = connection.await {
-                eprintln!("connection error: {}", e);
-            }
-        });
-
-        list_dbs(&mut client).await?
-    };
+    let mut client = conf.connect(NoTls)?;
+    let databases: Vec<String> = list_dbs(&mut client)?;
 
     let mut extensions_map: HashMap<(String, String, String), InstalledExtension> = HashMap::new();
     for db in databases.iter() {
         conf.dbname(db);
-
-        let (client, connection) = conf.connect(NoTls).await?;
-        tokio::spawn(async move {
-            if let Err(e) = connection.await {
-                eprintln!("connection error: {}", e);
-            }
-        });
-
-        let extensions: Vec<(String, String, i32)> = client
+        let mut db_client = conf.connect(NoTls)?;
+        let extensions: Vec<(String, String, i32)> = db_client
             .query(
                 "SELECT extname, extversion, extowner::integer FROM pg_catalog.pg_extension",
                 &[],
-            )
-            .await?
+            )?
             .iter()
             .map(|row| {
                 (

compute_tools/src/lib.rs

@@ -21,7 +21,6 @@ mod migration;
 pub mod monitor;
 pub mod params;
 pub mod pg_helpers;
-pub mod rsyslog;
 pub mod spec;
 mod spec_apply;
 pub mod swap;

compute_tools/src/logger.rs

@@ -1,5 +1,3 @@
-use std::collections::HashMap;
-use tracing::info;
 use tracing_subscriber::layer::SubscriberExt;
 use tracing_subscriber::prelude::*;
 
@@ -24,8 +22,7 @@ pub async fn init_tracing_and_logging(default_log_level: &str) -> anyhow::Result
         .with_writer(std::io::stderr);
 
     // Initialize OpenTelemetry
-    let otlp_layer =
-        tracing_utils::init_tracing("compute_ctl", tracing_utils::ExportConfig::default()).await;
+    let otlp_layer = tracing_utils::init_tracing("compute_ctl").await;
 
     // Put it all together
     tracing_subscriber::registry()
@@ -45,50 +42,3 @@ pub async fn init_tracing_and_logging(default_log_level: &str) -> anyhow::Result
 pub fn inlinify(s: &str) -> String {
     s.replace('\n', "\u{200B}")
 }
-
-pub fn startup_context_from_env() -> Option<opentelemetry::Context> {
-    // Extract OpenTelemetry context for the startup actions from the
-    // TRACEPARENT and TRACESTATE env variables, and attach it to the current
-    // tracing context.
-    //
-    // This is used to propagate the context for the 'start_compute' operation
-    // from the neon control plane. This allows linking together the wider
-    // 'start_compute' operation that creates the compute container, with the
-    // startup actions here within the container.
-    //
-    // There is no standard for passing context in env variables, but a lot of
-    // tools use TRACEPARENT/TRACESTATE, so we use that convention too. See
-    // https://github.com/open-telemetry/opentelemetry-specification/issues/740
-    //
-    // Switch to the startup context here, and exit it once the startup has
-    // completed and Postgres is up and running.
-    //
-    // If this pod is pre-created without binding it to any particular endpoint
-    // yet, this isn't the right place to enter the startup context. In that
-    // case, the control plane should pass the tracing context as part of the
-    // /configure API call.
-    //
-    // NOTE: This is supposed to only cover the *startup* actions. Once
-    // postgres is configured and up-and-running, we exit this span. Any other
-    // actions that are performed on incoming HTTP requests, for example, are
-    // performed in separate spans.
-    //
-    // XXX: If the pod is restarted, we perform the startup actions in the same
-    // context as the original startup actions, which probably doesn't make
-    // sense.
-    let mut startup_tracing_carrier: HashMap<String, String> = HashMap::new();
-    if let Ok(val) = std::env::var("TRACEPARENT") {
-        startup_tracing_carrier.insert("traceparent".to_string(), val);
-    }
-    if let Ok(val) = std::env::var("TRACESTATE") {
-        startup_tracing_carrier.insert("tracestate".to_string(), val);
-    }
-    if !startup_tracing_carrier.is_empty() {
-        use opentelemetry::propagation::TextMapPropagator;
-        use opentelemetry_sdk::propagation::TraceContextPropagator;
-        info!("got startup tracing context from env variables");
-        Some(TraceContextPropagator::new().extract(&startup_tracing_carrier))
-    } else {
-        None
-    }
-}

compute_tools/src/metrics.rs

@@ -54,7 +54,9 @@ pub(crate) static REMOTE_EXT_REQUESTS_TOTAL: Lazy<IntCounterVec> = Lazy::new(||
     register_int_counter_vec!(
         "compute_ctl_remote_ext_requests_total",
         "Total number of requests made by compute_ctl to download extensions from S3 proxy by status",
-        &["http_status", "filename"]
+        // Do not use any labels like extension name yet.
+        // We can add them later if needed.
+        &["http_status"]
     )
     .expect("failed to define a metric")
 });

compute_tools/src/monitor.rs

@@ -18,7 +18,7 @@ const MONITOR_CHECK_INTERVAL: Duration = Duration::from_millis(500);
 // should be handled gracefully.
 fn watch_compute_activity(compute: &ComputeNode) {
     // Suppose that `connstr` doesn't change
-    let connstr = compute.params.connstr.clone();
+    let connstr = compute.connstr.clone();
     let conf = compute.get_conn_conf(Some("compute_ctl:activity_monitor"));
 
     // During startup and configuration we connect to every Postgres database,

compute_tools/src/pg_helpers.rs

@@ -186,40 +186,15 @@ impl DatabaseExt for Database {
 /// Postgres SQL queries and DATABASE_URL.
 pub trait Escaping {
     fn pg_quote(&self) -> String;
-    fn pg_quote_dollar(&self) -> (String, String);
 }
 
 impl Escaping for PgIdent {
     /// This is intended to mimic Postgres quote_ident(), but for simplicity it
     /// always quotes provided string with `""` and escapes every `"`.
     /// **Not idempotent**, i.e. if string is already escaped it will be escaped again.
-    /// N.B. it's not useful for escaping identifiers that are used inside WHERE
-    /// clause, use `escape_literal()` instead.
     fn pg_quote(&self) -> String {
-        format!("\"{}\"", self.replace('"', "\"\""))
-    }
-
-    /// This helper is intended to be used for dollar-escaping strings for usage
-    /// inside PL/pgSQL procedures. In addition to dollar-escaping the string,
-    /// it also returns a tag that is intended to be used inside the outer
-    /// PL/pgSQL procedure. If you do not need an outer tag, just discard it.
-    /// Here we somewhat mimic the logic of Postgres' `pg_get_functiondef()`,
-    /// <https://github.com/postgres/postgres/blob/8b49392b270b4ac0b9f5c210e2a503546841e832/src/backend/utils/adt/ruleutils.c#L2924>
-    fn pg_quote_dollar(&self) -> (String, String) {
-        let mut tag: String = "".to_string();
-        let mut outer_tag = "x".to_string();
-
-        // Find the first suitable tag that is not present in the string.
-        // Postgres' max role/DB name length is 63 bytes, so even in the
-        // worst case it won't take long.
-        while self.contains(&format!("${tag}$")) || self.contains(&format!("${outer_tag}$")) {
-            tag += "x";
-            outer_tag = tag.clone() + "x";
-        }
-
-        let escaped = format!("${tag}${self}${tag}$");
-
-        (escaped, outer_tag)
+        let result = format!("\"{}\"", self.replace('"', "\"\""));
+        result
     }
 }
 
@@ -251,13 +226,10 @@ pub async fn get_existing_dbs_async(
     // invalid state. See:
     //   https://github.com/postgres/postgres/commit/a4b4cc1d60f7e8ccfcc8ff8cb80c28ee411ad9a9
     let rowstream = client
-        // We use a subquery instead of a fancy `datdba::regrole::text AS owner`,
-        // because the latter automatically wraps the result in double quotes,
-        // if the role name contains special characters.
         .query_raw::<str, &String, &[String; 0]>(
             "SELECT
                 datname AS name,
-                (SELECT rolname FROM pg_roles WHERE oid = datdba) AS owner,
+                datdba::regrole::text AS owner,
                 NOT datallowconn AS restrict_conn,
                 datconnlimit = - 2 AS invalid
             FROM

compute_tools/src/rsyslog.rs

@@ -1,77 +0,0 @@
-use std::process::Command;
-use std::{fs::OpenOptions, io::Write};
-
-use anyhow::{Context, Result};
-use tracing::info;
-
-fn get_rsyslog_pid() -> Option<String> {
-    let output = Command::new("pgrep")
-        .arg("rsyslogd")
-        .output()
-        .expect("Failed to execute pgrep");
-
-    if !output.stdout.is_empty() {
-        let pid = std::str::from_utf8(&output.stdout)
-            .expect("Invalid UTF-8 in process output")
-            .trim()
-            .to_string();
-        Some(pid)
-    } else {
-        None
-    }
-}
-
-// Restart rsyslogd to apply the new configuration.
-// This is necessary, because there is no other way to reload the rsyslog configuration.
-//
-// Rsyslogd shouldn't lose any messages, because of the restart,
-// because it tracks the last read position in the log files
-// and will continue reading from that position.
-// TODO: test it properly
-//
-fn restart_rsyslog() -> Result<()> {
-    let old_pid = get_rsyslog_pid().context("rsyslogd is not running")?;
-    info!("rsyslogd is running with pid: {}, restart it", old_pid);
-
-    // kill it to restart
-    let _ = Command::new("pkill")
-        .arg("rsyslogd")
-        .output()
-        .context("Failed to stop rsyslogd")?;
-
-    Ok(())
-}
-
-pub fn configure_audit_rsyslog(
-    log_directory: &str,
-    tag: &str,
-    remote_endpoint: &str,
-) -> Result<()> {
-    let config_content: String = format!(
-        include_str!("config_template/compute_audit_rsyslog_template.conf"),
-        log_directory = log_directory,
-        tag = tag,
-        remote_endpoint = remote_endpoint
-    );
-
-    info!("rsyslog config_content: {}", config_content);
-
-    let rsyslog_conf_path = "/etc/rsyslog.d/compute_audit_rsyslog.conf";
-    let mut file = OpenOptions::new()
-        .create(true)
-        .write(true)
-        .truncate(true)
-        .open(rsyslog_conf_path)?;
-
-    file.write_all(config_content.as_bytes())?;
-
-    info!(
-        "rsyslog configuration file {} added successfully. Starting rsyslogd",
-        rsyslog_conf_path
-    );
-
-    // start the service, using the configuration
-    restart_rsyslog()?;
-
-    Ok(())
-}

compute_tools/src/spec_apply.rs

@@ -4,427 +4,15 @@ use std::future::Future;
 use std::iter::{empty, once};
 use std::sync::Arc;
 
-use anyhow::{Context, Result};
-use compute_api::responses::ComputeStatus;
-use compute_api::spec::{ComputeAudit, ComputeFeature, ComputeSpec, Database, PgIdent, Role};
+use anyhow::Result;
+use compute_api::spec::{ComputeFeature, ComputeSpec, Database, PgIdent, Role};
 use futures::future::join_all;
 use tokio::sync::RwLock;
 use tokio_postgres::Client;
-use tokio_postgres::error::SqlState;
-use tracing::{Instrument, debug, error, info, info_span, instrument, warn};
+use tracing::{Instrument, debug, info_span, warn};
 
-use crate::compute::{ComputeNode, ComputeState};
-use crate::pg_helpers::{
-    DatabaseExt, Escaping, GenericOptionsSearch, RoleExt, get_existing_dbs_async,
-    get_existing_roles_async,
-};
-use crate::spec_apply::ApplySpecPhase::{
-    CreateAndAlterDatabases, CreateAndAlterRoles, CreateAvailabilityCheck, CreateNeonSuperuser,
-    CreatePgauditExtension, CreatePgauditlogtofileExtension, CreateSchemaNeon,
-    DisablePostgresDBPgAudit, DropInvalidDatabases, DropRoles, FinalizeDropLogicalSubscriptions,
-    HandleNeonExtension, HandleOtherExtensions, RenameAndDeleteDatabases, RenameRoles,
-    RunInEachDatabase,
-};
-use crate::spec_apply::PerDatabasePhase::{
-    ChangeSchemaPerms, DeleteDBRoleReferences, DropLogicalSubscriptions, HandleAnonExtension,
-};
-
-impl ComputeNode {
-    /// Apply the spec to the running PostgreSQL instance.
-    /// The caller can decide to run with multiple clients in parallel, or
-    /// single mode.  Either way, the commands executed will be the same, and
-    /// only commands run in different databases are parallelized.
-    #[instrument(skip_all)]
-    pub fn apply_spec_sql(
-        &self,
-        spec: Arc<ComputeSpec>,
-        conf: Arc<tokio_postgres::Config>,
-        concurrency: usize,
-    ) -> Result<()> {
-        info!("Applying config with max {} concurrency", concurrency);
-        debug!("Config: {:?}", spec);
-
-        let rt = tokio::runtime::Handle::current();
-        rt.block_on(async {
-            // Proceed with post-startup configuration. Note, that order of operations is important.
-            let client = Self::get_maintenance_client(&conf).await?;
-            let spec = spec.clone();
-
-            let databases = get_existing_dbs_async(&client).await?;
-            let roles = get_existing_roles_async(&client)
-                .await?
-                .into_iter()
-                .map(|role| (role.name.clone(), role))
-                .collect::<HashMap<String, Role>>();
-
-            // Check if we need to drop subscriptions before starting the endpoint.
-            //
-            // It is important to do this operation exactly once when endpoint starts on a new branch.
-            // Otherwise, we may drop not inherited, but newly created subscriptions.
-            //
-            // We cannot rely only on spec.drop_subscriptions_before_start flag,
-            // because if for some reason compute restarts inside VM,
-            // it will start again with the same spec and flag value.
-            //
-            // To handle this, we save the fact of the operation in the database
-            // in the neon.drop_subscriptions_done table.
-            // If the table does not exist, we assume that the operation was never performed, so we must do it.
-            // If table exists, we check if the operation was performed on the current timelilne.
-            //
-            let mut drop_subscriptions_done = false;
-
-            if spec.drop_subscriptions_before_start {
-                let timeline_id = self.get_timeline_id().context("timeline_id must be set")?;
-                let query = format!("select 1 from neon.drop_subscriptions_done where timeline_id = '{}'", timeline_id);
-
-                info!("Checking if drop subscription operation was already performed for timeline_id: {}", timeline_id);
-
-                drop_subscriptions_done =  match
-                    client.simple_query(&query).await {
-                    Ok(result) => {
-                        matches!(&result[0], postgres::SimpleQueryMessage::Row(_))
-                    },
-                    Err(e) =>
-                    {
-                        match e.code() {
-                            Some(&SqlState::UNDEFINED_TABLE) => false,
-                            _ => {
-                                // We don't expect any other error here, except for the schema/table not existing
-                                error!("Error checking if drop subscription operation was already performed: {}", e);
-                                return Err(e.into());
-                            }
-                        }
-                    }
-                }
-            };
-
-
-            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<_>>(),
-            );
-
-            let ctx = Arc::new(tokio::sync::RwLock::new(MutableApplyContext {
-                roles,
-                dbs: databases,
-            }));
-
-            // Apply special pre drop database phase.
-            // NOTE: we use the code of RunInEachDatabase phase for parallelism
-            // and connection management, but we don't really run it in *each* database,
-            // only in databases, we're about to drop.
-            info!("Applying PerDatabase (pre-dropdb) phase");
-            let concurrency_token = Arc::new(tokio::sync::Semaphore::new(concurrency));
-
-            // Run the phase for each database that we're about to drop.
-            let db_processes = spec
-                .delta_operations
-                .iter()
-                .flatten()
-                .filter_map(move |op| {
-                    if op.action.as_str() == "delete_db" {
-                        Some(op.name.clone())
-                    } else {
-                        None
-                    }
-                })
-                .map(|dbname| {
-                    let spec = spec.clone();
-                    let ctx = ctx.clone();
-                    let jwks_roles = jwks_roles.clone();
-                    let mut conf = conf.as_ref().clone();
-                    let concurrency_token = concurrency_token.clone();
-                    // We only need dbname field for this phase, so set other fields to dummy values
-                    let db = DB::UserDB(Database {
-                        name: dbname.clone(),
-                        owner: "cloud_admin".to_string(),
-                        options: None,
-                        restrict_conn: false,
-                        invalid: false,
-                    });
-
-                    debug!("Applying per-database phases for Database {:?}", &db);
-
-                    match &db {
-                        DB::SystemDB => {}
-                        DB::UserDB(db) => {
-                            conf.dbname(db.name.as_str());
-                        }
-                    }
-
-                    let conf = Arc::new(conf);
-                    let fut = Self::apply_spec_sql_db(
-                        spec.clone(),
-                        conf,
-                        ctx.clone(),
-                        jwks_roles.clone(),
-                        concurrency_token.clone(),
-                        db,
-                        [DropLogicalSubscriptions].to_vec(),
-                    );
-
-                    Ok(tokio::spawn(fut))
-                })
-                .collect::<Vec<Result<_, anyhow::Error>>>();
-
-            for process in db_processes.into_iter() {
-                let handle = process?;
-                if let Err(e) = handle.await? {
-                    // Handle the error case where the database does not exist
-                    // We do not check whether the DB exists or not in the deletion phase,
-                    // so we shouldn't be strict about it in pre-deletion cleanup as well.
-                    if e.to_string().contains("does not exist") {
-                        warn!("Error dropping subscription: {}", e);
-                    } else {
-                        return Err(e);
-                    }
-                };
-            }
-
-            for phase in [
-                CreateNeonSuperuser,
-                DropInvalidDatabases,
-                RenameRoles,
-                CreateAndAlterRoles,
-                RenameAndDeleteDatabases,
-                CreateAndAlterDatabases,
-                CreateSchemaNeon,
-            ] {
-                info!("Applying phase {:?}", &phase);
-                apply_operations(
-                    spec.clone(),
-                    ctx.clone(),
-                    jwks_roles.clone(),
-                    phase,
-                    || async { Ok(&client) },
-                )
-                .await?;
-            }
-
-            info!("Applying RunInEachDatabase2 phase");
-            let concurrency_token = Arc::new(tokio::sync::Semaphore::new(concurrency));
-
-            let db_processes = spec
-                .cluster
-                .databases
-                .iter()
-                .map(|db| DB::new(db.clone()))
-                // include
-                .chain(once(DB::SystemDB))
-                .map(|db| {
-                    let spec = spec.clone();
-                    let ctx = ctx.clone();
-                    let jwks_roles = jwks_roles.clone();
-                    let mut conf = conf.as_ref().clone();
-                    let concurrency_token = concurrency_token.clone();
-                    let db = db.clone();
-
-                    debug!("Applying per-database phases for Database {:?}", &db);
-
-                    match &db {
-                        DB::SystemDB => {}
-                        DB::UserDB(db) => {
-                            conf.dbname(db.name.as_str());
-                        }
-                    }
-
-                    let conf = Arc::new(conf);
-                    let mut phases = vec![
-                        DeleteDBRoleReferences,
-                        ChangeSchemaPerms,
-                        HandleAnonExtension,
-                    ];
-
-                    if spec.drop_subscriptions_before_start && !drop_subscriptions_done {
-                        info!("Adding DropLogicalSubscriptions phase because drop_subscriptions_before_start is set");
-                        phases.push(DropLogicalSubscriptions);
-                    }
-
-                    let fut = Self::apply_spec_sql_db(
-                        spec.clone(),
-                        conf,
-                        ctx.clone(),
-                        jwks_roles.clone(),
-                        concurrency_token.clone(),
-                        db,
-                        phases,
-                    );
-
-                    Ok(tokio::spawn(fut))
-                })
-                .collect::<Vec<Result<_, anyhow::Error>>>();
-
-            for process in db_processes.into_iter() {
-                let handle = process?;
-                handle.await??;
-            }
-
-            let mut phases = vec![
-                HandleOtherExtensions,
-                HandleNeonExtension, // This step depends on CreateSchemaNeon
-                CreateAvailabilityCheck,
-                DropRoles,
-            ];
-
-            // This step depends on CreateSchemaNeon
-            if spec.drop_subscriptions_before_start && !drop_subscriptions_done {
-                info!("Adding FinalizeDropLogicalSubscriptions phase because drop_subscriptions_before_start is set");
-                phases.push(FinalizeDropLogicalSubscriptions);
-            }
-
-            // Keep DisablePostgresDBPgAudit phase at the end,
-            // so that all config operations are audit logged.
-            match spec.audit_log_level
-            {
-                ComputeAudit::Hipaa => {
-                    phases.push(CreatePgauditExtension);
-                    phases.push(CreatePgauditlogtofileExtension);
-                    phases.push(DisablePostgresDBPgAudit);
-                }
-                ComputeAudit::Log => { /* not implemented yet */ }
-                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(())
-    }
-
-    /// Apply SQL migrations of the RunInEachDatabase phase.
-    ///
-    /// May opt to not connect to databases that don't have any scheduled
-    /// operations.  The function is concurrency-controlled with the provided
-    /// semaphore.  The caller has to make sure the semaphore isn't exhausted.
-    async fn apply_spec_sql_db(
-        spec: Arc<ComputeSpec>,
-        conf: Arc<tokio_postgres::Config>,
-        ctx: Arc<tokio::sync::RwLock<MutableApplyContext>>,
-        jwks_roles: Arc<HashSet<String>>,
-        concurrency_token: Arc<tokio::sync::Semaphore>,
-        db: DB,
-        subphases: Vec<PerDatabasePhase>,
-    ) -> Result<()> {
-        let _permit = concurrency_token.acquire().await?;
-
-        let mut client_conn = None;
-
-        for subphase in subphases {
-            apply_operations(
-                spec.clone(),
-                ctx.clone(),
-                jwks_roles.clone(),
-                RunInEachDatabase {
-                    db: db.clone(),
-                    subphase,
-                },
-                // Only connect if apply_operation actually wants a connection.
-                // It's quite possible this database doesn't need any queries,
-                // so by not connecting we save time and effort connecting to
-                // that database.
-                || async {
-                    if client_conn.is_none() {
-                        let db_client = Self::get_maintenance_client(&conf).await?;
-                        client_conn.replace(db_client);
-                    }
-                    let client = client_conn.as_ref().unwrap();
-                    Ok(client)
-                },
-            )
-            .await?;
-        }
-
-        drop(client_conn);
-
-        Ok::<(), anyhow::Error>(())
-    }
-
-    /// Choose how many concurrent connections to use for applying the spec changes.
-    pub fn max_service_connections(
-        &self,
-        compute_state: &ComputeState,
-        spec: &ComputeSpec,
-    ) -> usize {
-        // If the cluster is in Init state we don't have to deal with user connections,
-        // and can thus use all `max_connections` connection slots. However, that's generally not
-        // very efficient, so we generally still limit it to a smaller number.
-        if compute_state.status == ComputeStatus::Init {
-            // If the settings contain 'max_connections', use that as template
-            if let Some(config) = spec.cluster.settings.find("max_connections") {
-                config.parse::<usize>().ok()
-            } else {
-                // Otherwise, try to find the setting in the postgresql_conf string
-                spec.cluster
-                    .postgresql_conf
-                    .iter()
-                    .flat_map(|conf| conf.split("\n"))
-                    .filter_map(|line| {
-                        if !line.contains("max_connections") {
-                            return None;
-                        }
-
-                        let (key, value) = line.split_once("=")?;
-                        let key = key
-                            .trim_start_matches(char::is_whitespace)
-                            .trim_end_matches(char::is_whitespace);
-
-                        let value = value
-                            .trim_start_matches(char::is_whitespace)
-                            .trim_end_matches(char::is_whitespace);
-
-                        if key != "max_connections" {
-                            return None;
-                        }
-
-                        value.parse::<usize>().ok()
-                    })
-                    .next()
-            }
-            // If max_connections is present, use at most 1/3rd of that.
-            // When max_connections is lower than 30, try to use at least 10 connections, but
-            // never more than max_connections.
-            .map(|limit| match limit {
-                0..10 => limit,
-                10..30 => 10,
-                30.. => limit / 3,
-            })
-            // If we didn't find max_connections, default to 10 concurrent connections.
-            .unwrap_or(10)
-        } else {
-            // state == Running
-            // Because the cluster is already in the Running state, we should assume users are
-            // already connected to the cluster, and high concurrency could negatively
-            // impact user connectivity. Therefore, we can limit concurrency to the number of
-            // reserved superuser connections, which users wouldn't be able to use anyway.
-            spec.cluster
-                .settings
-                .find("superuser_reserved_connections")
-                .iter()
-                .filter_map(|val| val.parse::<usize>().ok())
-                .map(|val| if val > 1 { val - 1 } else { 1 })
-                .last()
-                .unwrap_or(3)
-        }
-    }
-}
+use crate::compute::construct_superuser_query;
+use crate::pg_helpers::{DatabaseExt, Escaping, GenericOptionsSearch, RoleExt, escape_literal};
 
 #[derive(Clone)]
 pub enum DB {
@@ -469,7 +57,7 @@ pub enum PerDatabasePhase {
 
 #[derive(Clone, Debug)]
 pub enum ApplySpecPhase {
-    CreateNeonSuperuser,
+    CreateSuperUser,
     DropInvalidDatabases,
     RenameRoles,
     CreateAndAlterRoles,
@@ -477,9 +65,6 @@ pub enum ApplySpecPhase {
     CreateAndAlterDatabases,
     CreateSchemaNeon,
     RunInEachDatabase { db: DB, subphase: PerDatabasePhase },
-    CreatePgauditExtension,
-    CreatePgauditlogtofileExtension,
-    DisablePostgresDBPgAudit,
     HandleOtherExtensions,
     HandleNeonExtension,
     CreateAvailabilityCheck,
@@ -596,10 +181,14 @@ async fn get_operations<'a>(
     apply_spec_phase: &'a ApplySpecPhase,
 ) -> Result<Box<dyn Iterator<Item = Operation> + 'a + Send>> {
     match apply_spec_phase {
-        ApplySpecPhase::CreateNeonSuperuser => Ok(Box::new(once(Operation {
-            query: include_str!("sql/create_neon_superuser.sql").to_string(),
-            comment: None,
-        }))),
+        ApplySpecPhase::CreateSuperUser => {
+            let query = construct_superuser_query(spec);
+
+            Ok(Box::new(once(Operation {
+                query,
+                comment: None,
+            })))
+        }
         ApplySpecPhase::DropInvalidDatabases => {
             let mut ctx = ctx.write().await;
             let databases = &mut ctx.dbs;
@@ -733,15 +322,14 @@ async fn get_operations<'a>(
                         // We do not check whether the DB exists or not,
                         // Postgres will take care of it for us
                         "delete_db" => {
-                            let (db_name, outer_tag) = op.name.pg_quote_dollar();
                             // In Postgres we can't drop a database if it is a template.
                             // So we need to unset the template flag first, but it could
                             // be a retry, so we could've already dropped the database.
                             // Check that database exists first to make it idempotent.
                             let unset_template_query: String = format!(
                                 include_str!("sql/unset_template_for_drop_dbs.sql"),
-                                datname = db_name,
-                                outer_tag = outer_tag,
+                                datname_str = escape_literal(&op.name),
+                                datname = &op.name.pg_quote()
                             );
 
                             // Use FORCE to drop database even if there are active connections.
@@ -848,8 +436,6 @@ async fn get_operations<'a>(
                                 comment: None,
                             },
                             Operation {
-                                // ALL PRIVILEGES grants CREATE, CONNECT, and TEMPORARY on the database
-                                // (see https://www.postgresql.org/docs/current/ddl-priv.html)
                                 query: format!(
                                     "GRANT ALL PRIVILEGES ON DATABASE {} TO neon_superuser",
                                     db.name.pg_quote()
@@ -909,11 +495,9 @@ async fn get_operations<'a>(
                 PerDatabasePhase::DropLogicalSubscriptions => {
                     match &db {
                         DB::UserDB(db) => {
-                            let (db_name, outer_tag) = db.name.pg_quote_dollar();
                             let drop_subscription_query: String = format!(
                                 include_str!("sql/drop_subscriptions.sql"),
-                                datname_str = db_name,
-                                outer_tag = outer_tag,
+                                datname_str = escape_literal(&db.name),
                             );
 
                             let operations = vec![Operation {
@@ -952,7 +536,6 @@ async fn get_operations<'a>(
                                     DB::SystemDB => PgIdent::from("cloud_admin").pg_quote(),
                                     DB::UserDB(db) => db.owner.pg_quote(),
                                 };
-                                let (escaped_role, outer_tag) = op.name.pg_quote_dollar();
 
                                 Some(vec![
                                     // This will reassign all dependent objects to the db owner
@@ -967,9 +550,7 @@ async fn get_operations<'a>(
                                     Operation {
                                         query: format!(
                                             include_str!("sql/pre_drop_role_revoke_privileges.sql"),
-                                            // N.B. this has to be properly dollar-escaped with `pg_quote_dollar()`
-                                            role_name = escaped_role,
-                                            outer_tag = outer_tag,
+                                            role_name = quoted,
                                         ),
                                         comment: None,
                                     },
@@ -994,14 +575,12 @@ async fn get_operations<'a>(
                         DB::SystemDB => return Ok(Box::new(empty())),
                         DB::UserDB(db) => db,
                     };
-                    let (db_owner, outer_tag) = db.owner.pg_quote_dollar();
 
                     let operations = vec![
                         Operation {
                             query: format!(
                                 include_str!("sql/set_public_schema_owner.sql"),
-                                db_owner = db_owner,
-                                outer_tag = outer_tag,
+                                db_owner = db.owner.pg_quote()
                             ),
                             comment: None,
                         },
@@ -1121,25 +700,6 @@ async fn get_operations<'a>(
             }
             Ok(Box::new(empty()))
         }
-        ApplySpecPhase::CreatePgauditExtension => Ok(Box::new(once(Operation {
-            query: String::from("CREATE EXTENSION IF NOT EXISTS pgaudit"),
-            comment: Some(String::from("create pgaudit extensions")),
-        }))),
-        ApplySpecPhase::CreatePgauditlogtofileExtension => Ok(Box::new(once(Operation {
-            query: String::from("CREATE EXTENSION IF NOT EXISTS pgauditlogtofile"),
-            comment: Some(String::from("create pgauditlogtofile extensions")),
-        }))),
-        // Disable pgaudit logging for postgres database.
-        // Postgres is neon system database used by monitors
-        // and compute_ctl tuning functions and thus generates a lot of noise.
-        // We do not consider data stored in this database as sensitive.
-        ApplySpecPhase::DisablePostgresDBPgAudit => {
-            let query = "ALTER DATABASE postgres SET pgaudit.log to 'none'";
-            Ok(Box::new(once(Operation {
-                query: query.to_string(),
-                comment: Some(query.to_string()),
-            })))
-        }
         ApplySpecPhase::HandleNeonExtension => {
             let operations = vec![
                 Operation {

compute_tools/src/sql/create_neon_superuser.sql

@@ -1,8 +0,0 @@
-DO $$
-    BEGIN
-        IF NOT EXISTS (SELECT FROM pg_catalog.pg_roles WHERE rolname = 'neon_superuser')
-        THEN
-            CREATE ROLE neon_superuser CREATEDB CREATEROLE NOLOGIN REPLICATION BYPASSRLS IN ROLE pg_read_all_data, pg_write_all_data;
-        END IF;
-    END
-$$;

compute_tools/src/sql/drop_subscriptions.sql

@@ -1,4 +1,4 @@
-DO ${outer_tag}$
+DO $$
 DECLARE
     subname TEXT;
 BEGIN
@@ -9,4 +9,4 @@ BEGIN
         EXECUTE format('DROP SUBSCRIPTION %I;', subname);
     END LOOP;
 END;
-${outer_tag}$;
+$$;

compute_tools/src/sql/pre_drop_role_revoke_privileges.sql

@@ -1,7 +1,8 @@
-DO ${outer_tag}$
+SET SESSION ROLE neon_superuser;
+
+DO $$
 DECLARE
     schema TEXT;
-    grantor TEXT;
     revoke_query TEXT;
 BEGIN
     FOR schema IN
@@ -14,25 +15,14 @@ BEGIN
         -- ii) it's easy to add more schemas to the list if needed.
         WHERE schema_name IN ('public')
     LOOP
-        FOR grantor IN EXECUTE
-            format(
-                'SELECT DISTINCT rtg.grantor FROM information_schema.role_table_grants AS rtg WHERE grantee = %s',
-                -- N.B. this has to be properly dollar-escaped with `pg_quote_dollar()`
-                quote_literal({role_name})
-            )
-        LOOP
-            EXECUTE format('SET LOCAL ROLE %I', grantor);
+        revoke_query := format(
+            'REVOKE ALL PRIVILEGES ON ALL TABLES IN SCHEMA %I FROM {role_name} GRANTED BY neon_superuser;',
+            schema
+        );
 
-            revoke_query := format(
-                'REVOKE ALL PRIVILEGES ON ALL TABLES IN SCHEMA %I FROM %I GRANTED BY %I',
-                schema,
-                -- N.B. this has to be properly dollar-escaped with `pg_quote_dollar()`
-                {role_name},
-                grantor
-            );
-
-            EXECUTE revoke_query;
-        END LOOP;
+        EXECUTE revoke_query;
     END LOOP;
 END;
-${outer_tag}$;
+$$;
+
+RESET ROLE;

compute_tools/src/sql/set_public_schema_owner.sql

@@ -1,4 +1,5 @@
-DO ${outer_tag}$
+DO
+$$
     DECLARE
         schema_owner TEXT;
     BEGIN
@@ -15,8 +16,8 @@ DO ${outer_tag}$
 
             IF schema_owner = 'cloud_admin' OR schema_owner = 'zenith_admin'
             THEN
-                EXECUTE format('ALTER SCHEMA public OWNER TO %I', {db_owner});
+                ALTER SCHEMA public OWNER TO {db_owner};
             END IF;
         END IF;
     END
-${outer_tag}$;
+$$;

compute_tools/src/sql/unset_template_for_drop_dbs.sql

@@ -1,12 +1,12 @@
-DO ${outer_tag}$
+DO $$
     BEGIN
         IF EXISTS(
             SELECT 1
             FROM pg_catalog.pg_database
-            WHERE datname = {datname}
+            WHERE datname = {datname_str}
         )
         THEN
-            EXECUTE format('ALTER DATABASE %I is_template false', {datname});
+            ALTER DATABASE {datname} is_template false;
         END IF;
     END
-${outer_tag}$;
+$$;

compute_tools/tests/pg_helpers_tests.rs

@@ -61,23 +61,6 @@ test.escaping = 'here''s a backslash \\ and a quote '' and a double-quote " hoor
         assert_eq!(ident.pg_quote(), "\"\"\"name\"\";\\n select 1;\"");
     }
 
-    #[test]
-    fn ident_pg_quote_dollar() {
-        let test_cases = vec![
-            ("name", ("$$name$$", "x")),
-            ("name$$", ("$x$name$$$x$", "xx")),
-            ("name$$$", ("$x$name$$$$x$", "xx")),
-            ("name$$$$", ("$x$name$$$$$x$", "xx")),
-            ("name$x$", ("$xx$name$x$$xx$", "xxx")),
-        ];
-
-        for (input, expected) in test_cases {
-            let (escaped, tag) = PgIdent::from(input).pg_quote_dollar();
-            assert_eq!(escaped, expected.0);
-            assert_eq!(tag, expected.1);
-        }
-    }
-
     #[test]
     fn generic_options_search() {
         let generic_options: GenericOptions = Some(vec![

control_plane/src/background_process.rs

@@ -25,7 +25,7 @@ use anyhow::Context;
 use camino::{Utf8Path, Utf8PathBuf};
 use nix::errno::Errno;
 use nix::fcntl::{FcntlArg, FdFlag};
-use nix::sys::signal::{Signal, kill};
+use nix::sys::signal::{kill, Signal};
 use nix::unistd::Pid;
 use utils::pid_file::{self, PidFileRead};
 

control_plane/src/bin/neon_local.rs

@@ -5,16 +5,7 @@
 //! easier to work with locally. The python tests in `test_runner`
 //! rely on `neon_local` to set up the environment for each test.
 //!
-use std::borrow::Cow;
-use std::collections::{BTreeSet, HashMap};
-use std::fs::File;
-use std::os::fd::AsRawFd;
-use std::path::PathBuf;
-use std::process::exit;
-use std::str::FromStr;
-use std::time::Duration;
-
-use anyhow::{Context, Result, anyhow, bail};
+use anyhow::{anyhow, bail, Context, Result};
 use clap::Parser;
 use compute_api::spec::ComputeMode;
 use control_plane::endpoint::ComputeControlPlane;
@@ -28,7 +19,7 @@ use control_plane::storage_controller::{
     NeonStorageControllerStartArgs, NeonStorageControllerStopArgs, StorageController,
 };
 use control_plane::{broker, local_env};
-use nix::fcntl::{FlockArg, flock};
+use nix::fcntl::{flock, FlockArg};
 use pageserver_api::config::{
     DEFAULT_HTTP_LISTEN_PORT as DEFAULT_PAGESERVER_HTTP_PORT,
     DEFAULT_PG_LISTEN_PORT as DEFAULT_PAGESERVER_PG_PORT,
@@ -36,24 +27,31 @@ use pageserver_api::config::{
 use pageserver_api::controller_api::{
     NodeAvailabilityWrapper, PlacementPolicy, TenantCreateRequest,
 };
-use pageserver_api::models::{
-    ShardParameters, TenantConfigRequest, TimelineCreateRequest, TimelineInfo,
-};
+use pageserver_api::models::{ShardParameters, TimelineCreateRequest, TimelineInfo};
 use pageserver_api::shard::{ShardCount, ShardStripeSize, TenantShardId};
 use postgres_backend::AuthType;
 use postgres_connection::parse_host_port;
-use safekeeper_api::membership::SafekeeperGeneration;
 use safekeeper_api::{
     DEFAULT_HTTP_LISTEN_PORT as DEFAULT_SAFEKEEPER_HTTP_PORT,
     DEFAULT_PG_LISTEN_PORT as DEFAULT_SAFEKEEPER_PG_PORT,
 };
+use std::borrow::Cow;
+use std::collections::{BTreeSet, HashMap};
+use std::fs::File;
+use std::os::fd::AsRawFd;
+use std::path::PathBuf;
+use std::process::exit;
+use std::str::FromStr;
+use std::time::Duration;
 use storage_broker::DEFAULT_LISTEN_ADDR as DEFAULT_BROKER_ADDR;
 use tokio::task::JoinSet;
 use url::Host;
-use utils::auth::{Claims, Scope};
-use utils::id::{NodeId, TenantId, TenantTimelineId, TimelineId};
-use utils::lsn::Lsn;
-use utils::project_git_version;
+use utils::{
+    auth::{Claims, Scope},
+    id::{NodeId, TenantId, TenantTimelineId, TimelineId},
+    lsn::Lsn,
+    project_git_version,
+};
 
 // Default id of a safekeeper node, if not specified on the command line.
 const DEFAULT_SAFEKEEPER_ID: NodeId = NodeId(1);
@@ -599,15 +597,7 @@ struct EndpointStartCmdArgs {
     #[clap(long = "pageserver-id")]
     endpoint_pageserver_id: Option<NodeId>,
 
-    #[clap(
-        long,
-        help = "Safekeepers membership generation to prefix neon.safekeepers with. Normally neon_local sets it on its own, but this option allows to override. Non zero value forces endpoint to use membership configurations."
-    )]
-    safekeepers_generation: Option<u32>,
-    #[clap(
-        long,
-        help = "List of safekeepers endpoint will talk to. Normally neon_local chooses them on its own, but this option allows to override."
-    )]
+    #[clap(long)]
     safekeepers: Option<String>,
 
     #[clap(
@@ -628,9 +618,9 @@ struct EndpointStartCmdArgs {
     )]
     allow_multiple: bool,
 
-    #[clap(short = 't', long, value_parser= humantime::parse_duration, help = "timeout until we fail the command")]
-    #[arg(default_value = "90s")]
-    start_timeout: Duration,
+    #[clap(short = 't', long, help = "timeout until we fail the command")]
+    #[arg(default_value = "10s")]
+    start_timeout: humantime::Duration,
 }
 
 #[derive(clap::Args)]
@@ -931,9 +921,7 @@ fn handle_init(args: &InitCmdArgs) -> anyhow::Result<LocalEnv> {
     let init_conf: NeonLocalInitConf = if let Some(config_path) = &args.config {
         // User (likely the Python test suite) provided a description of the environment.
         if args.num_pageservers.is_some() {
-            bail!(
-                "Cannot specify both --num-pageservers and --config, use key `pageservers` in the --config file instead"
-            );
+            bail!("Cannot specify both --num-pageservers and --config, use key `pageservers` in the --config file instead");
         }
         // load and parse the file
         let contents = std::fs::read_to_string(config_path).with_context(|| {
@@ -965,7 +953,6 @@ fn handle_init(args: &InitCmdArgs) -> anyhow::Result<LocalEnv> {
                         id: pageserver_id,
                         listen_pg_addr: format!("127.0.0.1:{pg_port}"),
                         listen_http_addr: format!("127.0.0.1:{http_port}"),
-                        listen_https_addr: None,
                         pg_auth_type: AuthType::Trust,
                         http_auth_type: AuthType::Trust,
                         other: Default::default(),
@@ -980,7 +967,6 @@ fn handle_init(args: &InitCmdArgs) -> anyhow::Result<LocalEnv> {
             default_tenant_id: TenantId::from_array(std::array::from_fn(|_| 0)),
             storage_controller: None,
             control_plane_compute_hook_api: None,
-            generate_local_ssl_certs: false,
         }
     };
 
@@ -1131,16 +1117,12 @@ async fn handle_tenant(subcmd: &TenantCmd, env: &mut local_env::LocalEnv) -> any
             let tenant_id = get_tenant_id(args.tenant_id, env)?;
             let tenant_conf: HashMap<_, _> =
                 args.config.iter().flat_map(|c| c.split_once(':')).collect();
-            let config = PageServerNode::parse_config(tenant_conf)?;
 
-            let req = TenantConfigRequest { tenant_id, config };
-
-            let storage_controller = StorageController::from_env(env);
-            storage_controller
-                .set_tenant_config(&req)
+            pageserver
+                .tenant_config(tenant_id, tenant_conf)
                 .await
                 .with_context(|| format!("Tenant config failed for tenant with id {tenant_id}"))?;
-            println!("tenant {tenant_id} successfully configured via storcon");
+            println!("tenant {tenant_id} successfully configured on the pageserver");
         }
     }
     Ok(())
@@ -1333,14 +1315,10 @@ async fn handle_endpoint(subcmd: &EndpointCmd, env: &local_env::LocalEnv) -> Res
 
             match (mode, args.hot_standby) {
                 (ComputeMode::Static(_), true) => {
-                    bail!(
-                        "Cannot start a node in hot standby mode when it is already configured as a static replica"
-                    )
+                    bail!("Cannot start a node in hot standby mode when it is already configured as a static replica")
                 }
                 (ComputeMode::Primary, true) => {
-                    bail!(
-                        "Cannot start a node as a hot standby replica, it is already configured as primary node"
-                    )
+                    bail!("Cannot start a node as a hot standby replica, it is already configured as primary node")
                 }
                 _ => {}
             }
@@ -1367,7 +1345,6 @@ async fn handle_endpoint(subcmd: &EndpointCmd, env: &local_env::LocalEnv) -> Res
             let pageserver_id = args.endpoint_pageserver_id;
             let remote_ext_config = &args.remote_ext_config;
 
-            let safekeepers_generation = args.safekeepers_generation.map(SafekeeperGeneration::new);
             // If --safekeepers argument is given, use only the listed
             // safekeeper nodes; otherwise all from the env.
             let safekeepers = if let Some(safekeepers) = parse_safekeepers(&args.safekeepers)? {
@@ -1443,13 +1420,11 @@ async fn handle_endpoint(subcmd: &EndpointCmd, env: &local_env::LocalEnv) -> Res
             endpoint
                 .start(
                     &auth_token,
-                    safekeepers_generation,
                     safekeepers,
                     pageservers,
                     remote_ext_config.as_ref(),
                     stripe_size.0 as usize,
                     args.create_test_user,
-                    args.start_timeout,
                 )
                 .await?;
         }

control_plane/src/broker.rs

@@ -8,6 +8,7 @@
 use std::time::Duration;
 
 use anyhow::Context;
+
 use camino::Utf8PathBuf;
 
 use crate::{background_process, local_env};

control_plane/src/endpoint.rs

@@ -37,24 +37,29 @@
 //! ```
 //!
 use std::collections::BTreeMap;
-use std::net::{IpAddr, Ipv4Addr, SocketAddr, TcpStream};
+use std::net::IpAddr;
+use std::net::Ipv4Addr;
+use std::net::SocketAddr;
+use std::net::TcpStream;
 use std::path::PathBuf;
 use std::process::Command;
 use std::str::FromStr;
 use std::sync::Arc;
-use std::time::{Duration, Instant, SystemTime, UNIX_EPOCH};
+use std::time::Duration;
+use std::time::SystemTime;
+use std::time::UNIX_EPOCH;
 
-use anyhow::{Context, Result, anyhow, bail};
+use anyhow::{anyhow, bail, Context, Result};
 use compute_api::requests::ConfigurationRequest;
-use compute_api::responses::{ComputeCtlConfig, ComputeStatus, ComputeStatusResponse};
-use compute_api::spec::{
-    Cluster, ComputeAudit, ComputeFeature, ComputeMode, ComputeSpec, Database, PgIdent,
-    RemoteExtSpec, Role,
-};
-use nix::sys::signal::{Signal, kill};
+use compute_api::responses::ComputeCtlConfig;
+use compute_api::spec::Database;
+use compute_api::spec::PgIdent;
+use compute_api::spec::RemoteExtSpec;
+use compute_api::spec::Role;
+use nix::sys::signal::kill;
+use nix::sys::signal::Signal;
 use pageserver_api::shard::ShardStripeSize;
 use reqwest::header::CONTENT_TYPE;
-use safekeeper_api::membership::SafekeeperGeneration;
 use serde::{Deserialize, Serialize};
 use tracing::debug;
 use url::Host;
@@ -64,6 +69,9 @@ use crate::local_env::LocalEnv;
 use crate::postgresql_conf::PostgresConf;
 use crate::storage_controller::StorageController;
 
+use compute_api::responses::{ComputeStatus, ComputeStatusResponse};
+use compute_api::spec::{Cluster, ComputeFeature, ComputeMode, ComputeSpec};
+
 // contents of a endpoint.json file
 #[derive(Serialize, Deserialize, PartialEq, Eq, Clone, Debug)]
 pub struct EndpointConf {
@@ -229,9 +237,7 @@ impl ComputeControlPlane {
             });
 
             if let Some((key, _)) = duplicates.next() {
-                bail!(
-                    "attempting to create a duplicate primary endpoint on tenant {tenant_id}, timeline {timeline_id}: endpoint {key:?} exists already. please don't do this, it is not supported."
-                );
+                bail!("attempting to create a duplicate primary endpoint on tenant {tenant_id}, timeline {timeline_id}: endpoint {key:?} exists already. please don't do this, it is not supported.");
             }
         }
         Ok(())
@@ -578,17 +584,14 @@ impl Endpoint {
         Ok(safekeeper_connstrings)
     }
 
-    #[allow(clippy::too_many_arguments)]
     pub async fn start(
         &self,
         auth_token: &Option<String>,
-        safekeepers_generation: Option<SafekeeperGeneration>,
         safekeepers: Vec<NodeId>,
         pageservers: Vec<(Host, u16)>,
         remote_ext_config: Option<&String>,
         shard_stripe_size: usize,
         create_test_user: bool,
-        start_timeout: Duration,
     ) -> Result<()> {
         if self.status() == EndpointStatus::Running {
             anyhow::bail!("The endpoint is already running");
@@ -660,7 +663,6 @@ impl Endpoint {
             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,
@@ -669,7 +671,6 @@ impl Endpoint {
             local_proxy_config: None,
             reconfigure_concurrency: self.reconfigure_concurrency,
             drop_subscriptions_before_start: self.drop_subscriptions_before_start,
-            audit_log_level: ComputeAudit::Disabled,
         };
 
         // this strange code is needed to support respec() in tests
@@ -777,18 +778,17 @@ impl Endpoint {
         std::fs::write(pidfile_path, pid.to_string())?;
 
         // Wait for it to start
+        let mut attempt = 0;
         const ATTEMPT_INTERVAL: Duration = Duration::from_millis(100);
-        let start_at = Instant::now();
+        const MAX_ATTEMPTS: u32 = 10 * 90; // Wait up to 1.5 min
         loop {
+            attempt += 1;
             match self.get_status().await {
                 Ok(state) => {
                     match state.status {
                         ComputeStatus::Init => {
-                            if Instant::now().duration_since(start_at) > start_timeout {
-                                bail!(
-                                    "compute startup timed out {:?}; still in Init state",
-                                    start_timeout
-                                );
+                            if attempt == MAX_ATTEMPTS {
+                                bail!("compute startup timed out; still in Init state");
                             }
                             // keep retrying
                         }
@@ -815,11 +815,8 @@ impl Endpoint {
                     }
                 }
                 Err(e) => {
-                    if Instant::now().duration_since(start_at) > start_timeout {
-                        return Err(e).context(format!(
-                            "timed out {:?} waiting to connect to compute_ctl HTTP",
-                            start_timeout,
-                        ));
+                    if attempt == MAX_ATTEMPTS {
+                        return Err(e).context("timed out waiting to connect to compute_ctl HTTP");
                     }
                 }
             }

control_plane/src/local_env.rs

@@ -3,22 +3,28 @@
 //! Now it also provides init method which acts like a stub for proper installation
 //! script which will use local paths.
 
-use std::collections::HashMap;
-use std::net::{IpAddr, Ipv4Addr, SocketAddr};
-use std::path::{Path, PathBuf};
-use std::process::{Command, Stdio};
-use std::time::Duration;
-use std::{env, fs};
+use anyhow::{bail, Context};
 
-use anyhow::{Context, bail};
 use clap::ValueEnum;
 use postgres_backend::AuthType;
 use reqwest::Url;
 use serde::{Deserialize, Serialize};
-use utils::auth::{Claims, encode_from_key_file};
-use utils::id::{NodeId, TenantId, TenantTimelineId, TimelineId};
+use std::collections::HashMap;
+use std::env;
+use std::fs;
+use std::net::IpAddr;
+use std::net::Ipv4Addr;
+use std::net::SocketAddr;
+use std::path::{Path, PathBuf};
+use std::process::{Command, Stdio};
+use std::time::Duration;
+use utils::{
+    auth::{encode_from_key_file, Claims},
+    id::{NodeId, TenantId, TenantTimelineId, TimelineId},
+};
 
-use crate::pageserver::{PAGESERVER_REMOTE_STORAGE_DIR, PageServerNode};
+use crate::pageserver::PageServerNode;
+use crate::pageserver::PAGESERVER_REMOTE_STORAGE_DIR;
 use crate::safekeeper::SafekeeperNode;
 
 pub const DEFAULT_PG_VERSION: u32 = 16;
@@ -81,10 +87,6 @@ pub struct LocalEnv {
     // but deserialization into a generic toml object as `toml::Value::try_from` fails with an error.
     // https://toml.io/en/v1.0.0 does not contain a concept of "a table inside another table".
     pub branch_name_mappings: HashMap<String, Vec<(TenantId, TimelineId)>>,
-
-    /// Flag to generate SSL certificates for components that need it.
-    /// Also generates root CA certificate that is used to sign all other certificates.
-    pub generate_local_ssl_certs: bool,
 }
 
 /// On-disk state stored in `.neon/config`.
@@ -106,10 +108,6 @@ pub struct OnDiskConfig {
     pub control_plane_api: Option<Url>,
     pub control_plane_compute_hook_api: Option<Url>,
     branch_name_mappings: HashMap<String, Vec<(TenantId, TimelineId)>>,
-    // Note: skip serializing because in compat tests old storage controller fails
-    // to load new config file. May be removed after this field is in release branch.
-    #[serde(skip_serializing_if = "std::ops::Not::not")]
-    pub generate_local_ssl_certs: bool,
 }
 
 fn fail_if_pageservers_field_specified<'de, D>(_: D) -> Result<Vec<PageServerConf>, D::Error>
@@ -137,7 +135,6 @@ pub struct NeonLocalInitConf {
     pub safekeepers: Vec<SafekeeperConf>,
     pub control_plane_api: Option<Url>,
     pub control_plane_compute_hook_api: Option<Option<Url>>,
-    pub generate_local_ssl_certs: bool,
 }
 
 /// Broker config for cluster internal communication.
@@ -174,11 +171,6 @@ pub struct NeonStorageControllerConf {
 
     #[serde(with = "humantime_serde")]
     pub long_reconcile_threshold: Option<Duration>,
-
-    #[serde(default)]
-    pub use_https_pageserver_api: bool,
-
-    pub timelines_onto_safekeepers: bool,
 }
 
 impl NeonStorageControllerConf {
@@ -202,8 +194,6 @@ impl Default for NeonStorageControllerConf {
             max_secondary_lag_bytes: None,
             heartbeat_interval: Self::DEFAULT_HEARTBEAT_INTERVAL,
             long_reconcile_threshold: None,
-            use_https_pageserver_api: false,
-            timelines_onto_safekeepers: false,
         }
     }
 }
@@ -233,7 +223,6 @@ pub struct PageServerConf {
     pub id: NodeId,
     pub listen_pg_addr: String,
     pub listen_http_addr: String,
-    pub listen_https_addr: Option<String>,
     pub pg_auth_type: AuthType,
     pub http_auth_type: AuthType,
     pub no_sync: bool,
@@ -245,7 +234,6 @@ impl Default for PageServerConf {
             id: NodeId(0),
             listen_pg_addr: String::new(),
             listen_http_addr: String::new(),
-            listen_https_addr: None,
             pg_auth_type: AuthType::Trust,
             http_auth_type: AuthType::Trust,
             no_sync: false,
@@ -261,7 +249,6 @@ pub struct NeonLocalInitPageserverConf {
     pub id: NodeId,
     pub listen_pg_addr: String,
     pub listen_http_addr: String,
-    pub listen_https_addr: Option<String>,
     pub pg_auth_type: AuthType,
     pub http_auth_type: AuthType,
     #[serde(default, skip_serializing_if = "std::ops::Not::not")]
@@ -276,7 +263,6 @@ impl From<&NeonLocalInitPageserverConf> for PageServerConf {
             id,
             listen_pg_addr,
             listen_http_addr,
-            listen_https_addr,
             pg_auth_type,
             http_auth_type,
             no_sync,
@@ -286,7 +272,6 @@ impl From<&NeonLocalInitPageserverConf> for PageServerConf {
             id: *id,
             listen_pg_addr: listen_pg_addr.clone(),
             listen_http_addr: listen_http_addr.clone(),
-            listen_https_addr: listen_https_addr.clone(),
             pg_auth_type: *pg_auth_type,
             http_auth_type: *http_auth_type,
             no_sync: *no_sync,
@@ -431,41 +416,6 @@ impl LocalEnv {
         }
     }
 
-    pub fn ssl_ca_cert_path(&self) -> Option<PathBuf> {
-        if self.generate_local_ssl_certs {
-            Some(self.base_data_dir.join("rootCA.crt"))
-        } else {
-            None
-        }
-    }
-
-    pub fn ssl_ca_key_path(&self) -> Option<PathBuf> {
-        if self.generate_local_ssl_certs {
-            Some(self.base_data_dir.join("rootCA.key"))
-        } else {
-            None
-        }
-    }
-
-    pub fn generate_ssl_ca_cert(&self) -> anyhow::Result<()> {
-        let cert_path = self.ssl_ca_cert_path().unwrap();
-        let key_path = self.ssl_ca_key_path().unwrap();
-        if !fs::exists(cert_path.as_path())? {
-            generate_ssl_ca_cert(cert_path.as_path(), key_path.as_path())?;
-        }
-        Ok(())
-    }
-
-    pub fn generate_ssl_cert(&self, cert_path: &Path, key_path: &Path) -> anyhow::Result<()> {
-        self.generate_ssl_ca_cert()?;
-        generate_ssl_cert(
-            cert_path,
-            key_path,
-            self.ssl_ca_cert_path().unwrap().as_path(),
-            self.ssl_ca_key_path().unwrap().as_path(),
-        )
-    }
-
     /// Inspect the base data directory and extract the instance id and instance directory path
     /// for all storage controller instances
     pub async fn storage_controller_instances(&self) -> std::io::Result<Vec<(u8, PathBuf)>> {
@@ -515,9 +465,7 @@ impl LocalEnv {
             if old_timeline_id == &timeline_id {
                 Ok(())
             } else {
-                bail!(
-                    "branch '{branch_name}' is already mapped to timeline {old_timeline_id}, cannot map to another timeline {timeline_id}"
-                );
+                bail!("branch '{branch_name}' is already mapped to timeline {old_timeline_id}, cannot map to another timeline {timeline_id}");
             }
         } else {
             existing_values.push((tenant_id, timeline_id));
@@ -575,7 +523,6 @@ impl LocalEnv {
                 control_plane_api,
                 control_plane_compute_hook_api,
                 branch_name_mappings,
-                generate_local_ssl_certs,
             } = on_disk_config;
             LocalEnv {
                 base_data_dir: repopath.to_owned(),
@@ -590,7 +537,6 @@ impl LocalEnv {
                 control_plane_api: control_plane_api.unwrap(),
                 control_plane_compute_hook_api,
                 branch_name_mappings,
-                generate_local_ssl_certs,
             }
         };
 
@@ -626,7 +572,6 @@ impl LocalEnv {
                 struct PageserverConfigTomlSubset {
                     listen_pg_addr: String,
                     listen_http_addr: String,
-                    listen_https_addr: Option<String>,
                     pg_auth_type: AuthType,
                     http_auth_type: AuthType,
                     #[serde(default)]
@@ -651,7 +596,6 @@ impl LocalEnv {
                 let PageserverConfigTomlSubset {
                     listen_pg_addr,
                     listen_http_addr,
-                    listen_https_addr,
                     pg_auth_type,
                     http_auth_type,
                     no_sync,
@@ -669,7 +613,6 @@ impl LocalEnv {
                     },
                     listen_pg_addr,
                     listen_http_addr,
-                    listen_https_addr,
                     pg_auth_type,
                     http_auth_type,
                     no_sync,
@@ -697,7 +640,6 @@ impl LocalEnv {
                 control_plane_api: Some(self.control_plane_api.clone()),
                 control_plane_compute_hook_api: self.control_plane_compute_hook_api.clone(),
                 branch_name_mappings: self.branch_name_mappings.clone(),
-                generate_local_ssl_certs: self.generate_local_ssl_certs,
             },
         )
     }
@@ -780,7 +722,6 @@ impl LocalEnv {
             safekeepers,
             control_plane_api,
             control_plane_compute_hook_api,
-            generate_local_ssl_certs,
         } = conf;
 
         // Find postgres binaries.
@@ -829,13 +770,8 @@ impl LocalEnv {
             control_plane_api: control_plane_api.unwrap(),
             control_plane_compute_hook_api: control_plane_compute_hook_api.unwrap_or_default(),
             branch_name_mappings: Default::default(),
-            generate_local_ssl_certs,
         };
 
-        if generate_local_ssl_certs {
-            env.generate_ssl_ca_cert()?;
-        }
-
         // create endpoints dir
         fs::create_dir_all(env.endpoints_path())?;
 
@@ -919,80 +855,3 @@ fn generate_auth_keys(private_key_path: &Path, public_key_path: &Path) -> anyhow
     }
     Ok(())
 }
-
-fn generate_ssl_ca_cert(cert_path: &Path, key_path: &Path) -> anyhow::Result<()> {
-    // openssl req -x509 -newkey rsa:2048 -nodes -subj "/CN=Neon Local CA" -days 36500 \
-    // -out rootCA.crt -keyout rootCA.key
-    let keygen_output = Command::new("openssl")
-        .args([
-            "req", "-x509", "-newkey", "rsa:2048", "-nodes", "-days", "36500",
-        ])
-        .args(["-subj", "/CN=Neon Local CA"])
-        .args(["-out", cert_path.to_str().unwrap()])
-        .args(["-keyout", key_path.to_str().unwrap()])
-        .output()
-        .context("failed to generate CA certificate")?;
-    if !keygen_output.status.success() {
-        bail!(
-            "openssl failed: '{}'",
-            String::from_utf8_lossy(&keygen_output.stderr)
-        );
-    }
-    Ok(())
-}
-
-fn generate_ssl_cert(
-    cert_path: &Path,
-    key_path: &Path,
-    ca_cert_path: &Path,
-    ca_key_path: &Path,
-) -> anyhow::Result<()> {
-    // Generate Certificate Signing Request (CSR).
-    let mut csr_path = cert_path.to_path_buf();
-    csr_path.set_extension(".csr");
-
-    // openssl req -new -nodes -newkey rsa:2048 -keyout server.key -out server.csr \
-    // -subj "/CN=localhost" -addext "subjectAltName=DNS:localhost,IP:127.0.0.1"
-    let keygen_output = Command::new("openssl")
-        .args(["req", "-new", "-nodes"])
-        .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()])
-        .args(["-out", csr_path.to_str().unwrap()])
-        .output()
-        .context("failed to generate CSR")?;
-    if !keygen_output.status.success() {
-        bail!(
-            "openssl failed: '{}'",
-            String::from_utf8_lossy(&keygen_output.stderr)
-        );
-    }
-
-    // Sign CSR with CA key.
-    //
-    // openssl x509 -req -in server.csr -CA rootCA.crt -CAkey rootCA.key -CAcreateserial \
-    // -out server.crt -days 36500 -copy_extensions copyall
-    let keygen_output = Command::new("openssl")
-        .args(["x509", "-req"])
-        .args(["-in", csr_path.to_str().unwrap()])
-        .args(["-CA", ca_cert_path.to_str().unwrap()])
-        .args(["-CAkey", ca_key_path.to_str().unwrap()])
-        .arg("-CAcreateserial")
-        .args(["-out", cert_path.to_str().unwrap()])
-        .args(["-days", "36500"])
-        .args(["-copy_extensions", "copyall"])
-        .output()
-        .context("failed to sign CSR")?;
-    if !keygen_output.status.success() {
-        bail!(
-            "openssl failed: '{}'",
-            String::from_utf8_lossy(&keygen_output.stderr)
-        );
-    }
-
-    // Remove CSR file as it's not needed anymore.
-    fs::remove_file(csr_path)?;
-
-    Ok(())
-}

control_plane/src/pageserver.rs

@@ -7,6 +7,7 @@
 //! ```
 //!
 use std::collections::HashMap;
+
 use std::io;
 use std::io::Write;
 use std::num::NonZeroU64;
@@ -14,20 +15,22 @@ use std::path::PathBuf;
 use std::str::FromStr;
 use std::time::Duration;
 
-use anyhow::{Context, bail};
+use anyhow::{bail, Context};
 use camino::Utf8PathBuf;
 use pageserver_api::models::{self, TenantInfo, TimelineInfo};
 use pageserver_api::shard::TenantShardId;
 use pageserver_client::mgmt_api;
 use postgres_backend::AuthType;
-use postgres_connection::{PgConnectionConfig, parse_host_port};
-use reqwest::Certificate;
+use postgres_connection::{parse_host_port, PgConnectionConfig};
 use utils::auth::{Claims, Scope};
-use utils::id::{NodeId, TenantId, TimelineId};
-use utils::lsn::Lsn;
+use utils::id::NodeId;
+use utils::{
+    id::{TenantId, TimelineId},
+    lsn::Lsn,
+};
 
-use crate::background_process;
-use crate::local_env::{LocalEnv, NeonLocalInitPageserverConf, PageServerConf};
+use crate::local_env::{NeonLocalInitPageserverConf, PageServerConf};
+use crate::{background_process, local_env::LocalEnv};
 
 /// Directory within .neon which will be used by default for LocalFs remote storage.
 pub const PAGESERVER_REMOTE_STORAGE_DIR: &str = "local_fs_remote_storage/pageserver";
@@ -50,29 +53,12 @@ impl PageServerNode {
         let (host, port) =
             parse_host_port(&conf.listen_pg_addr).expect("Unable to parse listen_pg_addr");
         let port = port.unwrap_or(5432);
-
-        let ssl_ca_cert = env.ssl_ca_cert_path().map(|ssl_ca_file| {
-            let buf = std::fs::read(ssl_ca_file).expect("SSL root CA file should exist");
-            Certificate::from_pem(&buf).expect("CA certificate should be valid")
-        });
-
-        let endpoint = if env.storage_controller.use_https_pageserver_api {
-            format!(
-                "https://{}",
-                conf.listen_https_addr.as_ref().expect(
-                    "listen https address should be specified if use_https_pageserver_api is on"
-                )
-            )
-        } else {
-            format!("http://{}", conf.listen_http_addr)
-        };
-
         Self {
             pg_connection_config: PgConnectionConfig::new_host_port(host, port),
             conf: conf.clone(),
             env: env.clone(),
             http_client: mgmt_api::Client::new(
-                endpoint,
+                format!("http://{}", conf.listen_http_addr),
                 {
                     match conf.http_auth_type {
                         AuthType::Trust => None,
@@ -83,9 +69,7 @@ impl PageServerNode {
                     }
                 }
                 .as_deref(),
-                ssl_ca_cert,
-            )
-            .expect("Client constructs with no errors"),
+            ),
         }
     }
 
@@ -97,11 +81,7 @@ impl PageServerNode {
         &self,
         conf: NeonLocalInitPageserverConf,
     ) -> anyhow::Result<toml_edit::DocumentMut> {
-        assert_eq!(
-            &PageServerConf::from(&conf),
-            &self.conf,
-            "during neon_local init, we derive the runtime state of ps conf (self.conf) from the --config flag fully"
-        );
+        assert_eq!(&PageServerConf::from(&conf), &self.conf, "during neon_local init, we derive the runtime state of ps conf (self.conf) from the --config flag fully");
 
         // TODO(christian): instead of what we do here, create a pageserver_api::config::ConfigToml (PR #7656)
 
@@ -240,13 +220,6 @@ impl PageServerNode {
             .context("write identity toml")?;
         drop(identity_toml);
 
-        if self.env.generate_local_ssl_certs {
-            self.env.generate_ssl_cert(
-                datadir.join("server.crt").as_path(),
-                datadir.join("server.key").as_path(),
-            )?;
-        }
-
         // TODO: invoke a TBD config-check command to validate that pageserver will start with the written config
 
         // Write metadata file, used by pageserver on startup to register itself with
@@ -257,15 +230,6 @@ impl PageServerNode {
             parse_host_port(&self.conf.listen_http_addr).expect("Unable to parse listen_http_addr");
         let http_port = http_port.unwrap_or(9898);
 
-        let https_port = match self.conf.listen_https_addr.as_ref() {
-            Some(https_addr) => {
-                let (_https_host, https_port) =
-                    parse_host_port(https_addr).expect("Unable to parse listen_https_addr");
-                Some(https_port.unwrap_or(9899))
-            }
-            None => None,
-        };
-
         // Intentionally hand-craft JSON: this acts as an implicit format compat test
         // in case the pageserver-side structure is edited, and reflects the real life
         // situation: the metadata is written by some other script.
@@ -276,7 +240,6 @@ impl PageServerNode {
                 postgres_port: self.pg_connection_config.port(),
                 http_host: "localhost".to_string(),
                 http_port,
-                https_port,
                 other: HashMap::from([(
                     "availability_zone_id".to_string(),
                     serde_json::json!(az_id),

control_plane/src/postgresql_conf.rs

@@ -1,6 +1,3 @@
-use std::collections::HashMap;
-use std::fmt;
-
 ///
 /// Module for parsing postgresql.conf file.
 ///
@@ -9,6 +6,8 @@ use std::fmt;
 /// funny stuff like include-directives or funny escaping.
 use once_cell::sync::Lazy;
 use regex::Regex;
+use std::collections::HashMap;
+use std::fmt;
 
 /// In-memory representation of a postgresql.conf file
 #[derive(Default, Debug)]

control_plane/src/safekeeper.rs

@@ -14,15 +14,18 @@ use std::{io, result};
 
 use anyhow::Context;
 use camino::Utf8PathBuf;
-use http_utils::error::HttpErrorBody;
 use postgres_connection::PgConnectionConfig;
 use reqwest::{IntoUrl, Method};
 use thiserror::Error;
+
+use http_utils::error::HttpErrorBody;
 use utils::auth::{Claims, Scope};
 use utils::id::NodeId;
 
-use crate::background_process;
-use crate::local_env::{LocalEnv, SafekeeperConf};
+use crate::{
+    background_process,
+    local_env::{LocalEnv, SafekeeperConf},
+};
 
 #[derive(Error, Debug)]
 pub enum SafekeeperHttpError {

control_plane/src/storage_controller.rs

@@ -1,36 +1,44 @@
-use std::ffi::OsStr;
-use std::fs;
-use std::net::SocketAddr;
-use std::path::PathBuf;
-use std::process::ExitStatus;
-use std::str::FromStr;
-use std::sync::OnceLock;
-use std::time::{Duration, Instant};
-
+use crate::{
+    background_process,
+    local_env::{LocalEnv, NeonStorageControllerConf},
+};
 use camino::{Utf8Path, Utf8PathBuf};
 use hyper0::Uri;
 use nix::unistd::Pid;
-use pageserver_api::controller_api::{
-    NodeConfigureRequest, NodeDescribeResponse, NodeRegisterRequest, TenantCreateRequest,
-    TenantCreateResponse, TenantLocateResponse,
+use pageserver_api::{
+    controller_api::{
+        NodeConfigureRequest, NodeDescribeResponse, NodeRegisterRequest, TenantCreateRequest,
+        TenantCreateResponse, TenantLocateResponse, TenantShardMigrateRequest,
+        TenantShardMigrateResponse,
+    },
+    models::{
+        TenantShardSplitRequest, TenantShardSplitResponse, TimelineCreateRequest, TimelineInfo,
+    },
+    shard::{ShardStripeSize, TenantShardId},
 };
-use pageserver_api::models::{TenantConfigRequest, TimelineCreateRequest, TimelineInfo};
-use pageserver_api::shard::TenantShardId;
 use pageserver_client::mgmt_api::ResponseErrorMessageExt;
 use postgres_backend::AuthType;
 use reqwest::Method;
-use serde::de::DeserializeOwned;
-use serde::{Deserialize, Serialize};
+use serde::{de::DeserializeOwned, Deserialize, Serialize};
+use std::{
+    ffi::OsStr,
+    fs,
+    net::SocketAddr,
+    path::PathBuf,
+    process::ExitStatus,
+    str::FromStr,
+    sync::OnceLock,
+    time::{Duration, Instant},
+};
 use tokio::process::Command;
 use tracing::instrument;
 use url::Url;
-use utils::auth::{Claims, Scope, encode_from_key_file};
-use utils::id::{NodeId, TenantId};
+use utils::{
+    auth::{encode_from_key_file, Claims, Scope},
+    id::{NodeId, TenantId},
+};
 use whoami::username;
 
-use crate::background_process;
-use crate::local_env::{LocalEnv, NeonStorageControllerConf};
-
 pub struct StorageController {
     env: LocalEnv,
     private_key: Option<Vec<u8>>,
@@ -88,8 +96,7 @@ pub struct AttachHookRequest {
 
 #[derive(Serialize, Deserialize)]
 pub struct AttachHookResponse {
-    #[serde(rename = "gen")]
-    pub generation: Option<u32>,
+    pub gen: Option<u32>,
 }
 
 #[derive(Serialize, Deserialize)]
@@ -534,14 +541,6 @@ impl StorageController {
             args.push("--start-as-candidate".to_string());
         }
 
-        if self.config.use_https_pageserver_api {
-            args.push("--use-https-pageserver-api".to_string());
-        }
-
-        if let Some(ssl_ca_file) = self.env.ssl_ca_cert_path() {
-            args.push(format!("--ssl-ca-file={}", ssl_ca_file.to_str().unwrap()));
-        }
-
         if let Some(private_key) = &self.private_key {
             let claims = Claims::new(None, Scope::PageServerApi);
             let jwt_token =
@@ -584,10 +583,6 @@ impl StorageController {
             self.env.base_data_dir.display()
         ));
 
-        if self.config.timelines_onto_safekeepers {
-            args.push("--timelines-onto-safekeepers".to_string());
-        }
-
         background_process::start_process(
             COMMAND,
             &instance_dir,
@@ -784,7 +779,7 @@ impl StorageController {
             )
             .await?;
 
-        Ok(response.generation)
+        Ok(response.gen)
     }
 
     #[instrument(skip(self))]
@@ -834,6 +829,41 @@ impl StorageController {
         .await
     }
 
+    #[instrument(skip(self))]
+    pub async fn tenant_migrate(
+        &self,
+        tenant_shard_id: TenantShardId,
+        node_id: NodeId,
+    ) -> anyhow::Result<TenantShardMigrateResponse> {
+        self.dispatch(
+            Method::PUT,
+            format!("control/v1/tenant/{tenant_shard_id}/migrate"),
+            Some(TenantShardMigrateRequest {
+                node_id,
+                migration_config: None,
+            }),
+        )
+        .await
+    }
+
+    #[instrument(skip(self), fields(%tenant_id, %new_shard_count))]
+    pub async fn tenant_split(
+        &self,
+        tenant_id: TenantId,
+        new_shard_count: u8,
+        new_stripe_size: Option<ShardStripeSize>,
+    ) -> anyhow::Result<TenantShardSplitResponse> {
+        self.dispatch(
+            Method::PUT,
+            format!("control/v1/tenant/{tenant_id}/shard_split"),
+            Some(TenantShardSplitRequest {
+                new_shard_count,
+                new_stripe_size,
+            }),
+        )
+        .await
+    }
+
     #[instrument(skip_all, fields(node_id=%req.node_id))]
     pub async fn node_register(&self, req: NodeRegisterRequest) -> anyhow::Result<()> {
         self.dispatch::<_, ()>(Method::POST, "control/v1/node".to_string(), Some(req))
@@ -878,9 +908,4 @@ impl StorageController {
         )
         .await
     }
-
-    pub async fn set_tenant_config(&self, req: &TenantConfigRequest) -> anyhow::Result<()> {
-        self.dispatch(Method::PUT, "v1/tenant/config".to_string(), Some(req))
-            .await
-    }
 }

control_plane/storcon_cli/src/main.rs

@@ -1,29 +1,35 @@
-use std::collections::{HashMap, HashSet};
-use std::path::PathBuf;
-use std::str::FromStr;
-use std::time::Duration;
+use futures::StreamExt;
+use std::{
+    collections::{HashMap, HashSet},
+    str::FromStr,
+    time::Duration,
+};
 
 use clap::{Parser, Subcommand};
-use futures::StreamExt;
-use pageserver_api::controller_api::{
-    AvailabilityZone, MigrationConfig, NodeAvailabilityWrapper, NodeConfigureRequest,
-    NodeDescribeResponse, NodeRegisterRequest, NodeSchedulingPolicy, NodeShardResponse,
-    PlacementPolicy, SafekeeperDescribeResponse, SafekeeperSchedulingPolicyRequest,
-    ShardSchedulingPolicy, ShardsPreferredAzsRequest, ShardsPreferredAzsResponse,
-    SkSchedulingPolicy, TenantCreateRequest, TenantDescribeResponse, TenantPolicyRequest,
-    TenantShardMigrateRequest, TenantShardMigrateResponse,
+use pageserver_api::{
+    controller_api::{
+        AvailabilityZone, NodeAvailabilityWrapper, NodeDescribeResponse, NodeShardResponse,
+        SafekeeperDescribeResponse, SafekeeperSchedulingPolicyRequest, ShardSchedulingPolicy,
+        ShardsPreferredAzsRequest, ShardsPreferredAzsResponse, SkSchedulingPolicy,
+        TenantCreateRequest, TenantDescribeResponse, TenantPolicyRequest,
+    },
+    models::{
+        EvictionPolicy, EvictionPolicyLayerAccessThreshold, LocationConfigSecondary,
+        ShardParameters, TenantConfig, TenantConfigPatchRequest, TenantConfigRequest,
+        TenantShardSplitRequest, TenantShardSplitResponse,
+    },
+    shard::{ShardStripeSize, TenantShardId},
 };
-use pageserver_api::models::{
-    EvictionPolicy, EvictionPolicyLayerAccessThreshold, ShardParameters, TenantConfig,
-    TenantConfigPatchRequest, TenantConfigRequest, TenantShardSplitRequest,
-    TenantShardSplitResponse,
-};
-use pageserver_api::shard::{ShardStripeSize, TenantShardId};
 use pageserver_client::mgmt_api::{self};
 use reqwest::{Method, StatusCode, Url};
-use storage_controller_client::control_api::Client;
 use utils::id::{NodeId, TenantId, TimelineId};
 
+use pageserver_api::controller_api::{
+    NodeConfigureRequest, NodeRegisterRequest, NodeSchedulingPolicy, PlacementPolicy,
+    TenantShardMigrateRequest, TenantShardMigrateResponse,
+};
+use storage_controller_client::control_api::Client;
+
 #[derive(Subcommand, Debug)]
 enum Command {
     /// Register a pageserver with the storage controller.  This shouldn't usually be necessary,
@@ -113,15 +119,6 @@ enum Command {
         tenant_shard_id: TenantShardId,
         #[arg(long)]
         node: NodeId,
-        #[arg(long, default_value_t = true, action = clap::ArgAction::Set)]
-        prewarm: bool,
-        #[arg(long, default_value_t = false, action = clap::ArgAction::Set)]
-        override_scheduler: bool,
-    },
-    /// Watch the location of a tenant shard evolve, e.g. while expecting it to migrate
-    TenantShardWatch {
-        #[arg(long)]
-        tenant_shard_id: TenantShardId,
     },
     /// Migrate the secondary location for a tenant shard to a specific pageserver.
     TenantShardMigrateSecondary {
@@ -158,6 +155,12 @@ enum Command {
         #[arg(long)]
         tenant_id: TenantId,
     },
+    /// For a tenant which hasn't been onboarded to the storage controller yet, add it in secondary
+    /// mode so that it can warm up content on a pageserver.
+    TenantWarmup {
+        #[arg(long)]
+        tenant_id: TenantId,
+    },
     TenantSetPreferredAz {
         #[arg(long)]
         tenant_id: TenantId,
@@ -273,10 +276,6 @@ struct Cli {
     /// a token with both scopes to use with this tool.
     jwt: Option<String>,
 
-    #[arg(long)]
-    /// Trusted root CA certificate to use in https APIs.
-    ssl_ca_file: Option<PathBuf>,
-
     #[command(subcommand)]
     command: Command,
 }
@@ -387,17 +386,9 @@ async fn main() -> anyhow::Result<()> {
 
     let storcon_client = Client::new(cli.api.clone(), cli.jwt.clone());
 
-    let ssl_ca_cert = match &cli.ssl_ca_file {
-        Some(ssl_ca_file) => {
-            let buf = tokio::fs::read(ssl_ca_file).await?;
-            Some(reqwest::Certificate::from_pem(&buf)?)
-        }
-        None => None,
-    };
-
     let mut trimmed = cli.api.to_string();
     trimmed.pop();
-    let vps_client = mgmt_api::Client::new(trimmed, cli.jwt.as_deref(), ssl_ca_cert)?;
+    let vps_client = mgmt_api::Client::new(trimmed, cli.jwt.as_deref());
 
     match cli.command {
         Command::NodeRegister {
@@ -635,43 +626,19 @@ async fn main() -> anyhow::Result<()> {
         Command::TenantShardMigrate {
             tenant_shard_id,
             node,
-            prewarm,
-            override_scheduler,
         } => {
-            let migration_config = MigrationConfig {
-                prewarm,
-                override_scheduler,
-                ..Default::default()
-            };
-
             let req = TenantShardMigrateRequest {
                 node_id: node,
-                origin_node_id: None,
-                migration_config,
+                migration_config: None,
             };
 
-            match storcon_client
+            storcon_client
                 .dispatch::<TenantShardMigrateRequest, TenantShardMigrateResponse>(
                     Method::PUT,
                     format!("control/v1/tenant/{tenant_shard_id}/migrate"),
                     Some(req),
                 )
-                .await
-            {
-                Err(mgmt_api::Error::ApiError(StatusCode::PRECONDITION_FAILED, msg)) => {
-                    anyhow::bail!(
-                        "Migration to {node} rejected, may require `--force` ({}) ",
-                        msg
-                    );
-                }
-                Err(e) => return Err(e.into()),
-                Ok(_) => {}
-            }
-
-            watch_tenant_shard(storcon_client, tenant_shard_id, Some(node)).await?;
-        }
-        Command::TenantShardWatch { tenant_shard_id } => {
-            watch_tenant_shard(storcon_client, tenant_shard_id, None).await?;
+                .await?;
         }
         Command::TenantShardMigrateSecondary {
             tenant_shard_id,
@@ -679,8 +646,7 @@ async fn main() -> anyhow::Result<()> {
         } => {
             let req = TenantShardMigrateRequest {
                 node_id: node,
-                origin_node_id: None,
-                migration_config: MigrationConfig::default(),
+                migration_config: None,
             };
 
             storcon_client
@@ -865,11 +831,97 @@ async fn main() -> anyhow::Result<()> {
                 )
                 .await?;
         }
+        Command::TenantWarmup { tenant_id } => {
+            let describe_response = storcon_client
+                .dispatch::<(), TenantDescribeResponse>(
+                    Method::GET,
+                    format!("control/v1/tenant/{tenant_id}"),
+                    None,
+                )
+                .await;
+            match describe_response {
+                Ok(describe) => {
+                    if matches!(describe.policy, PlacementPolicy::Secondary) {
+                        // Fine: it's already known to controller in secondary mode: calling
+                        // again to put it into secondary mode won't cause problems.
+                    } else {
+                        anyhow::bail!("Tenant already present with policy {:?}", describe.policy);
+                    }
+                }
+                Err(mgmt_api::Error::ApiError(StatusCode::NOT_FOUND, _)) => {
+                    // Fine: this tenant isn't know to the storage controller yet.
+                }
+                Err(e) => {
+                    // Unexpected API error
+                    return Err(e.into());
+                }
+            }
+
+            vps_client
+                .location_config(
+                    TenantShardId::unsharded(tenant_id),
+                    pageserver_api::models::LocationConfig {
+                        mode: pageserver_api::models::LocationConfigMode::Secondary,
+                        generation: None,
+                        secondary_conf: Some(LocationConfigSecondary { warm: true }),
+                        shard_number: 0,
+                        shard_count: 0,
+                        shard_stripe_size: ShardParameters::DEFAULT_STRIPE_SIZE.0,
+                        tenant_conf: TenantConfig::default(),
+                    },
+                    None,
+                    true,
+                )
+                .await?;
+
+            let describe_response = storcon_client
+                .dispatch::<(), TenantDescribeResponse>(
+                    Method::GET,
+                    format!("control/v1/tenant/{tenant_id}"),
+                    None,
+                )
+                .await?;
+
+            let secondary_ps_id = describe_response
+                .shards
+                .first()
+                .unwrap()
+                .node_secondary
+                .first()
+                .unwrap();
+
+            println!("Tenant {tenant_id} warming up on pageserver {secondary_ps_id}");
+            loop {
+                let (status, progress) = vps_client
+                    .tenant_secondary_download(
+                        TenantShardId::unsharded(tenant_id),
+                        Some(Duration::from_secs(10)),
+                    )
+                    .await?;
+                println!(
+                    "Progress: {}/{} layers, {}/{} bytes",
+                    progress.layers_downloaded,
+                    progress.layers_total,
+                    progress.bytes_downloaded,
+                    progress.bytes_total
+                );
+                match status {
+                    StatusCode::OK => {
+                        println!("Download complete");
+                        break;
+                    }
+                    StatusCode::ACCEPTED => {
+                        // Loop
+                    }
+                    _ => {
+                        anyhow::bail!("Unexpected download status: {status}");
+                    }
+                }
+            }
+        }
         Command::TenantDrop { tenant_id, unclean } => {
             if !unclean {
-                anyhow::bail!(
-                    "This command is not a tenant deletion, and uncleanly drops all controller state for the tenant.  If you know what you're doing, add `--unclean` to proceed."
-                )
+                anyhow::bail!("This command is not a tenant deletion, and uncleanly drops all controller state for the tenant.  If you know what you're doing, add `--unclean` to proceed.")
             }
             storcon_client
                 .dispatch::<(), ()>(
@@ -881,9 +933,7 @@ async fn main() -> anyhow::Result<()> {
         }
         Command::NodeDrop { node_id, unclean } => {
             if !unclean {
-                anyhow::bail!(
-                    "This command is not a clean node decommission, and uncleanly drops all controller state for the node, without checking if any tenants still refer to it.  If you know what you're doing, add `--unclean` to proceed."
-                )
+                anyhow::bail!("This command is not a clean node decommission, and uncleanly drops all controller state for the node, without checking if any tenants still refer to it.  If you know what you're doing, add `--unclean` to proceed.")
             }
             storcon_client
                 .dispatch::<(), ()>(Method::POST, format!("debug/v1/node/{node_id}/drop"), None)
@@ -1058,8 +1108,7 @@ async fn main() -> anyhow::Result<()> {
                                 format!("control/v1/tenant/{}/migrate", mv.tenant_shard_id),
                                 Some(TenantShardMigrateRequest {
                                     node_id: mv.to,
-                                    origin_node_id: Some(mv.from),
-                                    migration_config: MigrationConfig::default(),
+                                    migration_config: None,
                                 }),
                             )
                             .await
@@ -1238,68 +1287,3 @@ async fn main() -> anyhow::Result<()> {
 
     Ok(())
 }
-
-static WATCH_INTERVAL: Duration = Duration::from_secs(5);
-
-async fn watch_tenant_shard(
-    storcon_client: Client,
-    tenant_shard_id: TenantShardId,
-    until_migrated_to: Option<NodeId>,
-) -> anyhow::Result<()> {
-    if let Some(until_migrated_to) = until_migrated_to {
-        println!(
-            "Waiting for tenant shard {} to be migrated to node {}",
-            tenant_shard_id, until_migrated_to
-        );
-    }
-
-    loop {
-        let desc = storcon_client
-            .dispatch::<(), TenantDescribeResponse>(
-                Method::GET,
-                format!("control/v1/tenant/{}", tenant_shard_id.tenant_id),
-                None,
-            )
-            .await?;
-
-        // Output the current state of the tenant shard
-        let shard = desc
-            .shards
-            .iter()
-            .find(|s| s.tenant_shard_id == tenant_shard_id)
-            .ok_or(anyhow::anyhow!("Tenant shard not found"))?;
-        let summary = format!(
-            "attached: {} secondary: {} {}",
-            shard
-                .node_attached
-                .map(|n| format!("{}", n))
-                .unwrap_or("none".to_string()),
-            shard
-                .node_secondary
-                .iter()
-                .map(|n| n.to_string())
-                .collect::<Vec<_>>()
-                .join(","),
-            if shard.is_reconciling {
-                "(reconciler active)"
-            } else {
-                "(reconciler idle)"
-            }
-        );
-        println!("{}", summary);
-
-        // Maybe drop out if we finished migration
-        if let Some(until_migrated_to) = until_migrated_to {
-            if shard.node_attached == Some(until_migrated_to) && !shard.is_reconciling {
-                println!(
-                    "Tenant shard {} is now on node {}",
-                    tenant_shard_id, until_migrated_to
-                );
-                break;
-            }
-        }
-
-        tokio::time::sleep(WATCH_INTERVAL).await;
-    }
-    Ok(())
-}

deny.toml

@@ -27,14 +27,6 @@ yanked = "warn"
 id = "RUSTSEC-2023-0071"
 reason = "the marvin attack only affects private key decryption, not public key signature verification"
 
-[[advisories.ignore]]
-id = "RUSTSEC-2024-0436"
-reason = "The paste crate is a build-only dependency with no runtime components. It is unlikely to have any security impact."
-
-[[advisories.ignore]]
-id = "RUSTSEC-2025-0014"
-reason = "The humantime is widely used and is not easy to replace right now. It is unmaintained, but it has no known vulnerabilities to care about. #11179"
-
 # This section is considered when running `cargo deny check licenses`
 # More documentation for the licenses section can be found here:
 # https://embarkstudios.github.io/cargo-deny/checks/licenses/cfg.html

docker-compose/ext-src/pgtap-src/test-upgrade.patch

@@ -7,7 +7,7 @@ index f255fe6..0a0fa65 100644
  GENERATED_SCHEDULE_DEPS = $(TB_DIR)/all_tests $(TB_DIR)/exclude_tests
  REGRESS = --schedule $(TB_DIR)/run.sch # Set this again just to be safe
 -REGRESS_OPTS = --inputdir=test --max-connections=$(PARALLEL_CONN) --schedule $(SETUP_SCH) $(REGRESS_CONF)
-+REGRESS_OPTS = --use-existing --dbname=contrib_regression --inputdir=test --max-connections=$(PARALLEL_CONN) --schedule $(SETUP_SCH) $(REGRESS_CONF)
++REGRESS_OPTS = --use-existing --dbname=pgtap_regression --inputdir=test --max-connections=$(PARALLEL_CONN) --schedule $(SETUP_SCH) $(REGRESS_CONF)
  SETUP_SCH = test/schedule/main.sch # schedule to use for test setup; this can be forcibly changed by some targets!
  IGNORE_TESTS = $(notdir $(EXCLUDE_TEST_FILES:.sql=))
  PARALLEL_TESTS = $(filter-out $(IGNORE_TESTS),$(filter-out $(SERIAL_TESTS),$(ALL_TESTS)))

docker-compose/test_extensions_upgrade.sh

@@ -6,16 +6,12 @@ generate_id() {
     local -n resvar=$1
     printf -v resvar '%08x%08x%08x%08x' $SRANDOM $SRANDOM $SRANDOM $SRANDOM
 }
-echo "${OLD_COMPUTE_TAG}"
-echo "${NEW_COMPUTE_TAG}"
-echo "${TEST_EXTENSIONS_TAG}"
-if [ -z "${OLD_COMPUTE_TAG:-}" ] || [ -z "${NEW_COMPUTE_TAG:-}" ] || [ -z "${TEST_EXTENSIONS_TAG:-}" ]; then
-  echo OLD_COMPUTE_TAG, NEW_COMPUTE_TAG and TEST_EXTENSIONS_TAG must be set
+if [ -z ${OLD_COMPUTE_TAG+x} ] || [ -z ${NEW_COMPUTE_TAG+x} ] || [ -z "${OLD_COMPUTE_TAG}" ] || [ -z "${NEW_COMPUTE_TAG}" ]; then
+  echo OLD_COMPUTE_TAG and NEW_COMPUTE_TAG must be defined
   exit 1
 fi
 export PG_VERSION=${PG_VERSION:-16}
 export PG_TEST_VERSION=${PG_VERSION}
-# Waits for compute node is ready
 function wait_for_ready {
   TIME=0
   while ! docker compose logs compute_is_ready | grep -q "accepting connections" && [ ${TIME} -le 300 ] ; do
@@ -27,45 +23,11 @@ function wait_for_ready {
     exit 2
   fi
 }
-# Creates extensions. Gets a string with space-separated extensions as a parameter
 function create_extensions() {
   for ext in ${1}; do
     docker compose exec neon-test-extensions psql -X -v ON_ERROR_STOP=1 -d contrib_regression -c "CREATE EXTENSION IF NOT EXISTS ${ext} CASCADE"
   done
 }
-# Creates a new timeline. Gets the parent ID and an extension name as parameters.
-# Saves the timeline ID in the variable EXT_TIMELINE
-function create_timeline() {
-  generate_id new_timeline_id
-
-  PARAMS=(
-      -sbf
-      -X POST
-      -H "Content-Type: application/json"
-      -d "{\"new_timeline_id\": \"${new_timeline_id}\", \"pg_version\": ${PG_VERSION}, \"ancestor_timeline_id\": \"${1}\"}"
-      "http://127.0.0.1:9898/v1/tenant/${tenant_id}/timeline/"
-  )
-  result=$(curl "${PARAMS[@]}")
-  echo $result | jq .
-  EXT_TIMELINE[${2}]=${new_timeline_id}
-}
-# Checks if the timeline ID of the compute node is expected. Gets the timeline ID as a parameter
-function check_timeline() {
-    TID=$(docker compose exec neon-test-extensions psql -Aqt -c "SHOW neon.timeline_id")
-    if [ "${TID}" != "${1}" ]; then
-      echo Timeline mismatch
-      exit 1
-    fi
-}
-# Restarts the compute node with the required compute tag and timeline.
-# Accepts the tag for the compute node and the timeline as parameters.
-function restart_compute() {
-  docker compose down compute compute_is_ready
-  COMPUTE_TAG=${1} TENANT_ID=${tenant_id} TIMELINE_ID=${2} docker compose up --quiet-pull -d --build compute compute_is_ready
-  wait_for_ready
-  check_timeline ${2}
-}
-declare -A EXT_TIMELINE
 EXTENSIONS='[
 {"extname": "plv8", "extdir": "plv8-src"},
 {"extname": "vector", "extdir": "pgvector-src"},
@@ -97,10 +59,8 @@ COMPUTE_TAG=${OLD_COMPUTE_TAG} docker compose --profile test-extensions up --qui
 wait_for_ready
 docker compose exec neon-test-extensions psql -c "DROP DATABASE IF EXISTS contrib_regression"
 docker compose exec neon-test-extensions psql -c "CREATE DATABASE contrib_regression"
-tenant_id=$(docker compose exec neon-test-extensions psql -Aqt -c "SHOW neon.tenant_id")
-EXT_TIMELINE["main"]=$(docker compose exec neon-test-extensions psql -Aqt -c "SHOW neon.timeline_id")
-create_timeline "${EXT_TIMELINE["main"]}" init
-restart_compute "${OLD_COMPUTE_TAG}" "${EXT_TIMELINE["init"]}"
+docker compose exec neon-test-extensions psql -c "CREATE DATABASE pgtap_regression"
+docker compose exec neon-test-extensions psql -d pgtap_regression -c "CREATE EXTENSION pgtap"
 create_extensions "${EXTNAMES}"
 if [ "${FORCE_ALL_UPGRADE_TESTS:-false}" = true ]; then
   exts="${EXTNAMES}"
@@ -111,13 +71,29 @@ fi
 if [ -z "${exts}" ]; then
   echo "No extensions were upgraded"
 else
+  tenant_id=$(docker compose exec neon-test-extensions psql -Aqt -c "SHOW neon.tenant_id")
+  timeline_id=$(docker compose exec neon-test-extensions psql -Aqt -c "SHOW neon.timeline_id")
   for ext in ${exts}; do
     echo Testing ${ext}...
-    create_timeline "${EXT_TIMELINE["main"]}" ${ext}
     EXTDIR=$(echo ${EXTENSIONS} | jq -r '.[] | select(.extname=="'${ext}'") | .extdir')
-    restart_compute "${OLD_COMPUTE_TAG}" "${EXT_TIMELINE[${ext}]}"
-    docker compose exec neon-test-extensions psql -d contrib_regression -c "CREATE EXTENSION ${ext} CASCADE"
-    restart_compute "${NEW_COMPUTE_TAG}" "${EXT_TIMELINE[${ext}]}"
+    generate_id new_timeline_id
+    PARAMS=(
+        -sbf
+        -X POST
+        -H "Content-Type: application/json"
+        -d "{\"new_timeline_id\": \"${new_timeline_id}\", \"pg_version\": ${PG_VERSION}, \"ancestor_timeline_id\": \"${timeline_id}\"}"
+        "http://127.0.0.1:9898/v1/tenant/${tenant_id}/timeline/"
+    )
+    result=$(curl "${PARAMS[@]}")
+    echo $result | jq .
+    TENANT_ID=${tenant_id} TIMELINE_ID=${new_timeline_id} COMPUTE_TAG=${OLD_COMPUTE_TAG} docker compose down compute compute_is_ready
+    COMPUTE_TAG=${NEW_COMPUTE_TAG} TENANT_ID=${tenant_id} TIMELINE_ID=${new_timeline_id} docker compose up --quiet-pull -d --build compute compute_is_ready
+    wait_for_ready
+    TID=$(docker compose exec neon-test-extensions psql -Aqt -c "SHOW neon.timeline_id")
+    if [ ${TID} != ${new_timeline_id} ]; then
+      echo Timeline mismatch
+      exit 1
+    fi
     docker compose exec neon-test-extensions psql -d contrib_regression -c "\dx ${ext}"
     if ! docker compose exec neon-test-extensions sh -c /ext-src/${EXTDIR}/test-upgrade.sh; then
       docker  compose exec neon-test-extensions  cat /ext-src/${EXTDIR}/regression.diffs

docs/rfcs/041-rel-sparse-keyspace.md

@@ -1,201 +0,0 @@
-# Sparse Keyspace for Relation Directories
-
-## Summary
-
-This is an RFC describing a new storage strategy for storing relation directories.
-
-## Motivation
-
-Postgres maintains a directory structure for databases and relations. In Neon, we store these information
-by serializing the directory data in a single key (see `pgdatadir_mapping.rs`).
-
-```rust
-// DbDir:
-// 00 00000000 00000000 00000000 00   00000000
-
-// RelDir:
-// 00 SPCNODE  DBNODE   00000000 00   00000001 (Postgres never uses relfilenode 0)
-```
-
-We have a dedicated structure on the ingestion path to serialize the relation directory into this single key.
-
-```rust
-#[derive(Debug, Serialize, Deserialize, Default)]
-pub(crate) struct RelDirectory {
-    // Set of relations that exist. (relfilenode, forknum)
-    //
-    // TODO: Store it as a btree or radix tree or something else that spans multiple
-    // key-value pairs, if you have a lot of relations
-    pub(crate) rels: HashSet<(Oid, u8)>,
-}
-```
-
-The current codebase has the following three access patterns for the relation directory.
-
-1. Check if a relation exists.
-2. List all relations.
-3. Create/drop a relation.
-
-For (1), we currently have to get the reldir key, deserialize it, and check whether the relation exists in the
-hash set. For (2), we get the reldir key and the hash set. For (3), we need first to get
-and deserialize the key, add the new relation record to the hash set, and then serialize it and write it back.
-
-If we have 100k relations in a database, we would have a 100k-large hash set. Then, every
-relation created and dropped would have deserialized and serialized this 100k-large hash set. This makes the
-relation create/drop process to be quadratic. When we check if a relation exists in the ingestion path,
-we would have to deserialize this super big 100k-large key before checking if a single relation exists.
-
-In this RFC, we will propose a new way to store the reldir data in the sparse keyspace and propose how
-to seamlessly migrate users to use the new keyspace.
-
-The PoC patch is implemented in [PR10316](https://github.com/neondatabase/neon/pull/10316).
-
-## Key Mapping
-
-We will use the recently introduced sparse keyspace to store actual data. Sparse keyspace was proposed in
-[038-aux-file-v2.md](038-aux-file-v2.md). The original reldir has one single value of `HashSet<(Oid, u8)>`
-for each of the databases (identified as `spcnode, dbnode`). We encode the `Oid` (`relnode, forknum`),
-into the key.
-
-```plain
-(REL_DIR_KEY_PREFIX, spcnode, dbnode, relnode, forknum, 1) -> deleted
-(REL_DIR_KEY_PREFIX, spcnode, dbnode, relnode, forknum, 1) -> exists
-```
-
-Assume all reldir data are stored in this new keyspace; the 3 reldir operations we mentioned before can be
-implemented as follows.
-
-1. Check if a relation exists: check if the key maps to "exists".
-2. List all relations: scan the sprase keyspace over the `rel_dir_key_prefix`. Extract relnode and forknum from the key.
-3. Create/drop a relation: write "exists" or "deleted" to the corresponding key of the relation. The delete tombstone will
-   be removed during image layer generation upon compaction.
-
-Note that "exists" and "deleted" will be encoded as a single byte as two variants of an enum.
-The mapping is implemented as `rel_tag_sparse_key` in the PoC patch.
-
-## Changes to Sparse Keyspace
-
-Previously, we only used sparse keyspaces for the aux files, which did not carry over when branching. The reldir
-information needs to be preserved from the parent branch to the child branch. Therefore, the read path needs
-to be updated accordingly to accommodate such "inherited sparse keys". This is done in
-[PR#10313](https://github.com/neondatabase/neon/pull/10313).
-
-## Coexistence of the Old and New Keyspaces
-
-Migrating to the new keyspace will be done gradually: when we flip a config item to enable the new reldir keyspace, the
-ingestion path will start to write to the new keyspace and the old reldir data will be kept in the old one. The read
-path needs to combine the data from both keyspaces.
-
-Theoretically, we could do a rewrite at the startup time that scans all relation directories and copies that data into the
-new keyspace. However, this could take a long time, especially if we have thousands of tenants doing the migration
-process simultaneously after the pageserver restarts. Therefore, we propose the coexistence strategy so that the
-migration can happen seamlessly and imposes no potential downtime for the user.
-
-With the coexistence assumption, the 3 reldir operations will be implemented as follows:
-
-1. Check if a relation exists
-   - Check the new keyspace if the key maps to any value. If it maps to "exists" or "deleted", directly
-    return it to the user.
-   - Otherwise, deserialize the old reldir key and get the result.
-2. List all relations: scan the sparse keyspace over the `rel_dir_key_prefix` and deserialize the old reldir key.
-   Combine them to obtain the final result.
-3. Create/drop a relation: write "exists" or "deleted" to the corresponding key of the relation into the new keyspace.
-   - We assume no overwrite of relations will happen (i.e., the user won't create a relation at the same Oid). This will be implemented as a runtime check.
-   - For relation creation, we add `sparse_reldir_tableX -> exists` to the keyspace.
-   - For relation drop, we first check if the relation is recorded in the old keyspace. If yes, we deserialize the old reldir key,
-    remove the relation, and then write it back. Otherwise, we put `sparse_reldir_tableX -> deleted` to the keyspace.
-   - The delete tombstone will be removed during image layer generation upon compaction.
-
-This process ensures that the transition will not introduce any downtime and all new updates are written to the new keyspace. The total
-amount of data in the storage would be `O(relations_modifications)` and we can guarantee `O(current_relations)` after compaction.
-There could be some relations that exist in the old reldir key for a long time. Refer to the "Full Migration" section on how to deal
-with them. Plus, for relation modifications, it will have `O(old_relations)` complexity until we do the full migration, which gives
-us `O(1)` complexity after fully opt-in the sparse keyspace.
-
-The process also implies that a relation will only exists either in the old reldir key or in the new sparse keyspace. It is not possible
-to have a table to be recorded in the old reldir key while later having a delete tombstone for it in the sparse keyspace at any LSN.
-
-We will introduce a config item and an index_part record to record the current status of the migration process.
-
-- Config item `enable_reldir_v2`: controls whether the ingestion path writes the reldir info into the new keyspace.
-- `index_part.json` field `reldir_v2_status`: whether the timeline has written any key into the new reldir keyspace.
-
-If `enable_reldir_v2` is set to `true` and the timeline ingests the first key into the new reldir keyspace, it will update
-`index_part.json` to set `reldir_v2_status` to `Status::Migrating`. Even if `enable_reldir_v2` gets flipped back to
-`false` (i.e., when the pageserver restarts and such config isn't persisted), the read/write path will still
-read/write to the new keyspace to avoid data inconsistency. This also indicates that the migration is one-way only:
-once v2 is enabled, the user cannot go back to v1.
-
-## Next Steps
-
-### Full Migration
-
-This won't be implemented in the project's first phase but might be implemented in the future. Having both v1 and
-v2 existing in the system would force us to keep the code to deserialize the old reldir key forever. To entirely deprecate this
-code path, we must ensure the timeline has no old reldir data.
-
-We can trigger a special image layer generation process at the gc-horizon. The generated image layers will cover several keyspaces:
-the old reldir key in each of the databases, and the new reldir sparse keyspace. It will remove the old reldir key while
-copying them into the corresponding keys in the sparse keyspace in the resulting image. This special process happens in
-the background during compaction. For example, assume this special process is triggered at LSN 0/180. The `create_image_layers`
-process discovers the following keys at this LSN.
-
-```plain
-db1/reldir_key -> (table 1, table 2, table 3)
-...db1 rel keys
-db2/reldir_key -> (table 4, table 5, table 6)
-...db2 rel keys
-sparse_reldir_db2_table7 -> exists
-sparse_reldir_db1_table8 -> deleted
-```
-
-It will generate the following keys:
-
-```plain
-db1/reldir_key -> () # we have to keep the key because it is part of `collect_keyspace`.
-...db1 rel keys
-db2/reldir_key -> ()
-...db2 rel keys
-
--- start image layer for the sparse keyspace at sparse_reldir_prefix at LSN 0/180
-sparse_reldir_db1_table1 -> exists
-sparse_reldir_db1_table2 -> exists
-sparse_reldir_db1_table3 -> exists
-sparse_reldir_db2_table4 -> exists
-sparse_reldir_db2_table5 -> exists
-sparse_reldir_db2_table6 -> exists
-sparse_reldir_db2_table7 -> exists
--- end image layer for the sparse keyspace at sparse_reldir_prefix+1
-
-# The `sparse_reldir_db1_table8` key gets dropped as part of the image layer generation code for the sparse keyspace.
-# Note that the read path will stop reading if a key is not found in the image layer covering the key range so there
-# are no correctness issue.
-```
-
-We must verify that no pending modifications to the old reldir exists in the delta/image layers above the gc-horizon before
-we start this process (We can do a vectored read to get the full key history of the old reldir key and ensure there are no more images
-above the gc-horizon). Otherwise, it will violate the property that "a relation will only exists either in the old reldir key or
-in the new sparse keyspace". After we run this migration process, we can mark `reldir_v2_status` in the `index_part.json` to
-`Status::Migrated`, and the read path won't need to read from the old reldir anymore. Once the status is set to `Migrated`, we
-don't need to add the key into `collect_keyspace` and therefore all of them will be removed from all future image layers.
-
-The migration process can be proactively triggered across all attached/detached tenants to help us fully remove the old reldir code.
-
-### Consolidate Relation Size Keys
-
-We have relsize at the end of all relation nodes.
-
-```plain
-// RelSize:
-// 00 SPCNODE  DBNODE   RELNODE  FORK FFFFFFFF
-```
-
-This means that computing logical size requires us to do several single-key gets across the keyspace,
-potentially requiring downloading many layer files. We could consolidate them into a single
-keyspace, improving logical size calculation performance.
-
-### Migrate DBDir Keys
-
-We assume the number of databases created by the users will be small, and therefore, the current way
-of storing the database directory would be acceptable. In the future, we could also migrate DBDir keys into
-the sparse keyspace to support large amount of databases.

docs/storage_controller.md

@@ -101,7 +101,7 @@ changes such as a pageserver node becoming unavailable, or the tenant's shard co
 postgres clients to handle such changes, the storage controller calls an API hook when a tenant's pageserver
 location changes.
 
-The hook is configured using the storage controller's `--control-plane-url` CLI option. If the hook requires
+The hook is configured using the storage controller's `--compute-hook-url` CLI option. If the hook requires
 JWT auth, the token may be provided with `--control-plane-jwt-token`. The hook will be invoked with a `PUT` request.
 
 In the Neon cloud service, this hook is implemented by Neon's internal cloud control plane. In `neon_local` systems

libs/compute_api/src/responses.rs

@@ -134,10 +134,8 @@ pub struct CatalogObjects {
     pub databases: Vec<Database>,
 }
 
-#[derive(Clone, Debug, Deserialize, Serialize)]
+#[derive(Debug, Deserialize, Serialize)]
 pub struct ComputeCtlConfig {
-    /// Set of JSON web keys that the compute can use to authenticate
-    /// communication from the control plane.
     pub jwks: JwkSet,
 }
 

libs/compute_api/src/spec.rs

@@ -101,17 +101,6 @@ pub struct ComputeSpec {
     pub timeline_id: Option<TimelineId>,
     pub pageserver_connstring: 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.
-    /// 2) If walproposer wants to update list of safekeepers to connect to
-    ///    taking them from some safekeeper mconf, it should check what value
-    ///    is newer by comparing the generation.
-    ///
-    /// Note: it could be SafekeeperGeneration, but this needs linking
-    /// compute_ctl with postgres_ffi.
-    #[serde(default)]
-    pub safekeepers_generation: Option<u32>,
     #[serde(default)]
     pub safekeeper_connstrings: Vec<String>,
 
@@ -155,16 +144,6 @@ pub struct ComputeSpec {
     /// over the same replication content from publisher.
     #[serde(default)] // Default false
     pub drop_subscriptions_before_start: bool,
-
-    /// 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,
 }
 
 /// Feature flag to signal `compute_ctl` to enable certain experimental functionality.
@@ -272,17 +251,6 @@ pub enum ComputeMode {
     Replica,
 }
 
-/// Log level for audit logging
-/// Disabled, log, hipaa
-/// Default is Disabled
-#[derive(Clone, Debug, Default, Eq, PartialEq, Deserialize, Serialize)]
-pub enum ComputeAudit {
-    #[default]
-    Disabled,
-    Log,
-    Hipaa,
-}
-
 #[derive(Clone, Debug, Default, Deserialize, Serialize, PartialEq, Eq)]
 pub struct Cluster {
     pub cluster_id: Option<String>,

libs/consumption_metrics/Cargo.toml

@@ -1,7 +1,7 @@
 [package]
 name = "consumption_metrics"
 version = "0.1.0"
-edition = "2024"
+edition = "2021"
 license = "Apache-2.0"
 
 [dependencies]

libs/desim/src/chan.rs

@@ -1,5 +1,4 @@
-use std::collections::VecDeque;
-use std::sync::Arc;
+use std::{collections::VecDeque, sync::Arc};
 
 use parking_lot::{Mutex, MutexGuard};
 

libs/desim/src/executor.rs

@@ -1,7 +1,11 @@
-use std::panic::AssertUnwindSafe;
-use std::sync::atomic::{AtomicBool, AtomicU8, AtomicU32, Ordering};
-use std::sync::{Arc, OnceLock, mpsc};
-use std::thread::JoinHandle;
+use std::{
+    panic::AssertUnwindSafe,
+    sync::{
+        atomic::{AtomicBool, AtomicU32, AtomicU8, Ordering},
+        mpsc, Arc, OnceLock,
+    },
+    thread::JoinHandle,
+};
 
 use tracing::{debug, error, trace};
 

libs/desim/src/network.rs

@@ -1,19 +1,26 @@
-use std::cmp::Ordering;
-use std::collections::{BinaryHeap, VecDeque};
-use std::fmt::{self, Debug};
-use std::ops::DerefMut;
-use std::sync::{Arc, mpsc};
+use std::{
+    cmp::Ordering,
+    collections::{BinaryHeap, VecDeque},
+    fmt::{self, Debug},
+    ops::DerefMut,
+    sync::{mpsc, Arc},
+};
 
-use parking_lot::lock_api::{MappedMutexGuard, MutexGuard};
-use parking_lot::{Mutex, RawMutex};
+use parking_lot::{
+    lock_api::{MappedMutexGuard, MutexGuard},
+    Mutex, RawMutex,
+};
 use rand::rngs::StdRng;
 use tracing::debug;
 
-use super::chan::Chan;
-use super::proto::AnyMessage;
-use crate::executor::{self, ThreadContext};
-use crate::options::NetworkOptions;
-use crate::proto::{NetEvent, NodeEvent};
+use crate::{
+    executor::{self, ThreadContext},
+    options::NetworkOptions,
+    proto::NetEvent,
+    proto::NodeEvent,
+};
+
+use super::{chan::Chan, proto::AnyMessage};
 
 pub struct NetworkTask {
     options: Arc<NetworkOptions>,

libs/desim/src/node_os.rs

@@ -2,11 +2,14 @@ use std::sync::Arc;
 
 use rand::Rng;
 
-use super::chan::Chan;
-use super::network::TCP;
-use super::world::{Node, NodeId, World};
 use crate::proto::NodeEvent;
 
+use super::{
+    chan::Chan,
+    network::TCP,
+    world::{Node, NodeId, World},
+};
+
 /// Abstraction with all functions (aka syscalls) available to the node.
 #[derive(Clone)]
 pub struct NodeOs {

libs/desim/src/options.rs

@@ -1,5 +1,4 @@
-use rand::Rng;
-use rand::rngs::StdRng;
+use rand::{rngs::StdRng, Rng};
 
 /// Describes random delays and failures. Delay will be uniformly distributed in [min, max].
 /// Connection failure will occur with the probablity fail_prob.

libs/desim/src/proto.rs

@@ -3,8 +3,7 @@ use std::fmt::Debug;
 use bytes::Bytes;
 use utils::lsn::Lsn;
 
-use crate::network::TCP;
-use crate::world::NodeId;
+use crate::{network::TCP, world::NodeId};
 
 /// Internal node events.
 #[derive(Debug)]

libs/desim/src/time.rs

@@ -1,8 +1,12 @@
-use std::cmp::Ordering;
-use std::collections::BinaryHeap;
-use std::ops::DerefMut;
-use std::sync::Arc;
-use std::sync::atomic::{AtomicU32, AtomicU64};
+use std::{
+    cmp::Ordering,
+    collections::BinaryHeap,
+    ops::DerefMut,
+    sync::{
+        atomic::{AtomicU32, AtomicU64},
+        Arc,
+    },
+};
 
 use parking_lot::Mutex;
 use tracing::trace;

libs/desim/src/world.rs

@@ -1,18 +1,19 @@
-use std::ops::DerefMut;
-use std::sync::{Arc, mpsc};
-
 use parking_lot::Mutex;
-use rand::SeedableRng;
-use rand::rngs::StdRng;
+use rand::{rngs::StdRng, SeedableRng};
+use std::{
+    ops::DerefMut,
+    sync::{mpsc, Arc},
+};
 
-use super::chan::Chan;
-use super::network::TCP;
-use super::node_os::NodeOs;
-use crate::executor::{ExternalHandle, Runtime};
-use crate::network::NetworkTask;
-use crate::options::NetworkOptions;
-use crate::proto::{NodeEvent, SimEvent};
-use crate::time::Timing;
+use crate::{
+    executor::{ExternalHandle, Runtime},
+    network::NetworkTask,
+    options::NetworkOptions,
+    proto::{NodeEvent, SimEvent},
+    time::Timing,
+};
+
+use super::{chan::Chan, network::TCP, node_os::NodeOs};
 
 pub type NodeId = u32;
 

libs/desim/tests/reliable_copy_test.rs

@@ -1,15 +1,14 @@
 //! Simple test to verify that simulator is working.
 #[cfg(test)]
 mod reliable_copy_test {
-    use std::sync::Arc;
-
     use anyhow::Result;
     use desim::executor::{self, PollSome};
-    use desim::node_os::NodeOs;
     use desim::options::{Delay, NetworkOptions};
-    use desim::proto::{AnyMessage, NetEvent, NodeEvent, ReplCell};
+    use desim::proto::{NetEvent, NodeEvent, ReplCell};
     use desim::world::{NodeId, World};
+    use desim::{node_os::NodeOs, proto::AnyMessage};
     use parking_lot::Mutex;
+    use std::sync::Arc;
     use tracing::info;
 
     /// Disk storage trait and implementation.

libs/http-utils/Cargo.toml

@@ -6,9 +6,11 @@ license.workspace = true
 
 [dependencies]
 anyhow.workspace = true
+backtrace.workspace = true
 bytes.workspace = true
+inferno.workspace = true
 fail.workspace = true
-futures.workspace = true
+flate2.workspace = true
 hyper0.workspace = true
 itertools.workspace = true
 jemalloc_pprof.workspace = true
@@ -22,7 +24,6 @@ serde_path_to_error.workspace = true
 thiserror.workspace = true
 tracing.workspace = true
 tokio.workspace = true
-tokio-rustls.workspace = true
 tokio-util.workspace = true
 url.workspace = true
 uuid.workspace = true

libs/http-utils/src/endpoint.rs

@@ -1,28 +1,30 @@
+use crate::error::{api_error_handler, route_error_handler, ApiError};
+use crate::pprof;
+use crate::request::{get_query_param, parse_query_param};
+use ::pprof::protos::Message as _;
+use ::pprof::ProfilerGuardBuilder;
+use anyhow::{anyhow, Context};
+use bytes::{Bytes, BytesMut};
+use hyper::header::{HeaderName, AUTHORIZATION, CONTENT_DISPOSITION};
+use hyper::http::HeaderValue;
+use hyper::Method;
+use hyper::{header::CONTENT_TYPE, Body, Request, Response};
+use metrics::{register_int_counter, Encoder, IntCounter, TextEncoder};
+use once_cell::sync::Lazy;
+use regex::Regex;
+use routerify::ext::RequestExt;
+use routerify::{Middleware, RequestInfo, Router, RouterBuilder};
+use tokio::sync::{mpsc, Mutex, Notify};
+use tokio_stream::wrappers::ReceiverStream;
+use tokio_util::io::ReaderStream;
+use tracing::{debug, info, info_span, warn, Instrument};
+use utils::auth::{AuthError, Claims, SwappableJwtAuth};
+
 use std::future::Future;
 use std::io::Write as _;
 use std::str::FromStr;
 use std::time::Duration;
 
-use anyhow::{Context, anyhow};
-use bytes::{Bytes, BytesMut};
-use hyper::header::{AUTHORIZATION, CONTENT_DISPOSITION, CONTENT_TYPE, HeaderName};
-use hyper::http::HeaderValue;
-use hyper::{Body, Method, Request, Response};
-use metrics::{Encoder, IntCounter, TextEncoder, register_int_counter};
-use once_cell::sync::Lazy;
-use pprof::ProfilerGuardBuilder;
-use pprof::protos::Message as _;
-use routerify::ext::RequestExt;
-use routerify::{Middleware, RequestInfo, Router, RouterBuilder};
-use tokio::sync::{Mutex, Notify, mpsc};
-use tokio_stream::wrappers::ReceiverStream;
-use tokio_util::io::ReaderStream;
-use tracing::{Instrument, debug, info, info_span, warn};
-use utils::auth::{AuthError, Claims, SwappableJwtAuth};
-
-use crate::error::{ApiError, api_error_handler, route_error_handler};
-use crate::request::{get_query_param, parse_query_param};
-
 static SERVE_METRICS_COUNT: Lazy<IntCounter> = Lazy::new(|| {
     register_int_counter!(
         "libmetrics_metric_handler_requests_total",
@@ -373,7 +375,7 @@ pub async fn profile_cpu_handler(req: Request<Body>) -> Result<Response<Body>, A
                 Err(_) => {
                     return Err(ApiError::Conflict(
                         "profiler already running (use ?force=true to cancel it)".into(),
-                    ));
+                    ))
                 }
             }
             tokio::time::sleep(Duration::from_millis(1)).await; // don't busy-wait
@@ -399,10 +401,12 @@ pub async fn profile_cpu_handler(req: Request<Body>) -> Result<Response<Body>, A
     // Return the report in the requested format.
     match format {
         Format::Pprof => {
-            let body = report
+            let mut body = Vec::new();
+            report
                 .pprof()
                 .map_err(|err| ApiError::InternalServerError(err.into()))?
-                .encode_to_vec();
+                .write_to_vec(&mut body)
+                .map_err(|err| ApiError::InternalServerError(err.into()))?;
 
             Response::builder()
                 .status(200)
@@ -445,6 +449,20 @@ pub async fn profile_heap_handler(req: Request<Body>) -> Result<Response<Body>,
         Some(format) => return Err(ApiError::BadRequest(anyhow!("invalid format {format}"))),
     };
 
+    // Functions and mappings to strip when symbolizing pprof profiles. If true,
+    // also remove child frames.
+    static STRIP_FUNCTIONS: Lazy<Vec<(Regex, bool)>> = Lazy::new(|| {
+        vec![
+            (Regex::new("^__rust").unwrap(), false),
+            (Regex::new("^_start$").unwrap(), false),
+            (Regex::new("^irallocx_prof").unwrap(), true),
+            (Regex::new("^prof_alloc_prep").unwrap(), true),
+            (Regex::new("^std::rt::lang_start").unwrap(), false),
+            (Regex::new("^std::sys::backtrace::__rust").unwrap(), false),
+        ]
+    });
+    const STRIP_MAPPINGS: &[&str] = &["libc", "libgcc", "pthread", "vdso"];
+
     // Obtain profiler handle.
     let mut prof_ctl = jemalloc_pprof::PROF_CTL
         .as_ref()
@@ -477,34 +495,52 @@ pub async fn profile_heap_handler(req: Request<Body>) -> Result<Response<Body>,
         }
 
         Format::Pprof => {
-            let data = tokio::task::spawn_blocking(move || prof_ctl.dump_pprof())
-                .await
-                .map_err(|join_err| ApiError::InternalServerError(join_err.into()))?
-                .map_err(ApiError::InternalServerError)?;
+            let data = tokio::task::spawn_blocking(move || {
+                let bytes = prof_ctl.dump_pprof()?;
+                // Symbolize the profile.
+                // TODO: consider moving this upstream to jemalloc_pprof and avoiding the
+                // serialization roundtrip.
+                let profile = pprof::decode(&bytes)?;
+                let profile = pprof::symbolize(profile)?;
+                let profile = pprof::strip_locations(profile, STRIP_MAPPINGS, &STRIP_FUNCTIONS);
+                pprof::encode(&profile)
+            })
+            .await
+            .map_err(|join_err| ApiError::InternalServerError(join_err.into()))?
+            .map_err(ApiError::InternalServerError)?;
             Response::builder()
                 .status(200)
                 .header(CONTENT_TYPE, "application/octet-stream")
-                .header(CONTENT_DISPOSITION, "attachment; filename=\"heap.pb.gz\"")
+                .header(CONTENT_DISPOSITION, "attachment; filename=\"heap.pb\"")
                 .body(Body::from(data))
                 .map_err(|err| ApiError::InternalServerError(err.into()))
         }
 
         Format::Svg => {
-            let svg = tokio::task::spawn_blocking(move || prof_ctl.dump_flamegraph())
-                .await
-                .map_err(|join_err| ApiError::InternalServerError(join_err.into()))?
-                .map_err(ApiError::InternalServerError)?;
+            let body = tokio::task::spawn_blocking(move || {
+                let bytes = prof_ctl.dump_pprof()?;
+                let profile = pprof::decode(&bytes)?;
+                let profile = pprof::symbolize(profile)?;
+                let profile = pprof::strip_locations(profile, STRIP_MAPPINGS, &STRIP_FUNCTIONS);
+                let mut opts = inferno::flamegraph::Options::default();
+                opts.title = "Heap inuse".to_string();
+                opts.count_name = "bytes".to_string();
+                pprof::flamegraph(profile, &mut opts)
+            })
+            .await
+            .map_err(|join_err| ApiError::InternalServerError(join_err.into()))?
+            .map_err(ApiError::InternalServerError)?;
             Response::builder()
                 .status(200)
                 .header(CONTENT_TYPE, "image/svg+xml")
-                .body(Body::from(svg))
+                .body(Body::from(body))
                 .map_err(|err| ApiError::InternalServerError(err.into()))
         }
     }
 }
 
-pub fn add_request_id_middleware<B: hyper::body::HttpBody + Send + Sync + 'static>()
--> Middleware<B, ApiError> {
+pub fn add_request_id_middleware<B: hyper::body::HttpBody + Send + Sync + 'static>(
+) -> Middleware<B, ApiError> {
     Middleware::pre(move |req| async move {
         let request_id = match req.headers().get(&X_REQUEST_ID_HEADER) {
             Some(request_id) => request_id
@@ -628,7 +664,7 @@ pub fn auth_middleware<B: hyper::body::HttpBody + Send + Sync + 'static>(
                 None => {
                     return Err(ApiError::Unauthorized(
                         "missing authorization header".to_string(),
-                    ));
+                    ))
                 }
             }
         }
@@ -681,13 +717,11 @@ pub fn check_permission_with(
 
 #[cfg(test)]
 mod tests {
-    use std::future::poll_fn;
-    use std::net::{IpAddr, SocketAddr};
-
+    use super::*;
     use hyper::service::Service;
     use routerify::RequestServiceBuilder;
-
-    use super::*;
+    use std::future::poll_fn;
+    use std::net::{IpAddr, SocketAddr};
 
     #[tokio::test]
     async fn test_request_id_returned() {

libs/http-utils/src/error.rs

@@ -1,10 +1,10 @@
+use hyper::{header, Body, Response, StatusCode};
+use serde::{Deserialize, Serialize};
 use std::borrow::Cow;
 use std::error::Error as StdError;
-
-use hyper::{Body, Response, StatusCode, header};
-use serde::{Deserialize, Serialize};
 use thiserror::Error;
 use tracing::{error, info, warn};
+
 use utils::auth::AuthError;
 
 #[derive(Debug, Error)]

libs/http-utils/src/failpoints.rs

@@ -1,10 +1,11 @@
+use crate::error::ApiError;
+use crate::json::{json_request, json_response};
+
 use hyper::{Body, Request, Response, StatusCode};
 use serde::{Deserialize, Serialize};
 use tokio_util::sync::CancellationToken;
-use utils::failpoint_support::apply_failpoint;
 
-use crate::error::ApiError;
-use crate::json::{json_request, json_response};
+use utils::failpoint_support::apply_failpoint;
 
 pub type ConfigureFailpointsRequest = Vec<FailpointConfig>;
 

libs/http-utils/src/json.rs

@@ -1,6 +1,6 @@
 use anyhow::Context;
 use bytes::Buf;
-use hyper::{Body, Request, Response, StatusCode, header};
+use hyper::{header, Body, Request, Response, StatusCode};
 use serde::{Deserialize, Serialize};
 
 use super::error::ApiError;

libs/http-utils/src/lib.rs

@@ -2,11 +2,11 @@ pub mod endpoint;
 pub mod error;
 pub mod failpoints;
 pub mod json;
+pub mod pprof;
 pub mod request;
-pub mod server;
 
 extern crate hyper0 as hyper;
 
 /// Current fast way to apply simple http routing in various Neon binaries.
 /// Re-exported for sake of uniform approach, that could be later replaced with better alternatives, if needed.
-pub use routerify::{RequestServiceBuilder, RouterBuilder, RouterService, ext::RequestExt};
+pub use routerify::{ext::RequestExt, RouterBuilder, RouterService};

libs/http-utils/src/pprof.rs

@@ -0,0 +1,238 @@
+use anyhow::bail;
+use flate2::write::{GzDecoder, GzEncoder};
+use flate2::Compression;
+use itertools::Itertools as _;
+use pprof::protos::{Function, Line, Location, Message as _, Profile};
+use regex::Regex;
+
+use std::borrow::Cow;
+use std::collections::{HashMap, HashSet};
+use std::ffi::c_void;
+use std::io::Write as _;
+
+/// Decodes a gzip-compressed Protobuf-encoded pprof profile.
+pub fn decode(bytes: &[u8]) -> anyhow::Result<Profile> {
+    let mut gz = GzDecoder::new(Vec::new());
+    gz.write_all(bytes)?;
+    Ok(Profile::parse_from_bytes(&gz.finish()?)?)
+}
+
+/// Encodes a pprof profile as gzip-compressed Protobuf.
+pub fn encode(profile: &Profile) -> anyhow::Result<Vec<u8>> {
+    let mut gz = GzEncoder::new(Vec::new(), Compression::default());
+    profile.write_to_writer(&mut gz)?;
+    Ok(gz.finish()?)
+}
+
+/// Symbolizes a pprof profile using the current binary.
+pub fn symbolize(mut profile: Profile) -> anyhow::Result<Profile> {
+    if !profile.function.is_empty() {
+        return Ok(profile); // already symbolized
+    }
+
+    // Collect function names.
+    let mut functions: HashMap<String, Function> = HashMap::new();
+    let mut strings: HashMap<String, i64> = profile
+        .string_table
+        .into_iter()
+        .enumerate()
+        .map(|(i, s)| (s, i as i64))
+        .collect();
+
+    // Helper to look up or register a string.
+    let mut string_id = |s: &str| -> i64 {
+        // Don't use .entry() to avoid unnecessary allocations.
+        if let Some(id) = strings.get(s) {
+            return *id;
+        }
+        let id = strings.len() as i64;
+        strings.insert(s.to_string(), id);
+        id
+    };
+
+    for loc in &mut profile.location {
+        if !loc.line.is_empty() {
+            continue;
+        }
+
+        // Resolve the line and function for each location.
+        backtrace::resolve(loc.address as *mut c_void, |symbol| {
+            let Some(symbol_name) = symbol.name() else {
+                return;
+            };
+
+            let function_name = format!("{symbol_name:#}");
+            let functions_len = functions.len();
+            let function_id = functions
+                .entry(function_name)
+                .or_insert_with_key(|function_name| {
+                    let function_id = functions_len as u64 + 1;
+                    let system_name = String::from_utf8_lossy(symbol_name.as_bytes());
+                    let filename = symbol
+                        .filename()
+                        .map(|path| path.to_string_lossy())
+                        .unwrap_or(Cow::Borrowed(""));
+                    Function {
+                        id: function_id,
+                        name: string_id(function_name),
+                        system_name: string_id(&system_name),
+                        filename: string_id(&filename),
+                        ..Default::default()
+                    }
+                })
+                .id;
+            loc.line.push(Line {
+                function_id,
+                line: symbol.lineno().unwrap_or(0) as i64,
+                ..Default::default()
+            });
+        });
+    }
+
+    // Store the resolved functions, and mark the mapping as resolved.
+    profile.function = functions.into_values().sorted_by_key(|f| f.id).collect();
+    profile.string_table = strings
+        .into_iter()
+        .sorted_by_key(|(_, i)| *i)
+        .map(|(s, _)| s)
+        .collect();
+
+    for mapping in &mut profile.mapping {
+        mapping.has_functions = true;
+        mapping.has_filenames = true;
+    }
+
+    Ok(profile)
+}
+
+/// Strips locations (stack frames) matching the given mappings (substring) or function names
+/// (regex). The function bool specifies whether child frames should be stripped as well.
+///
+/// The string definitions are left behind in the profile for simplicity, to avoid rewriting all
+/// string references.
+pub fn strip_locations(
+    mut profile: Profile,
+    mappings: &[&str],
+    functions: &[(Regex, bool)],
+) -> Profile {
+    // Strip mappings.
+    let mut strip_mappings: HashSet<u64> = HashSet::new();
+
+    profile.mapping.retain(|mapping| {
+        let Some(name) = profile.string_table.get(mapping.filename as usize) else {
+            return true;
+        };
+        if mappings.iter().any(|substr| name.contains(substr)) {
+            strip_mappings.insert(mapping.id);
+            return false;
+        }
+        true
+    });
+
+    // Strip functions.
+    let mut strip_functions: HashMap<u64, bool> = HashMap::new();
+
+    profile.function.retain(|function| {
+        let Some(name) = profile.string_table.get(function.name as usize) else {
+            return true;
+        };
+        for (regex, strip_children) in functions {
+            if regex.is_match(name) {
+                strip_functions.insert(function.id, *strip_children);
+                return false;
+            }
+        }
+        true
+    });
+
+    // Strip locations. The bool specifies whether child frames should be stripped too.
+    let mut strip_locations: HashMap<u64, bool> = HashMap::new();
+
+    profile.location.retain(|location| {
+        for line in &location.line {
+            if let Some(strip_children) = strip_functions.get(&line.function_id) {
+                strip_locations.insert(location.id, *strip_children);
+                return false;
+            }
+        }
+        if strip_mappings.contains(&location.mapping_id) {
+            strip_locations.insert(location.id, false);
+            return false;
+        }
+        true
+    });
+
+    // Strip sample locations.
+    for sample in &mut profile.sample {
+        // First, find the uppermost function with child removal and truncate the stack.
+        if let Some(truncate) = sample
+            .location_id
+            .iter()
+            .rposition(|id| strip_locations.get(id) == Some(&true))
+        {
+            sample.location_id.drain(..=truncate);
+        }
+        // Next, strip any individual frames without child removal.
+        sample
+            .location_id
+            .retain(|id| !strip_locations.contains_key(id));
+    }
+
+    profile
+}
+
+/// Generates an SVG flamegraph from a symbolized pprof profile.
+pub fn flamegraph(
+    profile: Profile,
+    opts: &mut inferno::flamegraph::Options,
+) -> anyhow::Result<Vec<u8>> {
+    if profile.mapping.iter().any(|m| !m.has_functions) {
+        bail!("profile not symbolized");
+    }
+
+    // Index locations, functions, and strings.
+    let locations: HashMap<u64, Location> =
+        profile.location.into_iter().map(|l| (l.id, l)).collect();
+    let functions: HashMap<u64, Function> =
+        profile.function.into_iter().map(|f| (f.id, f)).collect();
+    let strings = profile.string_table;
+
+    // Resolve stacks as function names, and sum sample values per stack. Also reverse the stack,
+    // since inferno expects it bottom-up.
+    let mut stacks: HashMap<Vec<&str>, i64> = HashMap::new();
+    for sample in profile.sample {
+        let mut stack = Vec::with_capacity(sample.location_id.len());
+        for location in sample.location_id.into_iter().rev() {
+            let Some(location) = locations.get(&location) else {
+                bail!("missing location {location}");
+            };
+            for line in location.line.iter().rev() {
+                let Some(function) = functions.get(&line.function_id) else {
+                    bail!("missing function {}", line.function_id);
+                };
+                let Some(name) = strings.get(function.name as usize) else {
+                    bail!("missing string {}", function.name);
+                };
+                stack.push(name.as_str());
+            }
+        }
+        let Some(&value) = sample.value.first() else {
+            bail!("missing value");
+        };
+        *stacks.entry(stack).or_default() += value;
+    }
+
+    // Construct stack lines for inferno.
+    let lines = stacks
+        .into_iter()
+        .map(|(stack, value)| (stack.into_iter().join(";"), value))
+        .map(|(stack, value)| format!("{stack} {value}"))
+        .sorted()
+        .collect_vec();
+
+    // Construct the flamegraph.
+    let mut bytes = Vec::new();
+    let lines = lines.iter().map(|line| line.as_str());
+    inferno::flamegraph::from_lines(opts, lines, &mut bytes)?;
+    Ok(bytes)
+}

libs/http-utils/src/request.rs

@@ -1,13 +1,10 @@
 use core::fmt;
-use std::borrow::Cow;
-use std::str::FromStr;
-
-use anyhow::anyhow;
-use hyper::body::HttpBody;
-use hyper::{Body, Request};
-use routerify::ext::RequestExt;
+use std::{borrow::Cow, str::FromStr};
 
 use super::error::ApiError;
+use anyhow::anyhow;
+use hyper::{body::HttpBody, Body, Request};
+use routerify::ext::RequestExt;
 
 pub fn get_request_param<'a>(
     request: &'a Request<Body>,

libs/http-utils/src/server.rs

@@ -1,155 +0,0 @@
-use std::{error::Error, sync::Arc};
-
-use futures::StreamExt;
-use futures::stream::FuturesUnordered;
-use hyper0::Body;
-use hyper0::server::conn::Http;
-use routerify::{RequestService, RequestServiceBuilder};
-use tokio::io::{AsyncRead, AsyncWrite};
-use tokio_rustls::TlsAcceptor;
-use tokio_util::sync::CancellationToken;
-use tracing::{error, info};
-
-use crate::error::ApiError;
-
-/// A simple HTTP server over hyper library.
-/// You may want to use it instead of [`hyper0::server::Server`] because:
-/// 1. hyper0's Server was removed from hyper v1.
-///    It's recommended to replace hyepr0's Server with a manual loop, which is done here.
-/// 2. hyper0's Server doesn't support TLS out of the box, and there is no way
-///    to support it efficiently with the Accept trait that hyper0's Server uses.
-///    That's one of the reasons why it was removed from v1.
-///    <https://github.com/hyperium/hyper/blob/115339d3df50f20c8717680aa35f48858e9a6205/docs/ROADMAP.md#higher-level-client-and-server-problems>
-pub struct Server {
-    request_service: Arc<RequestServiceBuilder<Body, ApiError>>,
-    listener: tokio::net::TcpListener,
-    tls_acceptor: Option<TlsAcceptor>,
-}
-
-impl Server {
-    pub fn new(
-        request_service: Arc<RequestServiceBuilder<Body, ApiError>>,
-        listener: std::net::TcpListener,
-        tls_acceptor: Option<TlsAcceptor>,
-    ) -> anyhow::Result<Self> {
-        // Note: caller of from_std is responsible for setting nonblocking mode.
-        listener.set_nonblocking(true)?;
-        let listener = tokio::net::TcpListener::from_std(listener)?;
-
-        Ok(Self {
-            request_service,
-            listener,
-            tls_acceptor,
-        })
-    }
-
-    pub async fn serve(self, cancel: CancellationToken) -> anyhow::Result<()> {
-        fn suppress_io_error(err: &std::io::Error) -> bool {
-            use std::io::ErrorKind::*;
-            matches!(err.kind(), ConnectionReset | ConnectionAborted | BrokenPipe)
-        }
-        fn suppress_hyper_error(err: &hyper0::Error) -> bool {
-            if err.is_incomplete_message() || err.is_closed() || err.is_timeout() {
-                return true;
-            }
-            if let Some(inner) = err.source() {
-                if let Some(io) = inner.downcast_ref::<std::io::Error>() {
-                    return suppress_io_error(io);
-                }
-            }
-            false
-        }
-
-        let mut connections = FuturesUnordered::new();
-        loop {
-            tokio::select! {
-                stream = self.listener.accept() => {
-                    let (tcp_stream, remote_addr) = match stream {
-                        Ok(stream) => stream,
-                        Err(err) => {
-                            if !suppress_io_error(&err) {
-                                info!("Failed to accept TCP connection: {err:#}");
-                            }
-                            continue;
-                        }
-                    };
-
-                    let service = self.request_service.build(remote_addr);
-                    let tls_acceptor = self.tls_acceptor.clone();
-                    let cancel = cancel.clone();
-
-                    connections.push(tokio::spawn(
-                        async move {
-                            match tls_acceptor {
-                                Some(tls_acceptor) => {
-                                    // Handle HTTPS connection.
-                                    let tls_stream = tokio::select! {
-                                        tls_stream = tls_acceptor.accept(tcp_stream) => tls_stream,
-                                        _ = cancel.cancelled() => return,
-                                    };
-                                    let tls_stream = match tls_stream {
-                                        Ok(tls_stream) => tls_stream,
-                                        Err(err) => {
-                                            if !suppress_io_error(&err) {
-                                                info!("Failed to accept TLS connection: {err:#}");
-                                            }
-                                            return;
-                                        }
-                                    };
-                                    if let Err(err) = Self::serve_connection(tls_stream, service, cancel).await {
-                                        if !suppress_hyper_error(&err) {
-                                            info!("Failed to serve HTTPS connection: {err:#}");
-                                        }
-                                    }
-                                }
-                                None => {
-                                    // Handle HTTP connection.
-                                    if let Err(err) = Self::serve_connection(tcp_stream, service, cancel).await {
-                                        if !suppress_hyper_error(&err) {
-                                            info!("Failed to serve HTTP connection: {err:#}");
-                                        }
-                                    }
-                                }
-                            };
-                        }));
-                 }
-                Some(conn) = connections.next() => {
-                    if let Err(err) = conn {
-                        error!("Connection panicked: {err:#}");
-                    }
-                }
-                _ = cancel.cancelled() => {
-                    // Wait for graceful shutdown of all connections.
-                    while let Some(conn) = connections.next().await {
-                        if let Err(err) = conn {
-                            error!("Connection panicked: {err:#}");
-                        }
-                    }
-                    break;
-                }
-            }
-        }
-        Ok(())
-    }
-
-    /// Serves HTTP connection with graceful shutdown.
-    async fn serve_connection<I>(
-        io: I,
-        service: RequestService<Body, ApiError>,
-        cancel: CancellationToken,
-    ) -> Result<(), hyper0::Error>
-    where
-        I: AsyncRead + AsyncWrite + Unpin + Send + 'static,
-    {
-        let mut conn = Http::new().serve_connection(io, service).with_upgrades();
-
-        tokio::select! {
-            res = &mut conn => res,
-            _ = cancel.cancelled() => {
-                Pin::new(&mut conn).graceful_shutdown();
-                // Note: connection should still be awaited for graceful shutdown to complete.
-                conn.await
-            }
-        }
-    }
-}

libs/metrics/src/hll.rs

@@ -6,15 +6,17 @@
 //! Probabilistic cardinality estimators, such as the HyperLogLog algorithm,
 //! use significantly less memory than this, but can only approximate the cardinality.
 
-use std::hash::{BuildHasher, BuildHasherDefault, Hash};
-use std::sync::atomic::AtomicU8;
+use std::{
+    hash::{BuildHasher, BuildHasherDefault, Hash},
+    sync::atomic::AtomicU8,
+};
 
-use measured::LabelGroup;
-use measured::label::{LabelGroupVisitor, LabelName, LabelValue, LabelVisitor};
-use measured::metric::counter::CounterState;
-use measured::metric::name::MetricNameEncoder;
-use measured::metric::{Metric, MetricType, MetricVec};
-use measured::text::TextEncoder;
+use measured::{
+    label::{LabelGroupVisitor, LabelName, LabelValue, LabelVisitor},
+    metric::{counter::CounterState, name::MetricNameEncoder, Metric, MetricType, MetricVec},
+    text::TextEncoder,
+    LabelGroup,
+};
 use twox_hash::xxh3;
 
 /// Create an [`HyperLogLogVec`] and registers to default registry.
@@ -25,7 +27,9 @@ macro_rules! register_hll_vec {
         $crate::register(Box::new(hll_vec.clone())).map(|_| hll_vec)
     }};
 
-    ($N:literal, $NAME:expr, $HELP:expr, $LABELS_NAMES:expr $(,)?) => {{ $crate::register_hll_vec!($N, $crate::opts!($NAME, $HELP), $LABELS_NAMES) }};
+    ($N:literal, $NAME:expr, $HELP:expr, $LABELS_NAMES:expr $(,)?) => {{
+        $crate::register_hll_vec!($N, $crate::opts!($NAME, $HELP), $LABELS_NAMES)
+    }};
 }
 
 /// Create an [`HyperLogLog`] and registers to default registry.
@@ -36,7 +40,9 @@ macro_rules! register_hll {
         $crate::register(Box::new(hll.clone())).map(|_| hll)
     }};
 
-    ($N:literal, $NAME:expr, $HELP:expr $(,)?) => {{ $crate::register_hll!($N, $crate::opts!($NAME, $HELP)) }};
+    ($N:literal, $NAME:expr, $HELP:expr $(,)?) => {{
+        $crate::register_hll!($N, $crate::opts!($NAME, $HELP))
+    }};
 }
 
 /// HLL is a probabilistic cardinality measure.
@@ -189,10 +195,8 @@ impl<W: std::io::Write, const N: usize> measured::metric::MetricEncoding<TextEnc
 mod tests {
     use std::collections::HashSet;
 
-    use measured::FixedCardinalityLabel;
-    use measured::label::StaticLabelSet;
-    use rand::rngs::StdRng;
-    use rand::{Rng, SeedableRng};
+    use measured::{label::StaticLabelSet, FixedCardinalityLabel};
+    use rand::{rngs::StdRng, Rng, SeedableRng};
     use rand_distr::{Distribution, Zipf};
 
     use crate::HyperLogLogVec;