Simplify SNI parsing

More logging
Finally, build TLS config in proxy's main
2026-07-07 22:20:36 +00:00 · 2023-04-05 13:03:16 +03:00 · 2023-04-04 21:14:30 +03:00 · 2023-04-04 21:02:23 +03:00 · 2023-04-04 19:58:56 +03:00 · 2023-04-03 22:22:48 +03:00
79 changed files with 1982 additions and 2186 deletions
--- a/.github/actions/allure-report/action.yml
+++ b/.github/actions/allure-report/action.yml
@@ -15,32 +15,10 @@ outputs:
  report-url:
    description: 'Allure report URL'
    value: ${{ steps.generate-report.outputs.report-url }}
-  report-json-url:
-    description: 'Allure report JSON URL'
-    value: ${{ steps.generate-report.outputs.report-json-url }}

 runs:
  using: "composite"
-
  steps:
-    # We're using some of env variables quite offen, so let's set them once.
-    #
-    # It would be nice to have them set in common runs.env[0] section, but it doesn't work[1]
-    #
-    # - [0] https://docs.github.com/en/actions/creating-actions/metadata-syntax-for-github-actions#runsenv
-    # - [1] https://github.com/neondatabase/neon/pull/3907#discussion_r1154703456
-    #
-    - name: Set common environment variables
-      shell: bash -euxo pipefail {0}
-      run: |
-        echo "BUILD_TYPE=${BUILD_TYPE}"   >> $GITHUB_ENV
-        echo "BUCKET=${BUCKET}"           >> $GITHUB_ENV
-        echo "TEST_OUTPUT=${TEST_OUTPUT}" >> $GITHUB_ENV
-      env:
-        BUILD_TYPE: ${{ inputs.build_type }}
-        BUCKET: neon-github-public-dev
-        TEST_OUTPUT: /tmp/test_output
-
    - name: Validate input parameters
      shell: bash -euxo pipefail {0}
      run: |
@@ -98,14 +76,16 @@ runs:
          rm -f ${ALLURE_ZIP}
        fi
      env:
-        ALLURE_VERSION: 2.21.0
-        ALLURE_ZIP_MD5: c8db4dd8e2a7882583d569ed2c82879c
+        ALLURE_VERSION: 2.19.0
+        ALLURE_ZIP_MD5: ced21401a1a8b9dfb68cee9e4c210464

    - name: Upload Allure results
      if: ${{ inputs.action == 'store' }}
      env:
        REPORT_PREFIX: reports/${{ steps.calculate-vars.outputs.KEY }}/${{ inputs.build_type }}
        RAW_PREFIX: reports-raw/${{ steps.calculate-vars.outputs.KEY }}/${{ inputs.build_type }}
+        TEST_OUTPUT: /tmp/test_output
+        BUCKET: neon-github-public-dev
        TEST_SELECTION: ${{ steps.calculate-vars.outputs.TEST_SELECTION }}
      shell: bash -euxo pipefail {0}
      run: |
@@ -124,7 +104,7 @@ runs:
        EOF
        cat <<EOF > $TEST_OUTPUT/allure/results/environment.properties
          TEST_SELECTION=${{ inputs.test_selection }}
-          BUILD_TYPE=${BUILD_TYPE}
+          BUILD_TYPE=${{ inputs.build_type }}
        EOF

        ARCHIVE="${GITHUB_RUN_ID}-${TEST_SELECTION}-${GITHUB_RUN_ATTEMPT}-$(date +%s).tar.zst"
@@ -133,12 +113,13 @@ runs:
        tar -C ${TEST_OUTPUT}/allure/results -cf ${ARCHIVE} --zstd .
        aws s3 mv --only-show-errors ${ARCHIVE} "s3://${BUCKET}/${RAW_PREFIX}/${ARCHIVE}"

-    # Potentially we could have several running build for the same key (for example for the main branch), so we use improvised lock for this
+    # Potentially we could have several running build for the same key (for example for the main branch),  so we use improvised lock for this
    - name: Acquire Allure lock
      if: ${{ inputs.action == 'generate' }}
      shell: bash -euxo pipefail {0}
      env:
        LOCK_FILE: reports/${{ steps.calculate-vars.outputs.KEY }}/lock.txt
+        BUCKET: neon-github-public-dev
        TEST_SELECTION: ${{ steps.calculate-vars.outputs.TEST_SELECTION }}
      run: |
        LOCK_TIMEOUT=300 # seconds
@@ -168,6 +149,8 @@ runs:
      env:
        REPORT_PREFIX: reports/${{ steps.calculate-vars.outputs.KEY }}/${{ inputs.build_type }}
        RAW_PREFIX: reports-raw/${{ steps.calculate-vars.outputs.KEY }}/${{ inputs.build_type }}
+        TEST_OUTPUT: /tmp/test_output
+        BUCKET: neon-github-public-dev
      shell: bash -euxo pipefail {0}
      run: |
        # Get previously uploaded data for this run
@@ -203,24 +186,24 @@ runs:
        REPORT_URL=https://${BUCKET}.s3.amazonaws.com/${REPORT_PREFIX}/${GITHUB_RUN_ID}/index.html

        # Generate redirect
-        cat <<EOF > ${TEST_OUTPUT}/allure/index.html
+        cat <<EOF > ./index.html
          <!DOCTYPE html>

          <meta charset="utf-8">
          <title>Redirecting to ${REPORT_URL}</title>
          <meta http-equiv="refresh" content="0; URL=${REPORT_URL}">
        EOF
-        aws s3 cp --only-show-errors ${TEST_OUTPUT}/allure/index.html "s3://${BUCKET}/${REPORT_PREFIX}/latest/index.html"
+        aws s3 cp --only-show-errors ./index.html "s3://${BUCKET}/${REPORT_PREFIX}/latest/index.html"

        echo "[Allure Report](${REPORT_URL})" >> ${GITHUB_STEP_SUMMARY}
        echo "report-url=${REPORT_URL}" >> $GITHUB_OUTPUT
-        echo "report-json-url=${REPORT_URL%/index.html}/data/suites.json" >> $GITHUB_OUTPUT

    - name: Release Allure lock
      if: ${{ inputs.action == 'generate' && always() }}
      shell: bash -euxo pipefail {0}
      env:
        LOCK_FILE: reports/${{ steps.calculate-vars.outputs.KEY }}/lock.txt
+        BUCKET: neon-github-public-dev
        TEST_SELECTION: ${{ steps.calculate-vars.outputs.TEST_SELECTION }}
      run: |
        aws s3 cp --only-show-errors "s3://${BUCKET}/${LOCK_FILE}" ./lock.txt || exit 0
@@ -229,16 +212,11 @@ runs:
          aws s3 rm "s3://${BUCKET}/${LOCK_FILE}"
        fi

-    - name: Cleanup
-      if: always()
-      shell: bash -euxo pipefail {0}
-      run: |
-        rm -rf ${TEST_OUTPUT}/allure
-
    - uses: actions/github-script@v6
      if: ${{ inputs.action == 'generate' && always() }}
      env:
        REPORT_URL: ${{ steps.generate-report.outputs.report-url }}
+        BUILD_TYPE: ${{ inputs.build_type }}
        SHA: ${{ github.event.pull_request.head.sha || github.sha }}
      with:
        script: |
--- a/.github/actions/neon-project-create/action.yml
+++ b/.github/actions/neon-project-create/action.yml
@@ -14,12 +14,6 @@ inputs:
  api_host:
    desctiption: 'Neon API host'
    default: console.stage.neon.tech
-  provisioner:
-    desctiption: 'k8s-pod or k8s-neonvm'
-    default: 'k8s-pod'
-  compute_units:
-    desctiption: '[Min, Max] compute units; Min and Max are used for k8s-neonvm with autoscaling, for k8s-pod values Min and Max should be equal'
-    default: '[1, 1]'

 outputs:
  dsn:
@@ -37,10 +31,6 @@ runs:
      # A shell without `set -x` to not to expose password/dsn in logs
      shell: bash -euo pipefail {0}
      run: |
-        if [ "${PROVISIONER}" == "k8s-pod" ] && [ "${MIN_CU}" != "${MAX_CU}" ]; then
-          echo >&2 "For k8s-pod provisioner MIN_CU should be equal to MAX_CU"
-        fi
-
        project=$(curl \
          "https://${API_HOST}/api/v2/projects" \
          --fail \
@@ -52,9 +42,6 @@ runs:
              \"name\": \"Created by actions/neon-project-create; GITHUB_RUN_ID=${GITHUB_RUN_ID}\",
              \"pg_version\": ${POSTGRES_VERSION},
              \"region_id\": \"${REGION_ID}\",
-              \"provisioner\": \"${PROVISIONER}\",
-              \"autoscaling_limit_min_cu\": ${MIN_CU},
-              \"autoscaling_limit_max_cu\": ${MAX_CU},
              \"settings\": { }
            }
          }")
@@ -75,6 +62,3 @@ runs:
        API_KEY: ${{ inputs.api_key }}
        REGION_ID: ${{ inputs.region_id }}
        POSTGRES_VERSION: ${{ inputs.postgres_version }}
-        PROVISIONER: ${{ inputs.provisioner }}
-        MIN_CU: ${{ fromJSON(inputs.compute_units)[0] }}
-        MAX_CU: ${{ fromJSON(inputs.compute_units)[1] }}
--- a/.github/actions/run-python-test-set/action.yml
+++ b/.github/actions/run-python-test-set/action.yml
@@ -44,10 +44,6 @@ inputs:
    description: 'Secret access key'
    required: false
    default: ''
-  rerun_flaky:
-    description: 'Whether to rerun flaky tests'
-    required: false
-    default: 'false'

 runs:
  using: "composite"
@@ -105,7 +101,6 @@ runs:
        COMPATIBILITY_SNAPSHOT_DIR: /tmp/compatibility_snapshot_pg14
        ALLOW_BACKWARD_COMPATIBILITY_BREAKAGE: contains(github.event.pull_request.labels.*.name, 'backward compatibility breakage')
        ALLOW_FORWARD_COMPATIBILITY_BREAKAGE: contains(github.event.pull_request.labels.*.name, 'forward compatibility breakage')
-        RERUN_FLAKY: ${{ inputs.rerun_flaky }}
      shell: bash -euxo pipefail {0}
      run: |
        # PLATFORM will be embedded in the perf test report
@@ -148,13 +143,6 @@ runs:
          EXTRA_PARAMS="--out-dir $PERF_REPORT_DIR $EXTRA_PARAMS"
        fi

-        if [ "${RERUN_FLAKY}" == "true" ]; then
-          mkdir -p $TEST_OUTPUT
-          poetry run ./scripts/flaky_tests.py "${TEST_RESULT_CONNSTR}" --days 10 --output "$TEST_OUTPUT/flaky.json"
-
-          EXTRA_PARAMS="--flaky-tests-json $TEST_OUTPUT/flaky.json $EXTRA_PARAMS"
-        fi
-
        if [[ "${{ inputs.build_type }}" == "debug" ]]; then
          cov_prefix=(scripts/coverage "--profraw-prefix=$GITHUB_JOB" --dir=/tmp/coverage run)
        elif [[ "${{ inputs.build_type }}" == "release" ]]; then
--- a/.github/ansible/scripts/init_pageserver.sh
+++ b/.github/ansible/scripts/init_pageserver.sh
@@ -3,8 +3,6 @@
 # fetch params from meta-data service
 INSTANCE_ID=$(curl -s http://169.254.169.254/latest/meta-data/instance-id)
 AZ_ID=$(curl -s http://169.254.169.254/latest/meta-data/placement/availability-zone)
-INSTANCE_TYPE=$(curl -s http://169.254.169.254/latest/meta-data/instance-type)
-DISK_SIZE=$(df -B1 /storage | tail -1 | awk '{print $2}')

 # store fqdn hostname in var
 HOST=$(hostname -f)
@@ -20,9 +18,7 @@ cat <<EOF | tee /tmp/payload
  "http_host": "${HOST}",
  "http_port": 9898,
  "active": false,
-  "availability_zone_id": "${AZ_ID}",
-  "disk_size": ${DISK_SIZE},
-  "instance_type": "${INSTANCE_TYPE}"
+  "availability_zone_id": "${AZ_ID}"
 }
 EOF

--- a/.github/helm-values/dev-us-east-2-beta.neon-proxy-scram.yaml
+++ b/.github/helm-values/dev-us-east-2-beta.neon-proxy-scram.yaml
@@ -23,7 +23,6 @@ settings:
  authBackend: "console"
  authEndpoint: "http://neon-internal-api.aws.neon.build/management/api/v2"
  domain: "*.us-east-2.aws.neon.build"
-  extraDomains: ["*.us-east-2.postgres.zenith.tech", "*.us-east-2.retooldb-staging.com"]
  sentryEnvironment: "staging"
  wssPort: 8443
  metricCollectionEndpoint: "http://neon-internal-api.aws.neon.build/billing/api/v1/usage_events"
--- a/.github/workflows/benchmarking.yml
+++ b/.github/workflows/benchmarking.yml
@@ -111,7 +111,6 @@ jobs:
    strategy:
      fail-fast: false
      matrix:
-        # neon-captest-freetier: Run pgbench with freetier-limited compute
        # neon-captest-new: Run pgbench in a freshly created project
        # neon-captest-reuse: Same, but reusing existing project
        # neon-captest-prefetch: Same, with prefetching enabled (new project)
@@ -121,9 +120,6 @@ jobs:
        db_size: [ 10gb ]
        runner: [ us-east-2 ]
        include:
-          - platform: neon-captest-freetier
-            db_size: 3gb
-            runner: us-east-2
          - platform: neon-captest-prefetch
            db_size: 50gb
            runner: us-east-2
@@ -164,14 +160,13 @@ jobs:
        echo "${POSTGRES_DISTRIB_DIR}/v${DEFAULT_PG_VERSION}/bin" >> $GITHUB_PATH

    - name: Create Neon Project
-      if: contains(fromJson('["neon-captest-new", "neon-captest-prefetch", "neon-captest-freetier"]'), matrix.platform)
+      if: contains(fromJson('["neon-captest-new", "neon-captest-prefetch"]'), matrix.platform)
      id: create-neon-project
      uses: ./.github/actions/neon-project-create
      with:
        region_id: ${{ github.event.inputs.region_id || 'aws-us-east-2' }}
        postgres_version: ${{ env.DEFAULT_PG_VERSION }}
        api_key: ${{ secrets.NEON_STAGING_API_KEY }}
-        compute_units: ${{ (matrix.platform == 'neon-captest-freetier' && '[0.25, 0.25]') || '[1, 1]' }}

    - name: Set up Connection String
      id: set-up-connstr
@@ -180,7 +175,7 @@ jobs:
          neon-captest-reuse)
            CONNSTR=${{ secrets.BENCHMARK_CAPTEST_CONNSTR }}
            ;;
-          neon-captest-new | neon-captest-prefetch | neon-captest-freetier)
+          neon-captest-new | neon-captest-prefetch)
            CONNSTR=${{ steps.create-neon-project.outputs.dsn }}
            ;;
          rds-aurora)
@@ -190,7 +185,7 @@ jobs:
            CONNSTR=${{ secrets.BENCHMARK_RDS_POSTGRES_CONNSTR }}
            ;;
          *)
-            echo 2>&1 "Unknown PLATFORM=${PLATFORM}. Allowed only 'neon-captest-reuse', 'neon-captest-new', 'neon-captest-prefetch', neon-captest-freetier, 'rds-aurora', or 'rds-postgres'"
+            echo 2>&1 "Unknown PLATFORM=${PLATFORM}. Allowed only 'neon-captest-reuse', 'neon-captest-new', 'neon-captest-prefetch', 'rds-aurora', or 'rds-postgres'"
            exit 1
            ;;
        esac
--- a/.github/workflows/build_and_test.yml
+++ b/.github/workflows/build_and_test.yml
@@ -335,10 +335,6 @@ jobs:
          real_s3_region: us-west-2
          real_s3_access_key_id: "${{ secrets.AWS_ACCESS_KEY_ID_CI_TESTS_S3 }}"
          real_s3_secret_access_key: "${{ secrets.AWS_SECRET_ACCESS_KEY_CI_TESTS_S3 }}"
-          rerun_flaky: true
-        env:
-          TEST_RESULT_CONNSTR: ${{ secrets.REGRESS_TEST_RESULT_CONNSTR }}
-          CHECK_ONDISK_DATA_COMPATIBILITY: nonempty

      - name: Merge and upload coverage data
        if: matrix.build_type == 'debug'
@@ -375,90 +371,42 @@ jobs:
      # XXX: no coverage data handling here, since benchmarks are run on release builds,
      # while coverage is currently collected for the debug ones

-  create-test-report:
+  merge-allure-report:
    runs-on: [ self-hosted, gen3, small ]
    container:
      image: 369495373322.dkr.ecr.eu-central-1.amazonaws.com/rust:pinned
      options: --init
    needs: [ regress-tests, benchmarks ]
    if: ${{ !cancelled() }}
-
+    strategy:
+      fail-fast: false
+      matrix:
+        build_type: [ debug, release ]
    steps:
-      - uses: actions/checkout@v3
+      - name: Checkout
+        uses: actions/checkout@v3
+        with:
+          submodules: false

-      - name: Create Allure report (debug)
-        if: ${{ !cancelled() }}
-        id: create-allure-report-debug
+      - name: Create Allure report
+        id: create-allure-report
        uses: ./.github/actions/allure-report
        with:
          action: generate
-          build_type: debug
-
-      - name: Create Allure report (release)
-        if: ${{ !cancelled() }}
-        id: create-allure-report-release
-        uses: ./.github/actions/allure-report
-        with:
-          action: generate
-          build_type: release
-
-      - uses: actions/github-script@v6
-        if: >
-          !cancelled() &&
-          github.event_name == 'pull_request' && (
-            steps.create-allure-report-debug.outputs.report-url ||
-            steps.create-allure-report-release.outputs.report-url
-          )
-        with:
-          # Retry script for 5XX server errors: https://github.com/actions/github-script#retries
-          retries: 5
-          script: |
-            const reports = [{
-              buildType: "debug",
-              reportUrl: "${{ steps.create-allure-report-debug.outputs.report-url }}",
-              jsonUrl:   "${{ steps.create-allure-report-debug.outputs.report-json-url }}",
-            }, {
-              buildType: "release",
-              reportUrl: "${{ steps.create-allure-report-release.outputs.report-url }}",
-              jsonUrl:   "${{ steps.create-allure-report-release.outputs.report-json-url }}",
-            }]
-
-            const script = require("./scripts/pr-comment-test-report.js")
-            await script({
-              github,
-              context,
-              fetch,
-              reports,
-            })
+          build_type: ${{ matrix.build_type }}

      - name: Store Allure test stat in the DB
-        if: >
-          !cancelled() && (
-            steps.create-allure-report-debug.outputs.report-url ||
-            steps.create-allure-report-release.outputs.report-url
-          )
+        if: ${{ steps.create-allure-report.outputs.report-url }}
        env:
+          BUILD_TYPE: ${{ matrix.build_type }}
          SHA: ${{ github.event.pull_request.head.sha || github.sha }}
-          REPORT_JSON_URL_DEBUG: ${{ steps.create-allure-report-debug.outputs.report-json-url }}
-          REPORT_JSON_URL_RELEASE: ${{ steps.create-allure-report-release.outputs.report-json-url }}
+          REPORT_URL: ${{ steps.create-allure-report.outputs.report-url }}
          TEST_RESULT_CONNSTR: ${{ secrets.REGRESS_TEST_RESULT_CONNSTR }}
        run: |
+          curl --fail --output suites.json ${REPORT_URL%/index.html}/data/suites.json
          ./scripts/pysync

-          for report_url in $REPORT_JSON_URL_DEBUG $REPORT_JSON_URL_RELEASE; do
-            if [ -z "$report_url" ]; then
-              continue
-            fi
-
-            if [[ "$report_url" == "$REPORT_JSON_URL_DEBUG" ]]; then
-              BUILD_TYPE=debug
-            else
-              BUILD_TYPE=release
-            fi
-
-            curl --fail --output suites.json "${report_url}"
-            DATABASE_URL="$TEST_RESULT_CONNSTR" poetry run python3 scripts/ingest_regress_test_result.py --revision ${SHA} --reference ${GITHUB_REF} --build-type ${BUILD_TYPE} --ingest suites.json
-          done
+          DATABASE_URL="$TEST_RESULT_CONNSTR" poetry run python3 scripts/ingest_regress_test_result.py --revision ${SHA} --reference ${GITHUB_REF} --build-type ${BUILD_TYPE} --ingest suites.json

  coverage-report:
    runs-on: [ self-hosted, gen3, small ]
--- a/Cargo.lock
+++ b/Cargo.lock
@@ -2,6 +2,37 @@
 # It is not intended for manual editing.
 version = 3

+[[package]]
+name = "abnf"
+version = "0.12.0"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "33741baa462d86e43fdec5e8ffca7c6ac82847ad06cbfb382c1bdbf527de9e6b"
+dependencies = [
+ "abnf-core",
+ "nom",
+]
+
+[[package]]
+name = "abnf-core"
+version = "0.5.0"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "c44e09c43ae1c368fb91a03a566472d0087c26cf7e1b9e8e289c14ede681dd7d"
+dependencies = [
+ "nom",
+]
+
+[[package]]
+name = "abnf_to_pest"
+version = "0.5.1"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "939d59666dd9a7964a3a5312b9d24c9c107630752ee64f2dd5038189a23fe331"
+dependencies = [
+ "abnf",
+ "indexmap",
+ "itertools",
+ "pretty",
+]
+
 [[package]]
 name = "addr2line"
 version = "0.19.0"
@@ -63,6 +94,15 @@ version = "0.1.6"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "4b46cbb362ab8752921c97e041f5e366ee6297bd428a31275b9fcf1e380f7299"

+[[package]]
+name = "annotate-snippets"
+version = "0.9.1"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "c3b9d411ecbaf79885c6df4d75fff75858d5995ff25385657a28af47e82f9c36"
+dependencies = [
+ "unicode-width",
+]
+
 [[package]]
 name = "anyhow"
 version = "1.0.68"
@@ -81,6 +121,12 @@ dependencies = [
 "static_assertions",
 ]

+[[package]]
+name = "arrayvec"
+version = "0.5.2"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "23b62fc65de8e4e7f52534fb52b0f3ed04746ae267519eef2a83941e8085068b"
+
 [[package]]
 name = "asn1-rs"
 version = "0.5.1"
@@ -737,7 +783,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "213030a2b5a4e0c0892b6652260cf6ccac84827b83a85a534e178e3906c4cf1b"
 dependencies = [
 "ciborium-io",
- "half",
+ "half 1.8.2",
 ]

 [[package]]
@@ -1084,6 +1130,12 @@ dependencies = [
 "winapi",
 ]

+[[package]]
+name = "crunchy"
+version = "0.2.2"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "7a81dae078cea95a014a339291cec439d2f232ebe854a9d672b796c6afafa9b7"
+
 [[package]]
 name = "crypto-common"
 version = "0.1.6"
@@ -1216,6 +1268,42 @@ dependencies = [
 "rusticata-macros",
 ]

+[[package]]
+name = "dhall"
+version = "0.12.0"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "ec26264de25a8e3642fbb37abb24a6c6be9e19795444e6cf1bb88be5c2d55cc7"
+dependencies = [
+ "abnf_to_pest",
+ "annotate-snippets",
+ "elsa",
+ "half 2.2.1",
+ "hex",
+ "home",
+ "itertools",
+ "lazy_static",
+ "minicbor",
+ "once_cell",
+ "percent-encoding",
+ "pest",
+ "pest_consume",
+ "pest_generator",
+ "quote",
+ "sha2",
+ "url",
+]
+
+[[package]]
+name = "dhall_proc_macros"
+version = "0.6.1"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "efcdb228bf802b21cd843e5ac3959b6255966238e5ec06d2e4bc6b9935475653"
+dependencies = [
+ "proc-macro2",
+ "quote",
+ "syn",
+]
+
 [[package]]
 name = "digest"
 version = "0.10.6"
@@ -1238,12 +1326,27 @@ dependencies = [
 "syn",
 ]

+[[package]]
+name = "doc-comment"
+version = "0.3.3"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "fea41bba32d969b513997752735605054bc0dfa92b4c56bf1189f2e174be7a10"
+
 [[package]]
 name = "either"
 version = "1.8.1"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "7fcaabb2fef8c910e7f4c7ce9f67a1283a1715879a7c230ca9d6d1ae31f16d91"

+[[package]]
+name = "elsa"
+version = "1.8.0"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "f74077c3c3aedb99a2683919698285596662518ea13e5eedcf8bdd43b0d0453b"
+dependencies = [
+ "stable_deref_trait",
+]
+
 [[package]]
 name = "encoding_rs"
 version = "0.8.32"
@@ -1556,6 +1659,19 @@ version = "0.3.1"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "d2fabcfbdc87f4758337ca535fb41a6d701b65693ce38287d856d1674551ec9b"

+[[package]]
+name = "globset"
+version = "0.4.10"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "029d74589adefde59de1a0c4f4732695c32805624aec7b68d91503d4dba79afc"
+dependencies = [
+ "aho-corasick",
+ "bstr",
+ "fnv",
+ "log",
+ "regex",
+]
+
 [[package]]
 name = "h2"
 version = "0.3.15"
@@ -1581,6 +1697,15 @@ version = "1.8.2"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "eabb4a44450da02c90444cf74558da904edde8fb4e9035a9a6a4e15445af0bd7"

+[[package]]
+name = "half"
+version = "2.2.1"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "02b4af3693f1b705df946e9fe5631932443781d0aabb423b62fcd4d73f6d2fd0"
+dependencies = [
+ "crunchy",
+]
+
 [[package]]
 name = "hash32"
 version = "0.3.1"
@@ -1677,6 +1802,15 @@ dependencies = [
 "digest",
 ]

+[[package]]
+name = "home"
+version = "0.5.4"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "747309b4b440c06d57b0b25f2aee03ee9b5e5397d288c60e21fc709bb98a7408"
+dependencies = [
+ "winapi",
+]
+
 [[package]]
 name = "hostname"
 version = "0.3.1"
@@ -2123,6 +2257,27 @@ dependencies = [
 "unicase",
 ]

+[[package]]
+name = "minicbor"
+version = "0.18.0"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "2a20020e8e2d1881d8736f64011bb5ff99f1db9947ce3089706945c8915695cb"
+dependencies = [
+ "half 1.8.2",
+ "minicbor-derive",
+]
+
+[[package]]
+name = "minicbor-derive"
+version = "0.12.0"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "8608fb1c805b5b6b3d5ab7bd95c40c396df622b64d77b2d621a5eae1eed050ee"
+dependencies = [
+ "proc-macro2",
+ "quote",
+ "syn",
+]
+
 [[package]]
 name = "minimal-lexical"
 version = "0.2.1"
@@ -2551,6 +2706,72 @@ version = "2.2.0"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "478c572c3d73181ff3c2539045f6eb99e5491218eae919370993b890cdbdd98e"

+[[package]]
+name = "pest"
+version = "2.5.6"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "8cbd939b234e95d72bc393d51788aec68aeeb5d51e748ca08ff3aad58cb722f7"
+dependencies = [
+ "thiserror",
+ "ucd-trie",
+]
+
+[[package]]
+name = "pest_consume"
+version = "1.1.3"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "79447402d15d18e7142e14c72f2e63fa3d155be1bc5b70b3ccbb610ac55f536b"
+dependencies = [
+ "pest",
+ "pest_consume_macros",
+ "pest_derive",
+]
+
+[[package]]
+name = "pest_consume_macros"
+version = "1.1.0"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "9d8630a7a899cb344ec1c16ba0a6b24240029af34bdc0a21f84e411d7f793f29"
+dependencies = [
+ "proc-macro2",
+ "quote",
+ "syn",
+]
+
+[[package]]
+name = "pest_derive"
+version = "2.5.6"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "a81186863f3d0a27340815be8f2078dd8050b14cd71913db9fbda795e5f707d7"
+dependencies = [
+ "pest",
+ "pest_generator",
+]
+
+[[package]]
+name = "pest_generator"
+version = "2.5.6"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "75a1ef20bf3193c15ac345acb32e26b3dc3223aff4d77ae4fc5359567683796b"
+dependencies = [
+ "pest",
+ "pest_meta",
+ "proc-macro2",
+ "quote",
+ "syn",
+]
+
+[[package]]
+name = "pest_meta"
+version = "2.5.6"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "5e3b284b1f13a20dc5ebc90aff59a51b8d7137c221131b52a7260c08cbc1cc80"
+dependencies = [
+ "once_cell",
+ "pest",
+ "sha2",
+]
+
 [[package]]
 name = "petgraph"
 version = "0.6.2"
@@ -2761,6 +2982,18 @@ dependencies = [
 "workspace_hack",
 ]

+[[package]]
+name = "pretty"
+version = "0.11.3"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "83f3aa1e3ca87d3b124db7461265ac176b40c277f37e503eaa29c9c75c037846"
+dependencies = [
+ "arrayvec",
+ "log",
+ "typed-arena",
+ "unicode-segmentation",
+]
+
 [[package]]
 name = "prettyplease"
 version = "0.1.23"
@@ -2909,6 +3142,7 @@ dependencies = [
 "consumption_metrics",
 "futures",
 "git-version",
+ "globset",
 "hashbrown 0.13.2",
 "hashlink",
 "hex",
@@ -2939,6 +3173,7 @@ dependencies = [
 "rustls-pemfile",
 "scopeguard",
 "serde",
+ "serde_dhall",
 "serde_json",
 "sha2",
 "socket2",
@@ -3546,6 +3781,19 @@ dependencies = [
 "syn",
 ]

+[[package]]
+name = "serde_dhall"
+version = "0.12.1"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "655a5c686ad80aef90d2e6bfea3715778623c9a659017c8346bc97eb58f9b27d"
+dependencies = [
+ "dhall",
+ "dhall_proc_macros",
+ "doc-comment",
+ "serde",
+ "url",
+]
+
 [[package]]
 name = "serde_json"
 version = "1.0.91"
@@ -4433,12 +4681,24 @@ dependencies = [
 "utf-8",
 ]

+[[package]]
+name = "typed-arena"
+version = "2.0.2"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "6af6ae20167a9ece4bcb41af5b80f8a1f1df981f6391189ce00fd257af04126a"
+
 [[package]]
 name = "typenum"
 version = "1.16.0"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "497961ef93d974e23eb6f433eb5fe1b7930b659f06d12dec6fc44a8f554c0bba"

+[[package]]
+name = "ucd-trie"
+version = "0.1.5"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "9e79c4d996edb816c91e4308506774452e55e95c3c9de07b6729e17e15a5ef81"
+
 [[package]]
 name = "uname"
 version = "0.1.1"
@@ -4478,6 +4738,12 @@ dependencies = [
 "tinyvec",
 ]

+[[package]]
+name = "unicode-segmentation"
+version = "1.10.1"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "1dd624098567895118886609431a7c3b8f516e41d30e0643f03d94592a147e36"
+
 [[package]]
 name = "unicode-width"
 version = "0.1.10"
--- a/Cargo.toml
+++ b/Cargo.toml
@@ -47,6 +47,7 @@ futures = "0.3"
 futures-core = "0.3"
 futures-util = "0.3"
 git-version = "0.3"
+globset = "0.4.10"
 hashbrown = "0.13"
 hashlink = "0.8.1"
 hex = "0.4"
@@ -87,6 +88,7 @@ rustls-split = "0.3"
 scopeguard = "1.1"
 sentry = { version = "0.29", default-features = false, features = ["backtrace", "contexts", "panic", "rustls", "reqwest" ] }
 serde = { version = "1.0", features = ["derive"] }
+serde_dhall = { version = "0.12.1", default_features = false }
 serde_json = "1"
 serde_with = "2.0"
 sha2 = "0.10.2"
--- a/Dockerfile.compute-node
+++ b/Dockerfile.compute-node
@@ -38,7 +38,6 @@ RUN cd postgres && \
    echo 'trusted = true' >> /usr/local/pgsql/share/extension/insert_username.control && \
    echo 'trusted = true' >> /usr/local/pgsql/share/extension/intagg.control && \
    echo 'trusted = true' >> /usr/local/pgsql/share/extension/moddatetime.control && \
-    echo 'trusted = true' >> /usr/local/pgsql/share/extension/pg_stat_statements.control && \
    echo 'trusted = true' >> /usr/local/pgsql/share/extension/pgrowlocks.control && \
    echo 'trusted = true' >> /usr/local/pgsql/share/extension/pgstattuple.control && \
    echo 'trusted = true' >> /usr/local/pgsql/share/extension/refint.control && \
@@ -301,27 +300,6 @@ RUN wget https://github.com/okbob/plpgsql_check/archive/refs/tags/v2.3.2.tar.gz
    make -j $(getconf _NPROCESSORS_ONLN) install PG_CONFIG=/usr/local/pgsql/bin/pg_config USE_PGXS=1 && \
    echo 'trusted = true' >> /usr/local/pgsql/share/extension/plpgsql_check.control

-#########################################################################################
-#
-# Layer "timescaledb-pg-build"
-# compile timescaledb extension
-#
-#########################################################################################
-FROM build-deps AS timescaledb-pg-build
-COPY --from=pg-build /usr/local/pgsql/ /usr/local/pgsql/
-
-ENV PATH "/usr/local/pgsql/bin:$PATH"
-
-RUN apt-get update && \
-    apt-get install -y cmake && \
-    wget https://github.com/timescale/timescaledb/archive/refs/tags/2.10.1.tar.gz -O timescaledb.tar.gz && \
-    mkdir timescaledb-src && cd timescaledb-src && tar xvzf ../timescaledb.tar.gz --strip-components=1 -C . && \
-    ./bootstrap -DSEND_TELEMETRY_DEFAULT:BOOL=OFF -DUSE_TELEMETRY:BOOL=OFF -DAPACHE_ONLY:BOOL=ON && \
-    cd build && \
-    make -j $(getconf _NPROCESSORS_ONLN) && \
-    make install -j $(getconf _NPROCESSORS_ONLN) && \
-    echo "trusted = true" >> /usr/local/pgsql/share/extension/timescaledb.control
-
 #########################################################################################
 # 
 # Layer "rust extensions"
@@ -426,7 +404,6 @@ COPY --from=pgtap-pg-build /usr/local/pgsql/ /usr/local/pgsql/
 COPY --from=prefix-pg-build /usr/local/pgsql/ /usr/local/pgsql/
 COPY --from=hll-pg-build /usr/local/pgsql/ /usr/local/pgsql/
 COPY --from=plpgsql-check-pg-build /usr/local/pgsql/ /usr/local/pgsql/
-COPY --from=timescaledb-pg-build /usr/local/pgsql/ /usr/local/pgsql/
 COPY pgxn/ pgxn/

 RUN make -j $(getconf _NPROCESSORS_ONLN) \
--- a/compute_tools/src/bin/compute_ctl.rs
+++ b/compute_tools/src/bin/compute_ctl.rs
@@ -34,14 +34,13 @@ use std::fs::File;
 use std::panic;
 use std::path::Path;
 use std::process::exit;
-use std::sync::{Arc, Condvar, Mutex};
+use std::sync::{Arc, RwLock};
 use std::{thread, time::Duration};

 use anyhow::{Context, Result};
 use chrono::Utc;
 use clap::Arg;
 use tracing::{error, info};
-use url::Url;

 use compute_tools::compute::{ComputeMetrics, ComputeNode, ComputeState, ComputeStatus};
 use compute_tools::http::api::launch_http_server;
@@ -50,6 +49,7 @@ use compute_tools::monitor::launch_monitor;
 use compute_tools::params::*;
 use compute_tools::pg_helpers::*;
 use compute_tools::spec::*;
+use url::Url;

 fn main() -> Result<()> {
    init_tracing_and_logging(DEFAULT_LOG_LEVEL)?;
@@ -62,7 +62,7 @@ fn main() -> Result<()> {
    let connstr = matches
        .get_one::<String>("connstr")
        .expect("Postgres connection string is required");
-    let spec_json = matches.get_one::<String>("spec");
+    let spec = matches.get_one::<String>("spec");
    let spec_path = matches.get_one::<String>("spec-path");

    let compute_id = matches.get_one::<String>("compute-id");
@@ -71,107 +71,40 @@ fn main() -> Result<()> {
    // Try to use just 'postgres' if no path is provided
    let pgbin = matches.get_one::<String>("pgbin").unwrap();

-    let mut spec = Default::default();
-    let mut spec_set = false;
-    let mut live_config_allowed = false;
-    match spec_json {
+    let spec: ComputeSpec = match spec {
        // First, try to get cluster spec from the cli argument
-        Some(json) => {
-            spec = serde_json::from_str(json)?;
-            spec_set = true;
-        }
+        Some(json) => serde_json::from_str(json)?,
        None => {
            // Second, try to read it from the file if path is provided
            if let Some(sp) = spec_path {
                let path = Path::new(sp);
                let file = File::open(path)?;
-                spec = serde_json::from_reader(file)?;
-                spec_set = true;
+                serde_json::from_reader(file)?
            } else if let Some(id) = compute_id {
                if let Some(cp_base) = control_plane_uri {
-                    live_config_allowed = true;
-                    if let Ok(s) = get_spec_from_control_plane(cp_base, id) {
-                        spec = s;
-                        spec_set = true;
-                    }
+                    let cp_uri = format!("{cp_base}/management/api/v1/{id}/spec");
+                    let jwt: String = match std::env::var("NEON_CONSOLE_JWT") {
+                        Ok(v) => v,
+                        Err(_) => "".to_string(),
+                    };
+
+                    reqwest::blocking::Client::new()
+                        .get(cp_uri)
+                        .header("Authorization", jwt)
+                        .send()?
+                        .json()?
                } else {
-                    panic!("must specify both --control-plane-uri and --compute-id or none");
+                    panic!(
+                        "must specify --control-plane-uri \"{:#?}\" and --compute-id \"{:#?}\"",
+                        control_plane_uri, compute_id
+                    );
                }
            } else {
-                panic!(
-                    "compute spec should be provided by one of the following ways: \
-                    --spec OR --spec-path OR --control-plane-uri and --compute-id"
-                );
+                panic!("compute spec should be provided via --spec or --spec-path argument");
            }
        }
    };

-    let mut new_state = ComputeState::new();
-    if spec_set {
-        new_state.spec = spec;
-    }
-    let compute_node = ComputeNode {
-        start_time: Utc::now(),
-        connstr: Url::parse(connstr).context("cannot parse connstr as a URL")?,
-        pgdata: pgdata.to_string(),
-        pgbin: pgbin.to_string(),
-        live_config_allowed,
-        metrics: ComputeMetrics::default(),
-        state: Mutex::new(new_state),
-        state_changed: Condvar::new(),
-    };
-    let compute = Arc::new(compute_node);
-
-    // Launch http service first, so we were able to serve control-plane
-    // requests, while configuration is still in progress.
-    let _http_handle = launch_http_server(&compute).expect("cannot launch http endpoint thread");
-
-    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;
-            }
-        }
-    }
-
-    // We got all we need, fill in the state.
-    let mut state = compute.state.lock().unwrap();
-    let pageserver_connstr = state
-        .spec
-        .cluster
-        .settings
-        .find("neon.pageserver_connstring")
-        .expect("pageserver connstr should be provided");
-    let storage_auth_token = state.spec.storage_auth_token.clone();
-    let tenant = state
-        .spec
-        .cluster
-        .settings
-        .find("neon.tenant_id")
-        .expect("tenant id should be provided");
-    let timeline = state
-        .spec
-        .cluster
-        .settings
-        .find("neon.timeline_id")
-        .expect("tenant id should be provided");
-    let startup_tracing_context = state.spec.startup_tracing_context.clone();
-
-    state.pageserver_connstr = pageserver_connstr;
-    state.storage_auth_token = storage_auth_token;
-    state.tenant = tenant;
-    state.timeline = timeline;
-    state.status = ComputeStatus::Init;
-    compute.state_changed.notify_all();
-    drop(state);
-
    // Extract OpenTelemetry context for the startup actions from the spec, and
    // attach it to the current tracing context.
    //
@@ -187,7 +120,7 @@ fn main() -> Result<()> {
    // 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.
-    let startup_context_guard = if let Some(ref carrier) = startup_tracing_context {
+    let startup_context_guard = if let Some(ref carrier) = spec.startup_tracing_context {
        use opentelemetry::propagation::TextMapPropagator;
        use opentelemetry::sdk::propagation::TraceContextPropagator;
        Some(TraceContextPropagator::new().extract(carrier).attach())
@@ -195,7 +128,41 @@ fn main() -> Result<()> {
        None
    };

-    // Launch remaining service threads
+    let pageserver_connstr = spec
+        .cluster
+        .settings
+        .find("neon.pageserver_connstring")
+        .expect("pageserver connstr should be provided");
+    let storage_auth_token = spec.storage_auth_token.clone();
+    let tenant = spec
+        .cluster
+        .settings
+        .find("neon.tenant_id")
+        .expect("tenant id should be provided");
+    let timeline = spec
+        .cluster
+        .settings
+        .find("neon.timeline_id")
+        .expect("tenant id should be provided");
+
+    let compute_state = ComputeNode {
+        start_time: Utc::now(),
+        connstr: Url::parse(connstr).context("cannot parse connstr as a URL")?,
+        pgdata: pgdata.to_string(),
+        pgbin: pgbin.to_string(),
+        spec,
+        tenant,
+        timeline,
+        pageserver_connstr,
+        storage_auth_token,
+        metrics: ComputeMetrics::default(),
+        state: RwLock::new(ComputeState::new()),
+    };
+    let compute = Arc::new(compute_state);
+
+    // Launch service threads first, so we were able to serve availability
+    // requests, while configuration is still in progress.
+    let _http_handle = launch_http_server(&compute).expect("cannot launch http endpoint thread");
    let _monitor_handle = launch_monitor(&compute).expect("cannot launch compute monitor thread");

    // Start Postgres
@@ -205,7 +172,7 @@ fn main() -> Result<()> {
        Ok(pg) => Some(pg),
        Err(err) => {
            error!("could not start the compute node: {:?}", err);
-            let mut state = compute.state.lock().unwrap();
+            let mut state = compute.state.write().unwrap();
            state.error = Some(format!("{:?}", err));
            state.status = ComputeStatus::Failed;
            drop(state);
@@ -295,7 +262,7 @@ fn cli() -> clap::Command {
            Arg::new("control-plane-uri")
                .short('p')
                .long("control-plane-uri")
-                .value_name("CONTROL_PLANE_API_BASE_URI"),
+                .value_name("CONTROL_PLANE"),
        )
 }

--- a/compute_tools/src/compute.rs
+++ b/compute_tools/src/compute.rs
@@ -20,12 +20,12 @@ use std::path::Path;
 use std::process::{Command, Stdio};
 use std::str::FromStr;
 use std::sync::atomic::{AtomicU64, Ordering};
-use std::sync::{Condvar, Mutex};
+use std::sync::RwLock;

 use anyhow::{Context, Result};
 use chrono::{DateTime, Utc};
 use postgres::{Client, NoTls};
-use serde::Serialize;
+use serde::{Serialize, Serializer};
 use tokio_postgres;
 use tracing::{info, instrument, warn};

@@ -41,52 +41,41 @@ pub struct ComputeNode {
    pub connstr: url::Url,
    pub pgdata: String,
    pub pgbin: String,
-    pub metrics: ComputeMetrics,
-    /// We should only allow live re- / configuration of the compute node if
-    /// it uses 'pull model', i.e. it can go to control-plane and fetch
-    /// the latest configuration. Otherwise, there could be a case:
-    /// - we start compute with some spec provided as argument
-    /// - we push new spec and it does reconfiguration
-    /// - but then something happens and compute pod / VM is destroyed,
-    ///   so k8s controller starts it again with the **old** spec
-    /// and the same for empty computes:
-    /// - we started compute without any spec
-    /// - we push spec and it does configuration
-    /// - but then it is restarted without any spec again
-    pub live_config_allowed: bool,
-    /// Volatile part of the `ComputeNode`, which should be used under `Mutex`.
-    /// To allow HTTP API server to serving status requests, while configuration
-    /// is in progress, lock should be held only for short periods of time to do
-    /// read/write, not the whole configuration process.
-    pub state: Mutex<ComputeState>,
-    /// `Condvar` to allow notifying waiters about state changes.
-    pub state_changed: Condvar,
-}
-
-#[derive(Clone, Debug)]
-pub struct ComputeState {
-    pub status: ComputeStatus,
-    /// Timestamp of the last Postgres activity
-    pub last_active: DateTime<Utc>,
-    pub error: Option<String>,
    pub spec: ComputeSpec,
    pub tenant: String,
    pub timeline: String,
    pub pageserver_connstr: String,
    pub storage_auth_token: Option<String>,
+    pub metrics: ComputeMetrics,
+    /// Volatile part of the `ComputeNode` so should be used under `RwLock`
+    /// to allow HTTP API server to serve status requests, while configuration
+    /// is in progress.
+    pub state: RwLock<ComputeState>,
+}
+
+fn rfc3339_serialize<S>(x: &DateTime<Utc>, s: S) -> Result<S::Ok, S::Error>
+where
+    S: Serializer,
+{
+    x.to_rfc3339().serialize(s)
+}
+
+#[derive(Serialize)]
+#[serde(rename_all = "snake_case")]
+pub struct ComputeState {
+    pub status: ComputeStatus,
+    /// Timestamp of the last Postgres activity
+    #[serde(serialize_with = "rfc3339_serialize")]
+    pub last_active: DateTime<Utc>,
+    pub error: Option<String>,
 }

 impl ComputeState {
    pub fn new() -> Self {
        Self {
-            status: ComputeStatus::Empty,
+            status: ComputeStatus::Init,
            last_active: Utc::now(),
            error: None,
-            spec: ComputeSpec::default(),
-            tenant: String::new(),
-            timeline: String::new(),
-            pageserver_connstr: String::new(),
-            storage_auth_token: None,
        }
    }
 }
@@ -97,22 +86,11 @@ impl Default for ComputeState {
    }
 }

-#[derive(Serialize, Clone, Copy, PartialEq, Eq, Debug)]
+#[derive(Serialize, Clone, Copy, PartialEq, Eq)]
 #[serde(rename_all = "snake_case")]
 pub enum ComputeStatus {
-    // Spec wasn't provided at start, waiting for it to be
-    // provided by control-plane.
-    Empty,
-    // Compute configuration was requested.
-    ConfigurationPending,
-    // Compute node has spec and initial startup and
-    // configuration is in progress.
    Init,
-    // Compute is configured and running.
    Running,
-    // Either startup or configuration failed,
-    // compute will exit soon or is waiting for
-    // control-plane to terminate it.
    Failed,
 }

@@ -126,13 +104,11 @@ pub struct ComputeMetrics {

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

    pub fn get_status(&self) -> ComputeStatus {
-        self.state.lock().unwrap().status
+        self.state.read().unwrap().status
    }

    // Remove `pgdata` directory and create it again with right permissions.
@@ -148,15 +124,15 @@ impl ComputeNode {

    // Get basebackup from the libpq connection to pageserver using `connstr` and
    // unarchive it to `pgdata` directory overriding all its previous content.
-    #[instrument(skip(self, compute_state))]
-    fn get_basebackup(&self, compute_state: &ComputeState, lsn: &str) -> Result<()> {
+    #[instrument(skip(self))]
+    fn get_basebackup(&self, lsn: &str) -> Result<()> {
        let start_time = Utc::now();

-        let mut config = postgres::Config::from_str(&compute_state.pageserver_connstr)?;
+        let mut config = postgres::Config::from_str(&self.pageserver_connstr)?;

        // Use the storage auth token from the config file, if given.
        // Note: this overrides any password set in the connection string.
-        if let Some(storage_auth_token) = &compute_state.storage_auth_token {
+        if let Some(storage_auth_token) = &self.storage_auth_token {
            info!("Got storage auth token from spec file");
            config.password(storage_auth_token);
        } else {
@@ -165,14 +141,8 @@ impl ComputeNode {

        let mut client = config.connect(NoTls)?;
        let basebackup_cmd = match lsn {
-            "0/0" => format!(
-                "basebackup {} {}",
-                &compute_state.tenant, &compute_state.timeline
-            ), // First start of the compute
-            _ => format!(
-                "basebackup {} {} {}",
-                &compute_state.tenant, &compute_state.timeline, lsn
-            ),
+            "0/0" => format!("basebackup {} {}", &self.tenant, &self.timeline), // First start of the compute
+            _ => format!("basebackup {} {} {}", &self.tenant, &self.timeline, lsn),
        };
        let copyreader = client.copy_out(basebackup_cmd.as_str())?;

@@ -199,14 +169,14 @@ impl ComputeNode {

    // Run `postgres` in a special mode with `--sync-safekeepers` argument
    // and return the reported LSN back to the caller.
-    #[instrument(skip(self, storage_auth_token))]
-    fn sync_safekeepers(&self, storage_auth_token: Option<String>) -> Result<String> {
+    #[instrument(skip(self))]
+    fn sync_safekeepers(&self) -> Result<String> {
        let start_time = Utc::now();

        let sync_handle = Command::new(&self.pgbin)
            .args(["--sync-safekeepers"])
            .env("PGDATA", &self.pgdata) // we cannot use -D in this mode
-            .envs(if let Some(storage_auth_token) = &storage_auth_token {
+            .envs(if let Some(storage_auth_token) = &self.storage_auth_token {
                vec![("NEON_AUTH_TOKEN", storage_auth_token)]
            } else {
                vec![]
@@ -247,9 +217,9 @@ impl ComputeNode {

    /// Do all the preparations like PGDATA directory creation, configuration,
    /// safekeepers sync, basebackup, etc.
-    #[instrument(skip(self, compute_state))]
-    pub fn prepare_pgdata(&self, compute_state: &ComputeState) -> Result<()> {
-        let spec = &compute_state.spec;
+    #[instrument(skip(self))]
+    pub fn prepare_pgdata(&self) -> Result<()> {
+        let spec = &self.spec;
        let pgdata_path = Path::new(&self.pgdata);

        // Remove/create an empty pgdata directory and put configuration there.
@@ -258,18 +228,18 @@ impl ComputeNode {

        info!("starting safekeepers syncing");
        let lsn = self
-            .sync_safekeepers(compute_state.storage_auth_token.clone())
+            .sync_safekeepers()
            .with_context(|| "failed to sync safekeepers")?;
        info!("safekeepers synced at LSN {}", lsn);

        info!(
            "getting basebackup@{} from pageserver {}",
-            lsn, &compute_state.pageserver_connstr
+            lsn, &self.pageserver_connstr
        );
-        self.get_basebackup(compute_state, &lsn).with_context(|| {
+        self.get_basebackup(&lsn).with_context(|| {
            format!(
                "failed to get basebackup@{} from pageserver {}",
-                lsn, &compute_state.pageserver_connstr
+                lsn, &self.pageserver_connstr
            )
        })?;

@@ -282,16 +252,13 @@ impl ComputeNode {
    /// Start Postgres as a child process and manage DBs/roles.
    /// After that this will hang waiting on the postmaster process to exit.
    #[instrument(skip(self))]
-    pub fn start_postgres(
-        &self,
-        storage_auth_token: Option<String>,
-    ) -> Result<std::process::Child> {
+    pub fn start_postgres(&self) -> Result<std::process::Child> {
        let pgdata_path = Path::new(&self.pgdata);

        // Run postgres as a child process.
        let mut pg = Command::new(&self.pgbin)
            .args(["-D", &self.pgdata])
-            .envs(if let Some(storage_auth_token) = &storage_auth_token {
+            .envs(if let Some(storage_auth_token) = &self.storage_auth_token {
                vec![("NEON_AUTH_TOKEN", storage_auth_token)]
            } else {
                vec![]
@@ -304,9 +271,8 @@ impl ComputeNode {
        Ok(pg)
    }

-    /// Do initial configuration of the already started Postgres.
-    #[instrument(skip(self, compute_state))]
-    pub fn apply_config(&self, compute_state: &ComputeState) -> Result<()> {
+    #[instrument(skip(self))]
+    pub fn apply_config(&self) -> Result<()> {
        // If connection fails,
        // it may be the old node with `zenith_admin` superuser.
        //
@@ -337,19 +303,19 @@ impl ComputeNode {
        };

        // Proceed with post-startup configuration. Note, that order of operations is important.
-        handle_roles(&compute_state.spec, &mut client)?;
-        handle_databases(&compute_state.spec, &mut client)?;
-        handle_role_deletions(&compute_state.spec, self.connstr.as_str(), &mut client)?;
-        handle_grants(&compute_state.spec, self.connstr.as_str(), &mut client)?;
+        handle_roles(&self.spec, &mut client)?;
+        handle_databases(&self.spec, &mut client)?;
+        handle_role_deletions(self, &mut client)?;
+        handle_grants(self, &mut client)?;
        create_writability_check_data(&mut client)?;
-        handle_extensions(&compute_state.spec, &mut client)?;
+        handle_extensions(&self.spec, &mut client)?;

        // 'Close' connection
        drop(client);

        info!(
            "finished configuration of compute for project {}",
-            compute_state.spec.cluster.cluster_id
+            self.spec.cluster.cluster_id
        );

        Ok(())
@@ -357,22 +323,21 @@ impl ComputeNode {

    #[instrument(skip(self))]
    pub fn start_compute(&self) -> Result<std::process::Child> {
-        let compute_state = self.state.lock().unwrap().clone();
        info!(
            "starting compute for project {}, operation {}, tenant {}, timeline {}",
-            compute_state.spec.cluster.cluster_id,
-            compute_state.spec.operation_uuid.as_ref().unwrap(),
-            compute_state.tenant,
-            compute_state.timeline,
+            self.spec.cluster.cluster_id,
+            self.spec.operation_uuid.as_ref().unwrap(),
+            self.tenant,
+            self.timeline,
        );

-        self.prepare_pgdata(&compute_state)?;
+        self.prepare_pgdata()?;

        let start_time = Utc::now();

-        let pg = self.start_postgres(compute_state.storage_auth_token.clone())?;
+        let pg = self.start_postgres()?;

-        self.apply_config(&compute_state)?;
+        self.apply_config()?;

        let startup_end_time = Utc::now();
        self.metrics.config_ms.store(
--- a/compute_tools/src/http/api.rs
+++ b/compute_tools/src/http/api.rs
@@ -3,16 +3,12 @@ use std::net::SocketAddr;
 use std::sync::Arc;
 use std::thread;

-use crate::compute::{ComputeNode, ComputeStatus};
-use crate::http::requests::ConfigurationRequest;
-use crate::http::responses::{ComputeStatusResponse, GenericAPIError};
-
+use crate::compute::ComputeNode;
 use anyhow::Result;
 use hyper::service::{make_service_fn, service_fn};
 use hyper::{Body, Method, Request, Response, Server, StatusCode};
 use num_cpus;
 use serde_json;
-use tokio::task;
 use tracing::{error, info};
 use tracing_utils::http::OtelName;

@@ -27,10 +23,8 @@ async fn routes(req: Request<Body>, compute: &Arc<ComputeNode>) -> Response<Body
        // Serialized compute state.
        (&Method::GET, "/status") => {
            info!("serving /status GET request");
-            let state = compute.state.lock().unwrap();
-            let status_response = ComputeStatusResponse::from(state.clone());
-
-            Response::new(Body::from(serde_json::to_string(&status_response).unwrap()))
+            let state = compute.state.read().unwrap();
+            Response::new(Body::from(serde_json::to_string(&*state).unwrap()))
        }

        // Startup metrics in JSON format. Keep /metrics reserved for a possible
@@ -43,29 +37,12 @@ async fn routes(req: Request<Body>, compute: &Arc<ComputeNode>) -> Response<Body
        // Collect Postgres current usage insights
        (&Method::GET, "/insights") => {
            info!("serving /insights GET request");
-            let status = compute.get_status();
-            if status != ComputeStatus::Running {
-                let msg = format!("compute is not running, current status: {:?}", status);
-                error!(msg);
-                return Response::new(Body::from(msg));
-            }
-
            let insights = compute.collect_insights().await;
            Response::new(Body::from(insights))
        }

        (&Method::POST, "/check_writability") => {
            info!("serving /check_writability POST request");
-            let status = compute.get_status();
-            if status != ComputeStatus::Running {
-                let msg = format!(
-                    "invalid compute status for check_writability request: {:?}",
-                    status
-                );
-                error!(msg);
-                return Response::new(Body::from(msg));
-            }
-
            let res = crate::checker::check_writability(compute).await;
            match res {
                Ok(_) => Response::new(Body::from("true")),
@@ -84,23 +61,6 @@ async fn routes(req: Request<Body>, compute: &Arc<ComputeNode>) -> Response<Body
            ))
        }

-        // Accept spec in JSON format and request compute configuration. If
-        // anything goes wrong after we set the compute status to `ConfigurationPending`
-        // and update compute state with new spec, we basically leave compute
-        // in the potentially wrong state. That said, it's control-plane's
-        // responsibility to watch compute state after reconfiguration request
-        // and to clean restart in case of errors.
-        (&Method::POST, "/configure") => {
-            info!("serving /configure POST request");
-            match handle_configure_request(req, compute).await {
-                Ok(msg) => Response::new(Body::from(msg)),
-                Err((msg, code)) => {
-                    error!("error handling /configure request: {msg}");
-                    render_json_error(&msg, code)
-                }
-            }
-        }
-
        // Return the `404 Not Found` for any other routes.
        _ => {
            let mut not_found = Response::new(Body::from("404 Not Found"));
@@ -110,88 +70,6 @@ async fn routes(req: Request<Body>, compute: &Arc<ComputeNode>) -> Response<Body
    }
 }

-async fn handle_configure_request(
-    req: Request<Body>,
-    compute: &Arc<ComputeNode>,
-) -> Result<String, (String, StatusCode)> {
-    if !compute.live_config_allowed {
-        return Err((
-            "live configuration is not allowed for this compute node".to_string(),
-            StatusCode::PRECONDITION_FAILED,
-        ));
-    }
-
-    let body_bytes = hyper::body::to_bytes(req.into_body()).await.unwrap();
-    let spec_raw = String::from_utf8(body_bytes.to_vec()).unwrap();
-    if let Ok(request) = serde_json::from_str::<ConfigurationRequest>(&spec_raw) {
-        let spec = request.spec;
-        // XXX: wrap state update under lock in code blocks. Otherwise,
-        // we will try to `Send` `mut state` into the spawned thread
-        // bellow, which will cause error:
-        // ```
-        // error: future cannot be sent between threads safely
-        // ```
-        {
-            let mut state = compute.state.lock().unwrap();
-            if state.status != ComputeStatus::Empty {
-                let msg = format!(
-                    "invalid compute status for configuration request: {:?}",
-                    state.status.clone()
-                );
-                return Err((msg, StatusCode::PRECONDITION_FAILED));
-            }
-            state.spec = spec;
-            state.status = ComputeStatus::ConfigurationPending;
-            compute.state_changed.notify_all();
-            drop(state);
-            info!("set new spec and notified waiters");
-        }
-
-        // Spawn a blocking thread to wait for compute to become Running.
-        // This is needed to do not block the main pool of workers and
-        // be able to serve other requests while some particular request
-        // is waiting for compute to finish configuration.
-        let c = compute.clone();
-        task::spawn_blocking(move || {
-            let mut state = c.state.lock().unwrap();
-            while state.status != ComputeStatus::Running {
-                state = c.state_changed.wait(state).unwrap();
-                info!(
-                    "waiting for compute to become Running, current status: {:?}",
-                    state.status
-                );
-
-                if state.status == ComputeStatus::Failed {
-                    let err = state.error.clone().unwrap_or("unknown error".to_string());
-                    let msg = format!("compute configuration failed: {:?}", err);
-                    return Err((msg, StatusCode::INTERNAL_SERVER_ERROR));
-                }
-            }
-
-            Ok(())
-        })
-        .await
-        .unwrap()?;
-
-        // Return current compute state if everything went well.
-        let state = compute.state.lock().unwrap().clone();
-        let status_response = ComputeStatusResponse::from(state);
-        Ok(serde_json::to_string(&status_response).unwrap())
-    } else {
-        Err(("invalid spec".to_string(), StatusCode::BAD_REQUEST))
-    }
-}
-
-fn render_json_error(e: &str, status: StatusCode) -> Response<Body> {
-    let error = GenericAPIError {
-        error: e.to_string(),
-    };
-    Response::builder()
-        .status(status)
-        .body(Body::from(serde_json::to_string(&error).unwrap()))
-        .unwrap()
-}
-
 // Main Hyper HTTP server function that runs it and blocks waiting on it forever.
 #[tokio::main]
 async fn serve(state: Arc<ComputeNode>) {
--- a/compute_tools/src/http/mod.rs
+++ b/compute_tools/src/http/mod.rs
@@ -1,3 +1 @@
 pub mod api;
-pub mod requests;
-pub mod responses;
--- a/compute_tools/src/http/openapi_spec.yaml
+++ b/compute_tools/src/http/openapi_spec.yaml
@@ -11,7 +11,7 @@ paths:
    get:
      tags:
      - Info
-      summary: Get compute node internal status.
+      summary: Get compute node internal status
      description: ""
      operationId: getComputeStatus
      responses:
@@ -26,7 +26,7 @@ paths:
    get:
      tags:
      - Info
-      summary: Get compute node startup metrics in JSON format.
+      summary: Get compute node startup metrics in JSON format
      description: ""
      operationId: getComputeMetricsJSON
      responses:
@@ -41,9 +41,9 @@ paths:
    get:
      tags:
      - Info
-      summary: Get current compute insights in JSON format.
+      summary: Get current compute insights in JSON format
      description: |
-        Note, that this doesn't include any historical data.
+        Note, that this doesn't include any historical data
      operationId: getComputeInsights
      responses:
        200:
@@ -56,12 +56,12 @@ paths:
  /info:
    get:
      tags:
-      - Info
-      summary: Get info about the compute pod / VM.
+      - "info"
+      summary: Get info about the compute Pod/VM
      description: ""
      operationId: getInfo
      responses:
-        200:
+        "200":
          description: Info
          content:
            application/json:
@@ -72,7 +72,7 @@ paths:
    post:
      tags:
      - Check
-      summary: Check that we can write new data on this compute.
+      summary: Check that we can write new data on this compute
      description: ""
      operationId: checkComputeWritability
      responses:
@@ -82,64 +82,9 @@ paths:
            text/plain:
              schema:
                type: string
-                description: Error text or 'true' if check passed.
+                description: Error text or 'true' if check passed
                example: "true"

-  /configure:
-    post:
-      tags:
-      - Configure
-      summary: Perform compute node configuration.
-      description: |
-        This is a blocking API endpoint, i.e. it blocks waiting until
-        compute is finished configuration and is in `Running` state.
-        Optional non-blocking mode could be added later.
-      operationId: configureCompute
-      requestBody:
-        description: Configuration request.
-        required: true
-        content:
-          application/json:
-            schema:
-              type: object
-              required:
-                - spec
-              properties:
-                spec:
-                  # XXX: I don't want to explain current spec in the OpenAPI format,
-                  # as it could be changed really soon. Consider doing it later.
-                  type: object
-      responses:
-        200:
-          description: Compute configuration finished.
-          content:
-            application/json:
-              schema:
-                $ref: "#/components/schemas/ComputeState"
-        400:
-          description: Provided spec is invalid.
-          content:
-            application/json:
-              schema:
-                $ref: "#/components/schemas/GenericError"
-        412:
-          description: |
-            It's not possible to do live-configuration of the compute.
-            It's either in the wrong state, or compute doesn't use pull
-            mode of configuration.
-          content:
-            application/json:
-              schema:
-                $ref: "#/components/schemas/GenericError"
-        500:
-          description: |
-            Compute configuration request was processed, but error
-            occurred. Compute will likely shutdown soon.
-          content:
-            application/json:
-              schema:
-                $ref: "#/components/schemas/GenericError"
-
 components:
  securitySchemes:
    JWT:
@@ -150,7 +95,7 @@ components:
  schemas:
    ComputeMetrics:
      type: object
-      description: Compute startup metrics.
+      description: Compute startup metrics
      required:
        - sync_safekeepers_ms
        - basebackup_ms
@@ -168,7 +113,7 @@ components:

    Info:
      type: object
-      description: Information about VM/Pod.
+      description: Information about VM/Pod
      required:
        - num_cpus
      properties:
@@ -185,26 +130,17 @@ components:
          $ref: '#/components/schemas/ComputeStatus'
        last_active:
          type: string
-          description: The last detected compute activity timestamp in UTC and RFC3339 format.
+          description: The last detected compute activity timestamp in UTC and RFC3339 format
          example: "2022-10-12T07:20:50.52Z"
        error:
          type: string
-          description: Text of the error during compute startup, if any.
-          example: ""
-        tenant:
-          type: string
-          description: Identifier of the current tenant served by compute node, if any.
-          example: c9269c359e9a199fad1ea0981246a78f
-        timeline:
-          type: string
-          description: Identifier of the current timeline served by compute node, if any.
-          example: ece7de74d4b8cbe5433a68ce4d1b97b4
+          description: Text of the error during compute startup, if any

    ComputeInsights:
      type: object
      properties:
        pg_stat_statements:
-          description: Contains raw output from pg_stat_statements in JSON format.
+          description: Contains raw output from pg_stat_statements in JSON format
          type: array
          items:
            type: object
@@ -215,19 +151,6 @@ components:
        - init
        - failed
        - running
-      example: running
-
-    #
-    # Errors
-    #
-
-    GenericError:
-      type: object
-      required:
-        - error
-      properties:
-        error:
-          type: string

 security:
  - JWT: []
--- a/compute_tools/src/http/requests.rs
+++ b/compute_tools/src/http/requests.rs
@@ -1,11 +0,0 @@
-use serde::Deserialize;
-
-use crate::spec::ComputeSpec;
-
-/// We now pass only `spec` in the configuration request, but later we can
-/// extend it and something like `restart: bool` or something else. So put
-/// `spec` into a struct initially to be more flexible in the future.
-#[derive(Deserialize, Debug)]
-pub struct ConfigurationRequest {
-    pub spec: ComputeSpec,
-}
--- a/compute_tools/src/http/responses.rs
+++ b/compute_tools/src/http/responses.rs
@@ -1,40 +0,0 @@
-use serde::{Serialize, Serializer};
-
-use chrono::{DateTime, Utc};
-
-use crate::compute::{ComputeState, ComputeStatus};
-
-#[derive(Serialize, Debug)]
-pub struct GenericAPIError {
-    pub error: String,
-}
-
-#[derive(Serialize, Debug)]
-#[serde(rename_all = "snake_case")]
-pub struct ComputeStatusResponse {
-    pub tenant: String,
-    pub timeline: String,
-    pub status: ComputeStatus,
-    #[serde(serialize_with = "rfc3339_serialize")]
-    pub last_active: DateTime<Utc>,
-    pub error: Option<String>,
-}
-
-impl From<ComputeState> for ComputeStatusResponse {
-    fn from(state: ComputeState) -> Self {
-        ComputeStatusResponse {
-            tenant: state.tenant,
-            timeline: state.timeline,
-            status: state.status,
-            last_active: state.last_active,
-            error: state.error,
-        }
-    }
-}
-
-fn rfc3339_serialize<S>(x: &DateTime<Utc>, s: S) -> Result<S::Ok, S::Error>
-where
-    S: Serializer,
-{
-    x.to_rfc3339().serialize(s)
-}
--- a/compute_tools/src/monitor.rs
+++ b/compute_tools/src/monitor.rs
@@ -46,7 +46,7 @@ fn watch_compute_activity(compute: &ComputeNode) {
                            AND usename != 'cloud_admin';", // XXX: find a better way to filter other monitors?
                        &[],
                    );
-                let mut last_active = compute.state.lock().unwrap().last_active;
+                let mut last_active = compute.state.read().unwrap().last_active;

                if let Ok(backs) = backends {
                    let mut idle_backs: Vec<DateTime<Utc>> = vec![];
@@ -87,7 +87,7 @@ fn watch_compute_activity(compute: &ComputeNode) {
                }

                // Update the last activity in the shared state if we got a more recent one.
-                let mut state = compute.state.lock().unwrap();
+                let mut state = compute.state.write().unwrap();
                if last_active > state.last_active {
                    state.last_active = last_active;
                    debug!("set the last compute activity time to: {}", last_active);
--- a/compute_tools/src/pg_helpers.rs
+++ b/compute_tools/src/pg_helpers.rs
@@ -17,7 +17,7 @@ const POSTGRES_WAIT_TIMEOUT: Duration = Duration::from_millis(60 * 1000); // mil

 /// Rust representation of Postgres role info with only those fields
 /// that matter for us.
-#[derive(Clone, Deserialize, Debug)]
+#[derive(Clone, Deserialize)]
 pub struct Role {
    pub name: PgIdent,
    pub encrypted_password: Option<String>,
@@ -26,7 +26,7 @@ pub struct Role {

 /// Rust representation of Postgres database info with only those fields
 /// that matter for us.
-#[derive(Clone, Deserialize, Debug)]
+#[derive(Clone, Deserialize)]
 pub struct Database {
    pub name: PgIdent,
    pub owner: PgIdent,
@@ -36,7 +36,7 @@ pub struct Database {
 /// Common type representing both SQL statement params with or without value,
 /// like `LOGIN` or `OWNER username` in the `CREATE/ALTER ROLE`, and config
 /// options like `wal_level = logical`.
-#[derive(Clone, Deserialize, Debug)]
+#[derive(Clone, Deserialize)]
 pub struct GenericOption {
    pub name: String,
    pub value: Option<String>,
--- a/compute_tools/src/spec.rs
+++ b/compute_tools/src/spec.rs
@@ -8,13 +8,14 @@ use postgres::{Client, NoTls};
 use serde::Deserialize;
 use tracing::{info, info_span, instrument, span_enabled, warn, Level};

+use crate::compute::ComputeNode;
 use crate::config;
 use crate::params::PG_HBA_ALL_MD5;
 use crate::pg_helpers::*;

 /// Cluster spec or configuration represented as an optional number of
 /// delta operations + final cluster state description.
-#[derive(Clone, Deserialize, Debug, Default)]
+#[derive(Clone, Deserialize)]
 pub struct ComputeSpec {
    pub format_version: f32,
    pub timestamp: String,
@@ -30,7 +31,7 @@ pub struct ComputeSpec {

 /// Cluster state seen from the perspective of the external tools
 /// like Rails web console.
-#[derive(Clone, Deserialize, Debug, Default)]
+#[derive(Clone, Deserialize)]
 pub struct Cluster {
    pub cluster_id: String,
    pub name: String,
@@ -46,36 +47,13 @@ pub struct Cluster {
 /// - DROP ROLE
 /// - ALTER ROLE name RENAME TO new_name
 /// - ALTER DATABASE name RENAME TO new_name
-#[derive(Clone, Deserialize, Debug)]
+#[derive(Clone, Deserialize)]
 pub struct DeltaOp {
    pub action: String,
    pub name: PgIdent,
    pub new_name: Option<PgIdent>,
 }

-/// Request spec from the control-plane by compute_id. If `NEON_CONSOLE_JWT`
-/// env variable is set, it will be used for authorization.
-pub fn get_spec_from_control_plane(base_uri: &str, compute_id: &str) -> Result<ComputeSpec> {
-    let cp_uri = format!("{base_uri}/management/api/v2/computes/{compute_id}/spec");
-    let jwt: String = match std::env::var("NEON_CONSOLE_JWT") {
-        Ok(v) => v,
-        Err(_) => "".to_string(),
-    };
-    info!("getting spec from control plane: {}", cp_uri);
-
-    // TODO: check the response. We should distinguish cases when it's
-    // - network error, then retry
-    // - no spec for compute yet, then wait
-    // - compute id is unknown or any other error, then bail out
-    let spec = reqwest::blocking::Client::new()
-        .get(cp_uri)
-        .header("Authorization", jwt)
-        .send()?
-        .json()?;
-
-    Ok(spec)
-}
-
 /// It takes cluster specification and does the following:
 /// - Serialize cluster config and put it into `postgresql.conf` completely rewriting the file.
 /// - Update `pg_hba.conf` to allow external connections.
@@ -248,8 +226,8 @@ pub fn handle_roles(spec: &ComputeSpec, client: &mut Client) -> Result<()> {

 /// Reassign all dependent objects and delete requested roles.
 #[instrument(skip_all)]
-pub fn handle_role_deletions(spec: &ComputeSpec, connstr: &str, client: &mut Client) -> Result<()> {
-    if let Some(ops) = &spec.delta_operations {
+pub fn handle_role_deletions(node: &ComputeNode, client: &mut Client) -> Result<()> {
+    if let Some(ops) = &node.spec.delta_operations {
        // First, reassign all dependent objects to db owners.
        info!("reassigning dependent objects of to-be-deleted roles");

@@ -266,7 +244,7 @@ pub fn handle_role_deletions(spec: &ComputeSpec, connstr: &str, client: &mut Cli
            // Check that role is still present in Postgres, as this could be a
            // restart with the same spec after role deletion.
            if op.action == "delete_role" && existing_roles.iter().any(|r| r.name == op.name) {
-                reassign_owned_objects(spec, connstr, &op.name)?;
+                reassign_owned_objects(node, &op.name)?;
            }
        }

@@ -290,10 +268,10 @@ pub fn handle_role_deletions(spec: &ComputeSpec, connstr: &str, client: &mut Cli
 }

 // Reassign all owned objects in all databases to the owner of the database.
-fn reassign_owned_objects(spec: &ComputeSpec, connstr: &str, role_name: &PgIdent) -> Result<()> {
-    for db in &spec.cluster.databases {
+fn reassign_owned_objects(node: &ComputeNode, role_name: &PgIdent) -> Result<()> {
+    for db in &node.spec.cluster.databases {
        if db.owner != *role_name {
-            let mut conf = Config::from_str(connstr)?;
+            let mut conf = Config::from_str(node.connstr.as_str())?;
            conf.dbname(&db.name);

            let mut client = conf.connect(NoTls)?;
@@ -438,7 +416,9 @@ pub fn handle_databases(spec: &ComputeSpec, client: &mut Client) -> Result<()> {
 /// Grant CREATE ON DATABASE to the database owner and do some other alters and grants
 /// to allow users creating trusted extensions and re-creating `public` schema, for example.
 #[instrument(skip_all)]
-pub fn handle_grants(spec: &ComputeSpec, connstr: &str, client: &mut Client) -> Result<()> {
+pub fn handle_grants(node: &ComputeNode, client: &mut Client) -> Result<()> {
+    let spec = &node.spec;
+
    info!("cluster spec grants:");

    // We now have a separate `web_access` role to connect to the database
@@ -470,8 +450,8 @@ pub fn handle_grants(spec: &ComputeSpec, connstr: &str, client: &mut Client) ->
    // Do some per-database access adjustments. We'd better do this at db creation time,
    // but CREATE DATABASE isn't transactional. So we cannot create db + do some grants
    // atomically.
-    for db in &spec.cluster.databases {
-        let mut conf = Config::from_str(connstr)?;
+    for db in &node.spec.cluster.databases {
+        let mut conf = Config::from_str(node.connstr.as_str())?;
        conf.dbname(&db.name);

        let mut db_client = conf.connect(NoTls)?;
--- a/libs/pq_proto/src/lib.rs
+++ b/libs/pq_proto/src/lib.rs
@@ -293,9 +293,6 @@ impl FeStartupPacket {
        // We shouldn't advance `buf` as probably full message is not there yet,
        // so can't directly use Bytes::get_u32 etc.
        let len = (&buf[0..4]).read_u32::<BigEndian>().unwrap() as usize;
-        // The proposed replacement is `!(4..=MAX_STARTUP_PACKET_LENGTH).contains(&len)`
-        // which is less readable
-        #[allow(clippy::manual_range_contains)]
        if len < 4 || len > MAX_STARTUP_PACKET_LENGTH {
            return Err(ProtocolError::Protocol(format!(
                "invalid startup packet message length {}",
--- a/libs/remote_storage/src/lib.rs
+++ b/libs/remote_storage/src/lib.rs
@@ -78,6 +78,9 @@ impl RemotePath {
 /// providing basic CRUD operations for storage files.
 #[async_trait::async_trait]
 pub trait RemoteStorage: Send + Sync + 'static {
+    /// Lists all items the storage has right now.
+    async fn list(&self) -> anyhow::Result<Vec<RemotePath>>;
+
    /// Lists all top level subdirectories for a given prefix
    /// Note: here we assume that if the prefix is passed it was obtained via remote_object_id
    /// which already takes into account any kind of global prefix (prefix_in_bucket for S3 or storage_root for LocalFS)
--- a/libs/remote_storage/src/local_fs.rs
+++ b/libs/remote_storage/src/local_fs.rs
@@ -73,8 +73,10 @@ impl LocalFs {
            Ok(None)
        }
    }
+}

-    #[cfg(test)]
+#[async_trait::async_trait]
+impl RemoteStorage for LocalFs {
    async fn list(&self) -> anyhow::Result<Vec<RemotePath>> {
        Ok(get_all_files(&self.storage_root, true)
            .await?
@@ -89,10 +91,7 @@ impl LocalFs {
            })
            .collect())
    }
-}

-#[async_trait::async_trait]
-impl RemoteStorage for LocalFs {
    async fn list_prefixes(
        &self,
        prefix: Option<&RemotePath>,
--- a/libs/remote_storage/src/s3_bucket.rs
+++ b/libs/remote_storage/src/s3_bucket.rs
@@ -275,6 +275,50 @@ impl<S: AsyncRead> AsyncRead for RatelimitedAsyncRead<S> {

 #[async_trait::async_trait]
 impl RemoteStorage for S3Bucket {
+    async fn list(&self) -> anyhow::Result<Vec<RemotePath>> {
+        let mut document_keys = Vec::new();
+
+        let mut continuation_token = None;
+        loop {
+            let _guard = self
+                .concurrency_limiter
+                .acquire()
+                .await
+                .context("Concurrency limiter semaphore got closed during S3 list")?;
+
+            metrics::inc_list_objects();
+
+            let fetch_response = self
+                .client
+                .list_objects_v2()
+                .bucket(self.bucket_name.clone())
+                .set_prefix(self.prefix_in_bucket.clone())
+                .delimiter(REMOTE_STORAGE_PREFIX_SEPARATOR.to_string())
+                .set_continuation_token(continuation_token)
+                .set_max_keys(self.max_keys_per_list_response)
+                .send()
+                .await
+                .map_err(|e| {
+                    metrics::inc_list_objects_fail();
+                    e
+                })?;
+            document_keys.extend(
+                fetch_response
+                    .contents
+                    .unwrap_or_default()
+                    .into_iter()
+                    .filter_map(|o| Some(self.s3_object_to_relative_path(o.key()?))),
+            );
+
+            match fetch_response.next_continuation_token {
+                Some(new_token) => continuation_token = Some(new_token),
+                None => break,
+            }
+        }
+
+        Ok(document_keys)
+    }
+
    /// See the doc for `RemoteStorage::list_prefixes`
    /// Note: it wont include empty "directories"
    async fn list_prefixes(
--- a/libs/remote_storage/src/simulate_failures.rs
+++ b/libs/remote_storage/src/simulate_failures.rs
@@ -20,6 +20,7 @@ pub struct UnreliableWrapper {
 /// Used to identify retries of different unique operation.
 #[derive(Debug, Hash, Eq, PartialEq)]
 enum RemoteOp {
+    List,
    ListPrefixes(Option<RemotePath>),
    Upload(RemotePath),
    Download(RemotePath),
@@ -74,6 +75,12 @@ impl UnreliableWrapper {

 #[async_trait::async_trait]
 impl RemoteStorage for UnreliableWrapper {
+    /// Lists all items the storage has right now.
+    async fn list(&self) -> anyhow::Result<Vec<RemotePath>> {
+        self.attempt(RemoteOp::List)?;
+        self.inner.list().await
+    }
+
    async fn list_prefixes(
        &self,
        prefix: Option<&RemotePath>,
--- a/libs/utils/src/serde_percent.rs
+++ b/libs/utils/src/serde_percent.rs
@@ -11,14 +11,6 @@ use serde::{Deserialize, Serialize};
 pub struct Percent(#[serde(deserialize_with = "deserialize_pct_0_to_100")] u8);

 impl Percent {
-    pub const fn new(pct: u8) -> Option<Self> {
-        if pct <= 100 {
-            Some(Percent(pct))
-        } else {
-            None
-        }
-    }
-
    pub fn get(&self) -> u8 {
        self.0
    }
--- a/pageserver/src/disk_usage_eviction_task.rs
+++ b/pageserver/src/disk_usage_eviction_task.rs
@@ -639,7 +639,7 @@ mod filesystem_level_usage {
                ),
                (
                    "max_usage_pct",
-                    usage_pct >= self.config.max_usage_pct.get() as u64,
+                    usage_pct > self.config.max_usage_pct.get() as u64,
                ),
            ];

@@ -686,43 +686,4 @@ mod filesystem_level_usage {
            avail_bytes,
        })
    }
-
-    #[test]
-    fn max_usage_pct_pressure() {
-        use super::Usage as _;
-        use std::time::Duration;
-        use utils::serde_percent::Percent;
-
-        let mut usage = Usage {
-            config: &DiskUsageEvictionTaskConfig {
-                max_usage_pct: Percent::new(85).unwrap(),
-                min_avail_bytes: 0,
-                period: Duration::MAX,
-                #[cfg(feature = "testing")]
-                mock_statvfs: None,
-            },
-            total_bytes: 100_000,
-            avail_bytes: 0,
-        };
-
-        assert!(usage.has_pressure(), "expected pressure at 100%");
-
-        usage.add_available_bytes(14_000);
-        assert!(usage.has_pressure(), "expected pressure at 86%");
-
-        usage.add_available_bytes(999);
-        assert!(usage.has_pressure(), "expected pressure at 85.001%");
-
-        usage.add_available_bytes(1);
-        assert!(usage.has_pressure(), "expected pressure at precisely 85%");
-
-        usage.add_available_bytes(1);
-        assert!(!usage.has_pressure(), "no pressure at 84.999%");
-
-        usage.add_available_bytes(999);
-        assert!(!usage.has_pressure(), "no pressure at 84%");
-
-        usage.add_available_bytes(16_000);
-        assert!(!usage.has_pressure());
-    }
 }
--- a/poetry.lock
+++ b/poetry.lock
@@ -1,4 +1,4 @@
-# This file is automatically @generated by Poetry 1.4.1 and should not be changed by hand.
+# This file is automatically @generated by Poetry 1.4.0 and should not be changed by hand.

 [[package]]
 name = "aiohttp"
@@ -79,35 +79,37 @@ sa = ["sqlalchemy[postgresql-psycopg2binary] (>=1.3,<1.5)"]

 [[package]]
 name = "allure-pytest"
-version = "2.13.1"
+version = "2.10.0"
 description = "Allure pytest integration"
 category = "main"
 optional = false
 python-versions = "*"
 files = [
-    {file = "allure-pytest-2.13.1.tar.gz", hash = "sha256:68d69456eeb65af4061ec06a80bc941163b0616e8216554d36b070a6bf070e08"},
-    {file = "allure_pytest-2.13.1-py3-none-any.whl", hash = "sha256:a8de2fc3b3effe2d8f98801646920de3f055b779710f4c806dbee7c613c24633"},
+    {file = "allure-pytest-2.10.0.tar.gz", hash = "sha256:3b2ab67629f4cbd8617abd817d2b22292c6eb7efd5584f992d1af8143aea6ee7"},
+    {file = "allure_pytest-2.10.0-py3-none-any.whl", hash = "sha256:08274096594758447db54c3b2c382526ee04f1fe12119cdaee92d2d93c84b530"},
 ]

 [package.dependencies]
-allure-python-commons = "2.13.1"
+allure-python-commons = "2.10.0"
 pytest = ">=4.5.0"
+six = ">=1.9.0"

 [[package]]
 name = "allure-python-commons"
-version = "2.13.1"
+version = "2.10.0"
 description = "Common module for integrate allure with python-based frameworks"
 category = "main"
 optional = false
-python-versions = ">=3.6"
+python-versions = ">=3.5"
 files = [
-    {file = "allure-python-commons-2.13.1.tar.gz", hash = "sha256:3fc13e1da8ebb23f9ab5c9c72ad04595023cdd5078dbb8604939997faebed5cb"},
-    {file = "allure_python_commons-2.13.1-py3-none-any.whl", hash = "sha256:d08e04867bddf44fef55def3d67f4bc25af58a1bf9fcffcf4ec3331f7f2ef0d0"},
+    {file = "allure-python-commons-2.10.0.tar.gz", hash = "sha256:d4d31344b0f0037a4a11e16b91b28cf0eeb23ffa0e50c27fcfc6aabe72212d3c"},
+    {file = "allure_python_commons-2.10.0-py3-none-any.whl", hash = "sha256:2a717e8ca8d296bf89cd57f38fc3c21893bd7ea8cd02a6ae5420e6d1a6eda5d0"},
 ]

 [package.dependencies]
 attrs = ">=16.0.0"
 pluggy = ">=0.4.0"
+six = ">=1.9.0"

 [[package]]
 name = "async-timeout"
@@ -1930,22 +1932,6 @@ pytest = [
    {version = ">=6.2.4", markers = "python_version >= \"3.10\""},
 ]

-[[package]]
-name = "pytest-rerunfailures"
-version = "11.1.2"
-description = "pytest plugin to re-run tests to eliminate flaky failures"
-category = "main"
-optional = false
-python-versions = ">=3.7"
-files = [
-    {file = "pytest-rerunfailures-11.1.2.tar.gz", hash = "sha256:55611661e873f1cafa384c82f08d07883954f4b76435f4b8a5b470c1954573de"},
-    {file = "pytest_rerunfailures-11.1.2-py3-none-any.whl", hash = "sha256:d21fe2e46d9774f8ad95f1aa799544ae95cac3a223477af94aa985adfae92b7e"},
-]
-
-[package.dependencies]
-packaging = ">=17.1"
-pytest = ">=5.3"
-
 [[package]]
 name = "pytest-timeout"
 version = "2.1.0"
@@ -2611,4 +2597,4 @@ testing = ["func-timeout", "jaraco.itertools", "pytest (>=6)", "pytest-black (>=
 [metadata]
 lock-version = "2.0"
 python-versions = "^3.9"
-content-hash = "b689ffd6eae32b966f1744b5ac3343fe0dd26b31ee1f50e13daf5045ee0623e1"
+content-hash = "2515a9320c2960076012fbc036fb33c4f6a23515c8d143785931dc18c6722d91"
--- a/proxy/Cargo.toml
+++ b/proxy/Cargo.toml
@@ -16,6 +16,7 @@ clap.workspace = true
 consumption_metrics.workspace = true
 futures.workspace = true
 git-version.workspace = true
+globset.workspace = true
 hashbrown.workspace = true
 hashlink.workspace = true
 hex.workspace = true
@@ -44,6 +45,7 @@ rustls-pemfile.workspace = true
 rustls.workspace = true
 scopeguard.workspace = true
 serde.workspace = true
+serde_dhall.workspace = true
 serde_json.workspace = true
 sha2.workspace = true
 socket2.workspace = true
--- a/test_runner/fixtures/pageserver/init.py
+++ b/test_runner/fixtures/pageserver/init.py
--- a/proxy/config/foo.bar.localhost/server.crt
+++ b/proxy/config/foo.bar.localhost/server.crt
@@ -0,0 +1,77 @@
+Certificate:
+    Data:
+        Version: 3 (0x2)
+        Serial Number:
+            67:2b:fc:80:27:9f:65:dd:42:d7:ef:a8:0a:fe:bd:d1:a8:2d:c8:da
+        Signature Algorithm: sha256WithRSAEncryption
+        Issuer: CN = *.foo.bar.localhost
+        Validity
+            Not Before: Mar 30 12:39:55 2023 GMT
+            Not After : Sep  3 12:39:55 2202 GMT
+        Subject: CN = *.foo.bar.localhost
+        Subject Public Key Info:
+            Public Key Algorithm: rsaEncryption
+                Public-Key: (2048 bit)
+                Modulus:
+                    00:a3:67:45:c1:97:47:83:a1:1b:34:6a:a5:fa:1a:
+                    0c:d7:b9:4e:ef:bd:03:8f:64:bf:e3:ca:51:d8:22:
+                    1e:8b:52:71:09:4e:e3:43:2f:92:45:ea:61:86:06:
+                    fe:49:23:c4:18:a7:ef:4c:81:77:8d:ce:a5:1b:80:
+                    ad:b0:d1:19:71:68:9e:b7:53:6e:d4:9f:d7:ff:d9:
+                    c0:7a:92:8e:04:e9:2b:a4:df:b2:e4:a8:ae:28:da:
+                    c8:5a:f2:d0:b6:98:e3:c4:2d:3a:c7:c3:07:b6:32:
+                    15:0d:f9:e2:05:77:32:b6:d7:e3:64:b5:8c:c0:83:
+                    32:25:7d:7f:ad:88:39:25:68:3f:0f:48:4d:60:67:
+                    b9:47:ad:bd:6d:93:73:5c:78:41:d7:db:fa:e9:bf:
+                    6b:9a:6b:e0:66:c6:90:3c:da:fb:85:2c:45:32:6c:
+                    0f:18:66:6e:42:f7:0f:93:35:4f:3e:d1:1f:a8:fb:
+                    18:75:87:19:9a:3a:af:28:28:73:45:9a:87:89:b2:
+                    a6:33:1b:25:83:69:9e:75:8c:06:d6:f3:2a:b2:bc:
+                    52:64:27:8d:ee:ec:50:88:28:5c:86:6d:8a:92:50:
+                    00:10:dd:08:42:7c:0d:5a:f8:2b:a2:d6:df:23:0d:
+                    5f:a8:da:c6:ce:d5:c9:f6:10:a4:de:62:0c:9b:29:
+                    ca:af
+                Exponent: 65537 (0x10001)
+        X509v3 extensions:
+            X509v3 Subject Key Identifier: 
+                FE:74:48:82:19:2C:85:19:EB:55:37:8A:70:DF:94:2C:FA:6B:A9:6B
+            X509v3 Authority Key Identifier: 
+                FE:74:48:82:19:2C:85:19:EB:55:37:8A:70:DF:94:2C:FA:6B:A9:6B
+            X509v3 Basic Constraints: critical
+                CA:TRUE
+    Signature Algorithm: sha256WithRSAEncryption
+    Signature Value:
+        08:2b:2a:bc:b7:0f:3e:ec:96:b0:78:87:12:a5:f1:d7:0e:96:
+        08:65:34:7d:0a:9b:3d:bd:df:dc:de:3f:5e:0f:57:75:75:a3:
+        fd:a2:68:29:80:5b:ce:1a:ea:0f:7b:96:ab:ea:92:d0:da:07:
+        61:28:32:71:37:d1:3c:6a:ca:e8:f9:9b:80:49:c5:16:40:cc:
+        8d:50:8c:4f:4e:6d:73:7b:3a:55:6b:11:76:e3:68:fd:6f:a0:
+        c8:06:f9:a1:af:28:4d:b9:c5:0d:fd:2c:98:61:7d:22:b6:87:
+        43:2b:62:fc:25:9e:fb:f4:09:24:c1:3c:7f:c7:e8:04:b4:c5:
+        5a:4b:4e:17:5d:7f:38:f1:d4:35:0c:82:bf:20:46:c7:f4:96:
+        8f:12:94:c4:ee:92:e0:5d:09:45:de:a1:40:e5:b4:34:2f:11:
+        fe:72:5f:81:a5:11:24:a5:04:98:e5:07:59:dc:d8:dc:7b:f6:
+        12:ba:8b:d3:cf:dd:de:06:84:23:e7:b3:29:b2:8f:b1:6b:c3:
+        71:ee:da:bc:9e:b5:62:a6:68:cb:ea:49:19:34:6c:29:be:ce:
+        6d:3b:7a:59:28:59:67:83:e9:6d:37:06:fd:29:f7:ce:fc:fc:
+        72:de:23:f5:2b:f6:dc:d2:82:3e:45:bb:e1:ce:14:d7:85:d5:
+        ec:3b:1c:3c
+-----BEGIN CERTIFICATE-----
+MIIDHzCCAgegAwIBAgIUZyv8gCefZd1C1++oCv690agtyNowDQYJKoZIhvcNAQEL
+BQAwHjEcMBoGA1UEAwwTKi5mb28uYmFyLmxvY2FsaG9zdDAgFw0yMzAzMzAxMjM5
+NTVaGA8yMjAyMDkwMzEyMzk1NVowHjEcMBoGA1UEAwwTKi5mb28uYmFyLmxvY2Fs
+aG9zdDCCASIwDQYJKoZIhvcNAQEBBQADggEPADCCAQoCggEBAKNnRcGXR4OhGzRq
+pfoaDNe5Tu+9A49kv+PKUdgiHotScQlO40MvkkXqYYYG/kkjxBin70yBd43OpRuA
+rbDRGXFonrdTbtSf1//ZwHqSjgTpK6TfsuSorijayFry0LaY48QtOsfDB7YyFQ35
+4gV3MrbX42S1jMCDMiV9f62IOSVoPw9ITWBnuUetvW2Tc1x4Qdfb+um/a5pr4GbG
+kDza+4UsRTJsDxhmbkL3D5M1Tz7RH6j7GHWHGZo6rygoc0Wah4mypjMbJYNpnnWM
+BtbzKrK8UmQnje7sUIgoXIZtipJQABDdCEJ8DVr4K6LW3yMNX6jaxs7VyfYQpN5i
+DJspyq8CAwEAAaNTMFEwHQYDVR0OBBYEFP50SIIZLIUZ61U3inDflCz6a6lrMB8G
+A1UdIwQYMBaAFP50SIIZLIUZ61U3inDflCz6a6lrMA8GA1UdEwEB/wQFMAMBAf8w
+DQYJKoZIhvcNAQELBQADggEBAAgrKry3Dz7slrB4hxKl8dcOlghlNH0Kmz2939ze
+P14PV3V1o/2iaCmAW84a6g97lqvqktDaB2EoMnE30Txqyuj5m4BJxRZAzI1QjE9O
+bXN7OlVrEXbjaP1voMgG+aGvKE25xQ39LJhhfSK2h0MrYvwlnvv0CSTBPH/H6AS0
+xVpLThddfzjx1DUMgr8gRsf0lo8SlMTukuBdCUXeoUDltDQvEf5yX4GlESSlBJjl
+B1nc2Nx79hK6i9PP3d4GhCPnsymyj7Frw3Hu2ryetWKmaMvqSRk0bCm+zm07elko
+WWeD6W03Bv0p9878/HLeI/Ur9tzSgj5Fu+HOFNeF1ew7HDw=
+-----END CERTIFICATE-----
--- a/proxy/config/foo.bar.localhost/server.dhall
+++ b/proxy/config/foo.bar.localhost/server.dhall
@@ -0,0 +1,5 @@
+{
+    server_name = "*.foo.bar.localhost",
+    certificate = ./server.crt as Text,
+    private_key = ./server.key as Text,
+}
--- a/proxy/config/foo.bar.localhost/server.key
+++ b/proxy/config/foo.bar.localhost/server.key
@@ -0,0 +1,28 @@
+-----BEGIN PRIVATE KEY-----
+MIIEvgIBADANBgkqhkiG9w0BAQEFAASCBKgwggSkAgEAAoIBAQCjZ0XBl0eDoRs0
+aqX6GgzXuU7vvQOPZL/jylHYIh6LUnEJTuNDL5JF6mGGBv5JI8QYp+9MgXeNzqUb
+gK2w0RlxaJ63U27Un9f/2cB6ko4E6Suk37LkqK4o2sha8tC2mOPELTrHwwe2MhUN
+eIFdzK21+NktYzAgzIlfX+tiDklaD8PSE1gZ7lHrb1tk3NceEHX2/rpv2uaa+Bm
+xpA82vuFLEUybA8YZm5C9w+TNU8+0R+o+xh1hxmaOq8oKHNFmoeJsqYzGyWDaZ51
+jAbW8yqyvFJkJ43u7FCIKFyGbYqSUAAQ3QhCfA1a+Cui1t8jDV+o2sbO1cn2EKTe
+YgybKcqvAgMBAAECggEACcd9ZwqPzcKxyXiCgSmOYn53kkmjaIzCjKQaV/z+wkNi
+6FFY1UoH6oqOX3lLPjoOP92COy6RQjUtMAfT1Cu1L8BLE2uLgt0jjgGVs+lkMKYM
+pfwNDXD6pjQBOhjHrxcO7XDL0JJWcVCBAMp76qMb1D+u+poSqg1rcqrNeZVO3s6n
+ot4aejb+hDEB8t4ytpDnqULaPvnDmlc5WvVpS8qbGTRiq8DSe5RzEQZp7WHR9Una
+0o/rQPu+RiH2+7CesQugg9bc3iaW83DbJjLuODU/4Au/aU/7I0giV8T89UIvtXDl
+JzyVDRJNr1qPabZ/H2CFRkfHKovNRpJDEX48VuSR3QKBgQDmr61HA6rRh6zmmehd
+V1mroKb9UCR/edBklBSsO2kvWMHPtFEwlszebydcrJa2GwbE37NjKZwkEp1Q/mIk
+yX2mjd629j62vzjOejFrW22YOX2LN2cK2Ns9c5T0qnX7ttxy8lw3dE7uU9Y42X8E
+Wxm5yIiXImgxIaxCpceR4lzdqwKBgQC1VYUSZcxXIRNjyd98vcD7Ai3zTai4wO3k
+WC+Myu2DZr1ZhmY2N0M5b5UUsTL6yOt9fKgqMz3Ww21/zweXx91ts6XmygXIj5ZA
+7TyQxPE/PGzG6Olg5o00L//68P9jSBz9HgRWdRtxyJ8NWOUTTGpqA5dKjEj7tia/
+E/QcErSbDQKBgAgNug7wodYO1op2dRZNJmRHh4zwb1XD+vKH+PDKYjG09484zFzV
+5vEdEFK788cHyoS1Cp47pafcvoFFYEfIgQp/iXb5wda/dkw/F9qXpovZ9fgWRxKp
+332Vu22PRe8zwx6AN5f4B4lqg+AYN8b/JzbFOX+NQ/XzJwBsqTr+nB9hAoGBAIch
+McdiAQK07UQhvd+xcEwddayoJKF5dE4DwXuEBbc0Ksq6MxUX3YrBsjD3U+w7KfIb
+oR3BjcWrYMArwZbEJCiKBYmU5vZsuiWsJMQlXzomh1E7ZB8H8BYB5xpT2Z1cse2W
+Htlm74q9XHmP0zWsbmiOQIIXRJP/S6R89B6vedNJAoGBAJOjRfwepnFpIVXG0waS
+S1zavMQZvDsJcvNhedyi34ui5XXi79w+uc13xoFTilhG8DS+tOnw/LsDTP8NeLKv
+yENcF/zSHuC4GJJjxoJ+SPlaW+mlofsoCT7zRGRVG27xS6jOwo5/7fYg0aELsq2o
+oS/TVaWFJKVtJW64vHdGzMnx
+-----END PRIVATE KEY-----
--- a/proxy/config/neon.localhost/server.crt
+++ b/proxy/config/neon.localhost/server.crt
@@ -0,0 +1,77 @@
+Certificate:
+    Data:
+        Version: 3 (0x2)
+        Serial Number:
+            1b:91:d6:a3:a3:1b:c1:47:b3:76:16:a2:43:64:3b:63:61:0d:0c:c8
+        Signature Algorithm: sha256WithRSAEncryption
+        Issuer: CN = *.neon.localhost
+        Validity
+            Not Before: Mar 30 13:19:00 2023 GMT
+            Not After : Sep  3 13:19:00 2202 GMT
+        Subject: CN = *.neon.localhost
+        Subject Public Key Info:
+            Public Key Algorithm: rsaEncryption
+                Public-Key: (2048 bit)
+                Modulus:
+                    00:ab:27:17:8b:71:a5:08:65:ec:46:58:69:3d:7d:
+                    b7:91:34:33:3a:50:5c:27:91:34:70:fe:2c:81:fc:
+                    aa:f2:32:60:b3:49:ca:c1:c8:fb:76:73:97:ac:e9:
+                    ff:20:4a:73:48:62:fa:65:d3:81:65:1b:cc:78:00:
+                    6d:a8:32:8b:ff:61:3c:8b:a2:23:61:de:e3:b4:5f:
+                    cd:ce:7e:ce:05:49:5b:64:81:42:74:81:49:4a:5c:
+                    3a:f0:d9:40:2d:42:7d:ef:db:b9:d3:54:a5:11:52:
+                    0b:23:65:fd:c1:5e:58:f7:98:9c:ba:3d:9a:f9:a2:
+                    50:0c:c4:c8:bf:63:66:aa:e3:29:fe:40:ae:13:8c:
+                    18:af:20:24:41:63:3e:1d:af:cd:91:32:fa:b9:26:
+                    96:fb:35:12:1c:62:58:93:33:d2:2c:76:08:f9:c4:
+                    07:ef:fc:6f:eb:ea:c8:81:5d:d8:73:0a:05:47:79:
+                    52:b3:24:34:08:c3:b6:a0:aa:af:de:8b:62:f9:6f:
+                    3a:8f:eb:07:85:ef:2b:de:f7:21:69:7a:63:17:27:
+                    4a:88:b5:4c:e7:52:73:09:10:c6:ca:eb:f0:c8:31:
+                    cc:d4:59:d6:64:82:ac:ae:96:69:18:5e:19:17:5d:
+                    9b:86:39:36:a1:9f:90:34:45:73:9b:43:a2:b0:d3:
+                    8b:e1
+                Exponent: 65537 (0x10001)
+        X509v3 extensions:
+            X509v3 Subject Key Identifier: 
+                E3:8E:E8:BF:5E:72:36:0A:6D:7F:BE:AF:68:38:A7:CE:71:15:CE:14
+            X509v3 Authority Key Identifier: 
+                E3:8E:E8:BF:5E:72:36:0A:6D:7F:BE:AF:68:38:A7:CE:71:15:CE:14
+            X509v3 Basic Constraints: critical
+                CA:TRUE
+    Signature Algorithm: sha256WithRSAEncryption
+    Signature Value:
+        69:23:54:9d:eb:4c:57:f5:cf:8d:5c:d3:f8:2e:65:0b:f1:b8:
+        8e:a8:ef:67:dd:89:96:8d:df:f1:9b:36:3a:49:0d:55:dc:07:
+        eb:c5:e7:c8:73:12:02:6a:02:d3:ca:92:63:f0:61:4b:8a:2a:
+        32:c2:23:c2:53:ee:33:6b:eb:9f:e5:f4:df:78:55:bf:d5:86:
+        1e:e5:a8:90:e2:df:d7:c2:b8:63:27:a7:2d:ba:43:34:d3:45:
+        e0:94:53:0a:26:fb:66:ac:c3:96:76:c3:45:a1:ae:d6:30:e0:
+        b6:c0:a6:d1:8e:51:c7:56:fb:ed:5c:04:a2:66:b9:74:c6:6d:
+        ef:1e:9e:9a:58:7b:fc:e0:1c:94:fc:17:df:5b:70:e7:cd:f9:
+        22:49:3d:59:83:8e:c3:bf:bc:3b:39:68:9e:5a:34:88:1a:61:
+        f7:53:ac:86:de:76:85:75:f6:b7:86:3f:20:4b:98:63:97:03:
+        8b:29:37:32:2c:c1:9a:65:a2:58:17:f2:7b:79:e7:ee:6a:33:
+        5b:d0:bd:af:04:dd:02:43:98:a7:e9:0f:35:cb:c0:9d:a6:95:
+        bf:98:57:4d:cf:b8:a9:bb:de:0c:4d:51:93:df:62:f6:20:bf:
+        61:27:7d:2c:be:14:48:5d:d1:75:f9:cb:d9:b3:0a:2b:de:ea:
+        2a:4b:9e:c0
+-----BEGIN CERTIFICATE-----
+MIIDGTCCAgGgAwIBAgIUG5HWo6MbwUezdhaiQ2Q7Y2ENDMgwDQYJKoZIhvcNAQEL
+BQAwGzEZMBcGA1UEAwwQKi5uZW9uLmxvY2FsaG9zdDAgFw0yMzAzMzAxMzE5MDBa
+GA8yMjAyMDkwMzEzMTkwMFowGzEZMBcGA1UEAwwQKi5uZW9uLmxvY2FsaG9zdDCC
+ASIwDQYJKoZIhvcNAQEBBQADggEPADCCAQoCggEBAKsnF4txpQhl7EZYaT19t5E0
+MzpQXCeRNHD+LIH8qvIyYLNJysHI+3Zzl6zp/yBKc0hi+mXTgWUbzHgAbagyi/9h
+PIuiI2He47Rfzc5+zgVJW2SBQnSBSUpcOvDZQC1Cfe/budNUpRFSCyNl/cFeWPeY
+nLo9mvmiUAzEyL9jZqrjKf5ArhOMGK8gJEFjPh2vzZEy+rkmlvs1EhxiWJMz0ix2
+CPnEB+/8b+vqyIFd2HMKBUd5UrMkNAjDtqCqr96LYvlvOo/rB4XvK973IWl6Yxcn
+Soi1TOdScwkQxsrr8MgxzNRZ1mSCrK6WaRheGRddm4Y5NqGfkDRFc5tDorDTi+EC
+AwEAAaNTMFEwHQYDVR0OBBYEFOOO6L9ecjYKbX++r2g4p85xFc4UMB8GA1UdIwQY
+MBaAFOOO6L9ecjYKbX++r2g4p85xFc4UMA8GA1UdEwEB/wQFMAMBAf8wDQYJKoZI
+hvcNAQELBQADggEBAGkjVJ3rTFf1z41c0/guZQvxuI6o72fdiZaN3/GbNjpJDVXc
+B+vF58hzEgJqAtPKkmPwYUuKKjLCI8JT7jNr65/l9N94Vb/Vhh7lqJDi39fCuGMn
+py26QzTTReCUUwom+2asw5Z2w0WhrtYw4LbAptGOUcdW++1cBKJmuXTGbe8enppY
+e/zgHJT8F99bcOfN+SJJPVmDjsO/vDs5aJ5aNIgaYfdTrIbedoV19reGPyBLmGOX
+A4spNzIswZplolgX8nt55+5qM1vQva8E3QJDmKfpDzXLwJ2mlb+YV03PuKm73gxN
+UZPfYvYgv2EnfSy+FEhd0XX5y9mzCive6ipLnsA=
+-----END CERTIFICATE-----
--- a/proxy/config/neon.localhost/server.dhall
+++ b/proxy/config/neon.localhost/server.dhall
@@ -0,0 +1,5 @@
+{
+    server_name = "*.neon.localhost",
+    certificate = ./server.crt as Text,
+    private_key = ./server.key as Text,
+}
--- a/proxy/config/neon.localhost/server.key
+++ b/proxy/config/neon.localhost/server.key
@@ -0,0 +1,28 @@
+-----BEGIN PRIVATE KEY-----
+MIIEvgIBADANBgkqhkiG9w0BAQEFAASCBKgwggSkAgEAAoIBAQCrJxeLcaUIZexG
+WGk9fbeRNDM6UFwnkTRw/iyB/KryMmCzScrByPt2c5es6f8gSnNIYvpl04FlG8x4
+AG2oMov/YTyLoiNh3uO0X83Ofs4FSVtkgUJ0gUlKXDrw2UAtQn3v27nTVKURUgsj
+Zf3BXlj3mJy6PZr5olAMxMi/Y2aq4yn+QK4TjBivICRBYz4dr82RMvq5Jpb7NRIc
+YliTM9Isdgj5xAfv/G/r6siBXdhzCgVHeVKzJDQIw7agqq/ei2L5bzqP6weF7yve
+9yFpemMXJ0qItUznUnMJEMbK6/DIMczUWdZkgqyulmkYXhkXXZuGOTahn5A0RXOb
+Q6Kw04vhAgMBAAECggEADrLZw8urwlqL0x1uqa2X9pbGu4BWmb36Wfs03qj7aWId
+ieg8IIOz4jVagQAg5/4Hg9+e5OB9jAMPfqgoGA+B6cRzta45XwNhsjD0H4LRC1qE
+cyTityy58EfsIUPBzjaX/Yx08LWj7iaJ9wKgVgYAmqr28suto+NmVTe6jIKV46EL
+bWmnU0dySOa43ukdhkvQN+FG3hL4iIl+mZ5aTVY8dz885sxYdrKOyrzMREAvHFW6
+m01fWwgbdiMfR2Gu2ZWmvom4+PiE8EES8/Cpct4/E27SFLr3pdB+voIBSh6kotF9
+w0dNqnK1dyIC89gxhcH/PO4rC6uKPM68ZezBsqHZ3QKBgQDMB54bMbcVNK/92nRV
+xtM8sk567oAeDwL7VcMq35vmwZU1OcjPg/QswIIZNIx46SXO8a668Wn6OLZYq4dR
+FGBWpsMHbQSEdyurYY2bqi5tK5dnKiuCqNmTTtxPA2QgC+PdDcTlJ2FI/RmCeODM
+GUcKJd0FyR5BNy8TPM31kFs7zwKBgQDWv6A4cpKzzfY7hf/iOcgmgyipPfXBAdCE
+6w6HAEU5JKJDtxIC3roosxOVbrqCMGqPWKCkQvzOr7d9Ok5fYk4WF+qM67xxdzHa
+KzmE2+PKDsWBejxjnIMBtDKBkWhOU5/bg/HLDv4RPNwn7f0MjxFpxzYuq3q/dIPN
+TcZthbU5TwKBgQCJj1FAEILZ304RH2p0MrtVHvre01K58XEXN7mAfIbGTBpnanBD
+yTmlup18lPtowfjlz/j4va+wLvByVCPFvLE/euvfY9c54Icm43zwSQtIO622tq3j
+SCh5sx/CfgzRtnKJJbFstuJWrZ63YvxdX2WQJ/se3Xxyh9xLYiGSwSNh7QKBgDFO
+/rL3W8f9WrSAKCkBq3tsUkHKAEu45vAeKM/GuB5O0xNJTdFq4sPFmpGNQzXxeAZC
+C2CsIPA0WKVgZe5w3A0moKyK1FIZVFEL68Ed3Efg7Gi2cHdO0KXrgk1N3e1eNi5p
+NXOylZPPrZ1df+UKVK09GKvOo/iiAEF7wjwTn3DxAoGBAK1iTDV+HrGyxCi2PcoK
+yyFCB2QEFw5vvMMMu5lvRjaI9r+igEg1Y8DWhXpUb0hsXnTV192dwhqmB+NYn7Yz
+xwlFeKolv+j+5H+IQ4vmEleOlaBLGBH/lAdCcJ0bGcKHHRP+chF6En2tOTk45Gai
+4gsDafbyi89fJ/5EoLGRYMe/
+-----END PRIVATE KEY-----
--- a/proxy/config/servers.dhall
+++ b/proxy/config/servers.dhall
@@ -0,0 +1,7 @@
+let Server = { server_name : Text, certificate : Text, private_key : Text }
+
+let servers
+    : List Server
+    = [ ./foo.bar.localhost/server.dhall, ./neon.localhost/server.dhall ]
+
+in  servers
--- a/proxy/src/auth/backend/hacks.rs
+++ b/proxy/src/auth/backend/hacks.rs
@@ -53,7 +53,7 @@ pub async fn password_hack(
        .await?;

    info!(project = &payload.project, "received missing parameter");
-    creds.project = Some(payload.project);
+    creds.project = Some(payload.project.into());

    let mut node = api.wake_compute(extra, creds).await?;
    node.config.password(payload.password);
--- a/proxy/src/auth/credentials.rs
+++ b/proxy/src/auth/credentials.rs
@@ -2,7 +2,7 @@

 use crate::error::UserFacingError;
 use pq_proto::StartupMessageParams;
-use std::collections::HashSet;
+use std::borrow::Cow;
 use thiserror::Error;
 use tracing::info;

@@ -18,12 +18,6 @@ pub enum ClientCredsParseError {
    )]
    InconsistentProjectNames { domain: String, option: String },

-    #[error(
-        "Common name inferred from SNI ('{}') is not known",
-        .cn,
-    )]
-    UnknownCommonName { cn: String },
-
    #[error("Project name ('{0}') must contain only alphanumeric characters and hyphen.")]
    MalformedProjectName(String),
 }
@@ -36,7 +30,7 @@ impl UserFacingError for ClientCredsParseError {}
 pub struct ClientCredentials<'a> {
    pub user: &'a str,
    // TODO: this is a severe misnomer! We should think of a new name ASAP.
-    pub project: Option<String>,
+    pub project: Option<Cow<'a, str>>,
 }

 impl ClientCredentials<'_> {
@@ -50,7 +44,6 @@ impl<'a> ClientCredentials<'a> {
    pub fn parse(
        params: &'a StartupMessageParams,
        sni: Option<&str>,
-        common_names: Option<HashSet<String>>,
    ) -> Result<Self, ClientCredsParseError> {
        use ClientCredsParseError::*;

@@ -59,43 +52,29 @@ impl<'a> ClientCredentials<'a> {
        let user = get_param("user")?;

        // Project name might be passed via PG's command-line options.
-        let project_option = params
-            .options_raw()
-            .and_then(|mut options| options.find_map(|opt| opt.strip_prefix("project=")))
-            .map(|name| name.to_string());
+        let project_option = params.options_raw().and_then(|mut options| {
+            options
+                .find_map(|opt| opt.strip_prefix("project="))
+                .map(Cow::Borrowed)
+        });

-        let project_from_domain = if let Some(sni_str) = sni {
-            if let Some(cn) = common_names {
-                let common_name_from_sni = sni_str.split_once('.').map(|(_, domain)| domain);
+        // Alternative project name is in fact a subdomain from SNI.
+        let project_domain = sni.and_then(|sni| {
+            let (domain, _) = sni.split_once('.')?;
+            Some(Cow::from(domain.to_owned()))
+        });

-                let project = common_name_from_sni
-                    .and_then(|domain| {
-                        if cn.contains(domain) {
-                            subdomain_from_sni(sni_str, domain)
-                        } else {
-                            None
-                        }
-                    })
-                    .ok_or_else(|| UnknownCommonName {
-                        cn: common_name_from_sni.unwrap_or("").into(),
-                    })?;
-
-                Some(project)
-            } else {
-                None
-            }
-        } else {
-            None
-        };
-
-        let project = match (project_option, project_from_domain) {
+        let project = match (project_option, project_domain) {
            // Invariant: if we have both project name variants, they should match.
            (Some(option), Some(domain)) if option != domain => {
-                Some(Err(InconsistentProjectNames { domain, option }))
+                Some(Err(InconsistentProjectNames {
+                    domain: domain.into(),
+                    option: option.into(),
+                }))
            }
            // Invariant: project name may not contain certain characters.
            (a, b) => a.or(b).map(|name| match project_name_valid(&name) {
-                false => Err(MalformedProjectName(name)),
+                false => Err(MalformedProjectName(name.into())),
                true => Ok(name),
            }),
        }
@@ -111,12 +90,6 @@ fn project_name_valid(name: &str) -> bool {
    name.chars().all(|c| c.is_alphanumeric() || c == '-')
 }

-fn subdomain_from_sni(sni: &str, common_name: &str) -> Option<String> {
-    sni.strip_suffix(common_name)?
-        .strip_suffix('.')
-        .map(str::to_owned)
-}
-
 #[cfg(test)]
 mod tests {
    use super::*;
@@ -127,7 +100,7 @@ mod tests {
        // According to postgresql, only `user` should be required.
        let options = StartupMessageParams::new([("user", "john_doe")]);

-        let creds = ClientCredentials::parse(&options, None, None)?;
+        let creds = ClientCredentials::parse(&options, None)?;
        assert_eq!(creds.user, "john_doe");
        assert_eq!(creds.project, None);

@@ -142,7 +115,7 @@ mod tests {
            ("foo", "bar"),        // should be ignored
        ]);

-        let creds = ClientCredentials::parse(&options, None, None)?;
+        let creds = ClientCredentials::parse(&options, None)?;
        assert_eq!(creds.user, "john_doe");
        assert_eq!(creds.project, None);

@@ -154,9 +127,8 @@ mod tests {
        let options = StartupMessageParams::new([("user", "john_doe")]);

        let sni = Some("foo.localhost");
-        let common_names = Some(["localhost".into()].into());

-        let creds = ClientCredentials::parse(&options, sni, common_names)?;
+        let creds = ClientCredentials::parse(&options, sni)?;
        assert_eq!(creds.user, "john_doe");
        assert_eq!(creds.project.as_deref(), Some("foo"));

@@ -170,7 +142,7 @@ mod tests {
            ("options", "-ckey=1 project=bar -c geqo=off"),
        ]);

-        let creds = ClientCredentials::parse(&options, None, None)?;
+        let creds = ClientCredentials::parse(&options, None)?;
        assert_eq!(creds.user, "john_doe");
        assert_eq!(creds.project.as_deref(), Some("bar"));

@@ -182,41 +154,22 @@ mod tests {
        let options = StartupMessageParams::new([("user", "john_doe"), ("options", "project=baz")]);

        let sni = Some("baz.localhost");
-        let common_names = Some(["localhost".into()].into());

-        let creds = ClientCredentials::parse(&options, sni, common_names)?;
+        let creds = ClientCredentials::parse(&options, sni)?;
        assert_eq!(creds.user, "john_doe");
        assert_eq!(creds.project.as_deref(), Some("baz"));

        Ok(())
    }

-    #[test]
-    fn parse_multi_common_names() -> anyhow::Result<()> {
-        let options = StartupMessageParams::new([("user", "john_doe")]);
-
-        let common_names = Some(["a.com".into(), "b.com".into()].into());
-        let sni = Some("p1.a.com");
-        let creds = ClientCredentials::parse(&options, sni, common_names)?;
-        assert_eq!(creds.project.as_deref(), Some("p1"));
-
-        let common_names = Some(["a.com".into(), "b.com".into()].into());
-        let sni = Some("p1.b.com");
-        let creds = ClientCredentials::parse(&options, sni, common_names)?;
-        assert_eq!(creds.project.as_deref(), Some("p1"));
-
-        Ok(())
-    }
-
    #[test]
    fn parse_projects_different() {
        let options =
            StartupMessageParams::new([("user", "john_doe"), ("options", "project=first")]);

        let sni = Some("second.localhost");
-        let common_names = Some(["localhost".into()].into());

-        let err = ClientCredentials::parse(&options, sni, common_names).expect_err("should fail");
+        let err = ClientCredentials::parse(&options, sni).expect_err("should fail");
        match err {
            InconsistentProjectNames { domain, option } => {
                assert_eq!(option, "first");
@@ -225,20 +178,4 @@ mod tests {
            _ => panic!("bad error: {err:?}"),
        }
    }
-
-    #[test]
-    fn parse_inconsistent_sni() {
-        let options = StartupMessageParams::new([("user", "john_doe")]);
-
-        let sni = Some("project.localhost");
-        let common_names = Some(["example.com".into()].into());
-
-        let err = ClientCredentials::parse(&options, sni, common_names).expect_err("should fail");
-        match err {
-            UnknownCommonName { cn } => {
-                assert_eq!(cn, "localhost");
-            }
-            _ => panic!("bad error: {err:?}"),
-        }
-    }
 }
--- a/proxy/src/certs.rs
+++ b/proxy/src/certs.rs
@@ -0,0 +1,312 @@
+use rustls::{
+    server::{ClientHello, ResolvesServerCert},
+    sign::CertifiedKey,
+};
+use std::{collections::BTreeMap, io, ops::Bound, sync::Arc};
+use tracing::{info, warn};
+
+/// App-level configuration structs for TLS certificates.
+pub mod config {
+    use super::*;
+    use serde::{de, Deserialize};
+    use std::path::Path;
+
+    /// Collection of TLS-related configurations of virtual proxy servers.
+    #[derive(Debug, Default, Clone, Deserialize)]
+    #[serde(transparent)]
+    pub struct TlsServers(pub Vec<TlsServer>);
+
+    impl TlsServers {
+        /// Load [`Self`] config from a file.
+        pub fn from_config_file(path: impl AsRef<Path>) -> anyhow::Result<Self> {
+            info!(path = %path.as_ref().display(), "loading TLS servers config file");
+            let config = serde_dhall::from_file(path).parse()?;
+            Ok(config)
+        }
+    }
+
+    /// This lets us merge multiple configs into one (semigroup).
+    impl FromIterator<Self> for TlsServers {
+        fn from_iter<T: IntoIterator<Item = Self>>(iter: T) -> Self {
+            Self(iter.into_iter().flat_map(|xs| xs.0).collect())
+        }
+    }
+
+    /// Helps deserialize certificate chain from a string.
+    #[derive(Debug, Clone, Deserialize)]
+    #[serde(transparent)]
+    pub struct TlsCert(
+        /// The wrapped rustls certificate.
+        #[serde(deserialize_with = "deserialize_certs")]
+        pub Vec<rustls::Certificate>,
+    );
+
+    fn deserialize_certs<'de, D>(des: D) -> Result<Vec<rustls::Certificate>, D::Error>
+    where
+        D: serde::Deserializer<'de>,
+    {
+        let text = String::deserialize(des)?;
+        parse_certs(&mut text.as_bytes()).map_err(de::Error::custom)
+    }
+
+    /// Helps deserialize private key from a string.
+    #[derive(Debug, Clone, Deserialize)]
+    #[serde(transparent)]
+    pub struct TlsKey(
+        /// The wrapped rustls private key.
+        #[serde(deserialize_with = "deserialize_key")]
+        pub rustls::PrivateKey,
+    );
+
+    fn deserialize_key<'de, D>(des: D) -> Result<rustls::PrivateKey, D::Error>
+    where
+        D: serde::Deserializer<'de>,
+    {
+        let text = String::deserialize(des)?;
+        parse_key(&mut text.as_bytes()).map_err(de::Error::custom)
+    }
+
+    /// Represents TLS config of a single virtual proxy server.
+    #[derive(Debug, Clone, Deserialize)]
+    pub struct TlsServer {
+        /// Proxy server's certificate chain.
+        pub certificate: TlsCert,
+        /// Proxy server's private key.
+        pub private_key: TlsKey,
+    }
+
+    impl TlsServer {
+        pub fn into_certified_key(
+            self,
+        ) -> Result<rustls::sign::CertifiedKey, rustls::sign::SignError> {
+            Ok(rustls::sign::CertifiedKey::new(
+                self.certificate.0,
+                rustls::sign::any_supported_type(&self.private_key.0)?,
+            ))
+        }
+    }
+}
+
+/// Parse TLS certificate chain from a byte buffer.
+fn parse_certs(buf: &mut impl io::BufRead) -> io::Result<Vec<rustls::Certificate>> {
+    let chain = rustls_pemfile::certs(buf)?
+        .into_iter()
+        .map(rustls::Certificate)
+        .collect();
+
+    Ok(chain)
+}
+
+/// Parse exactly one TLS private key from a byte buffer.
+fn parse_key(buf: &mut impl io::BufRead) -> io::Result<rustls::PrivateKey> {
+    let mut keys = rustls_pemfile::pkcs8_private_keys(buf)?;
+
+    // We expect to see only 1 key.
+    if keys.len() != 1 {
+        return Err(io::Error::new(
+            io::ErrorKind::Other,
+            "there should be exactly one TLS key in buffer",
+        ));
+    }
+
+    Ok(rustls::PrivateKey(keys.pop().unwrap()))
+}
+
+/// Extract domain names from a certificate: first CN, then SANs.
+/// Further reading: <https://www.rfc-editor.org/rfc/rfc4985>.
+fn certificate_names(cert: &rustls::Certificate) -> anyhow::Result<Vec<String>> {
+    use x509_parser::{extensions::GeneralName, x509::AttributeTypeAndValue};
+
+    let get_dns_name = |gn: &GeneralName| match gn {
+        GeneralName::DNSName(name) => Some(name.to_string()),
+        _other => None,
+    };
+
+    let get_common_name = |attr: &AttributeTypeAndValue| {
+        // There really shouldn't be anything but string here.
+        attr.attr_value().as_string().expect("bad CN attribute")
+    };
+
+    let (rest, cert) = x509_parser::parse_x509_certificate(cert.0.as_ref())?;
+    anyhow::ensure!(rest.is_empty(), "excessive bytes in DER certificate");
+
+    // Extract CN, Common Name.
+    let mut names: Vec<String> = cert
+        .subject()
+        .iter_common_name()
+        .map(get_common_name)
+        .collect();
+
+    // Now append SANs, Subject Alternative Names, if any.
+    if let Some(extension) = cert.subject_alternative_name()? {
+        let alt_names = &extension.value.general_names;
+        names.extend(alt_names.iter().filter_map(get_dns_name));
+    }
+
+    Ok(names)
+}
+
+#[derive(Debug)]
+struct GlobMapBuilder<V> {
+    builder: globset::GlobSetBuilder,
+    values: BTreeMap<usize, V>,
+}
+
+impl<V> GlobMapBuilder<V> {
+    fn new() -> Self {
+        Self {
+            builder: globset::GlobSetBuilder::new(),
+            values: Default::default(),
+        }
+    }
+
+    fn add(&mut self, globs: impl IntoIterator<Item = globset::Glob>, value: V) -> &mut Self {
+        let mut cnt = 0;
+        for glob in globs {
+            self.builder.add(glob);
+            cnt += 1;
+        }
+
+        if cnt > 0 {
+            let offset = self.values.last_key_value().map(|(k, _)| *k).unwrap_or(0);
+            self.values.insert(offset + cnt, value);
+        }
+
+        self
+    }
+
+    fn build(self) -> Result<GlobMap<V>, globset::Error> {
+        Ok(GlobMap {
+            set: self.builder.build()?,
+            values: self.values,
+        })
+    }
+}
+
+/// Maps a set of matching globs to an arbitrary value.
+/// See the tests below in case this description doesn't help.
+#[derive(Debug)]
+struct GlobMap<V> {
+    /// An ordered set of all loaded globs.
+    set: globset::GlobSet,
+    /// Store single value per range of globs.
+    values: BTreeMap<usize, V>,
+}
+
+impl<V> GlobMap<V> {
+    fn query(&self, text: &str) -> Vec<&V> {
+        let indices = self.set.matches(text);
+        let mut res = Vec::with_capacity(indices.len());
+
+        for i in indices {
+            let mut range = self.values.range((Bound::Excluded(i), Bound::Unbounded));
+            let (_, value) = range.next().expect("invariant: entry must exist");
+            res.push(value);
+        }
+
+        res
+    }
+}
+
+pub struct CertResolverEntry {
+    pub raw: Arc<rustls::sign::CertifiedKey>,
+    pub names: Vec<String>,
+}
+
+pub struct CertResolver {
+    storage: GlobMap<CertResolverEntry>,
+}
+
+impl CertResolver {
+    pub fn new(config: config::TlsServers) -> anyhow::Result<Self> {
+        let mut builder = GlobMapBuilder::new();
+        for server in config.0 {
+            let Some(cert) = server.certificate.0.first() else {
+                warn!("found empty certificate, skipping");
+                continue;
+            };
+
+            let names = certificate_names(cert)?;
+            let globs = names
+                .iter()
+                .map(|s| globset::Glob::new(s))
+                .collect::<Result<Vec<_>, _>>()?;
+
+            info!(?names, "loading TLS certificate");
+            let entry = CertResolverEntry {
+                raw: Arc::new(server.into_certified_key()?),
+                names,
+            };
+
+            builder.add(globs, entry);
+        }
+
+        Ok(Self {
+            storage: builder.build()?,
+        })
+    }
+
+    #[tracing::instrument(name = "resolve_tls_cert", fields(%server_name), skip_all)]
+    pub fn resolve_raw(&self, server_name: &str) -> Option<&CertResolverEntry> {
+        info!("trying to resolve TLS certificate");
+        let entries = self.storage.query(server_name);
+        info!("found {} matching entries", entries.len());
+        entries.first().copied()
+    }
+}
+
+impl ResolvesServerCert for CertResolver {
+    fn resolve(&self, message: ClientHello) -> Option<Arc<CertifiedKey>> {
+        let name = message.server_name()?;
+        self.resolve_raw(name).map(|entry| entry.raw.clone())
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+    use globset::Glob;
+
+    #[test]
+    fn check_glob_map_basic() -> anyhow::Result<()> {
+        let mut builder = GlobMapBuilder::new();
+        builder
+            .add([Glob::new("*.localhost")?], 0)
+            .add([Glob::new("bar.localhost")?], 1)
+            .add([Glob::new("*.foo.localhost")?], 2);
+
+        let map = builder.build()?;
+
+        assert!(map.query("random").is_empty());
+        assert!(map.query("localhost").is_empty());
+        assert_eq!(map.query("foo.localhost"), [&0]);
+        assert_eq!(map.query("bar.localhost"), [&0, &1]);
+        assert_eq!(map.query("project.foo.localhost"), [&0, &2]);
+
+        Ok(())
+    }
+
+    #[test]
+    fn check_glob_map() -> anyhow::Result<()> {
+        let mut builder = GlobMapBuilder::new();
+        builder
+            .add(
+                [
+                    Glob::new("*.neon.tech")?,
+                    Glob::new("*.neon.internal.tech")?,
+                ],
+                "neon",
+            )
+            .add([Glob::new("*.localhost")?], "mock");
+
+        let map = builder.build()?;
+
+        assert!(map.query("random").is_empty());
+        assert!(map.query("localhost").is_empty());
+        assert_eq!(map.query("ep-1.neon.tech"), [&"neon"]);
+        assert_eq!(map.query("ep-1.neon.internal.tech"), [&"neon"]);
+        assert_eq!(map.query("ep-1.foo.localhost"), [&"mock"]);
+
+        Ok(())
+    }
+}
--- a/proxy/src/config.rs
+++ b/proxy/src/config.rs
@@ -1,12 +1,6 @@
-use crate::auth;
-use anyhow::{bail, ensure, Context, Ok};
-use rustls::sign;
-use std::{
-    collections::{HashMap, HashSet},
-    str::FromStr,
-    sync::Arc,
-    time::Duration,
-};
+use crate::{auth, certs};
+use anyhow::{bail, Context};
+use std::{str::FromStr, sync::Arc, time::Duration};

 pub struct ProxyConfig {
    pub tls_config: Option<TlsConfig>,
@@ -22,7 +16,6 @@ pub struct MetricCollectionConfig {

 pub struct TlsConfig {
    pub config: Arc<rustls::ServerConfig>,
-    pub common_names: Option<HashSet<String>>,
 }

 impl TlsConfig {
@@ -31,149 +24,21 @@ impl TlsConfig {
    }
 }

-/// Configure TLS for the main endpoint.
-pub fn configure_tls(
-    key_path: &str,
-    cert_path: &str,
-    certs_dir: Option<&String>,
-) -> anyhow::Result<TlsConfig> {
-    let mut cert_resolver = CertResolver::new();
+impl TlsConfig {
+    pub fn new(resolver: certs::CertResolver) -> anyhow::Result<Self> {
+        let rustls_config = rustls::ServerConfig::builder()
+            .with_safe_default_cipher_suites()
+            .with_safe_default_kx_groups()
+            // allow TLS 1.2 to be compatible with older client libraries
+            .with_protocol_versions(&[&rustls::version::TLS13, &rustls::version::TLS12])?
+            .with_no_client_auth()
+            .with_cert_resolver(Arc::new(resolver));

-    // add default certificate
-    cert_resolver.add_cert(key_path, cert_path)?;
-
-    // add extra certificates
-    if let Some(certs_dir) = certs_dir {
-        for entry in std::fs::read_dir(certs_dir)? {
-            let entry = entry?;
-            let path = entry.path();
-            if path.is_dir() {
-                // file names aligned with default cert-manager names
-                let key_path = path.join("tls.key");
-                let cert_path = path.join("tls.crt");
-                if key_path.exists() && cert_path.exists() {
-                    cert_resolver
-                        .add_cert(&key_path.to_string_lossy(), &cert_path.to_string_lossy())?;
-                }
-            }
-        }
-    }
-
-    let common_names = cert_resolver.get_common_names();
-
-    let config = rustls::ServerConfig::builder()
-        .with_safe_default_cipher_suites()
-        .with_safe_default_kx_groups()
-        // allow TLS 1.2 to be compatible with older client libraries
-        .with_protocol_versions(&[&rustls::version::TLS13, &rustls::version::TLS12])?
-        .with_no_client_auth()
-        .with_cert_resolver(Arc::new(cert_resolver))
-        .into();
-
-    Ok(TlsConfig {
-        config,
-        common_names: Some(common_names),
-    })
-}
-
-struct CertResolver {
-    certs: HashMap<String, Arc<rustls::sign::CertifiedKey>>,
-}
-
-impl CertResolver {
-    fn new() -> Self {
-        Self {
-            certs: HashMap::new(),
-        }
-    }
-
-    fn add_cert(&mut self, key_path: &str, cert_path: &str) -> anyhow::Result<()> {
-        let priv_key = {
-            let key_bytes = std::fs::read(key_path).context("TLS key file")?;
-            let mut keys = rustls_pemfile::pkcs8_private_keys(&mut &key_bytes[..])
-                .context(format!("Failed to read TLS keys at '{key_path}'"))?;
-
-            ensure!(keys.len() == 1, "keys.len() = {} (should be 1)", keys.len());
-            keys.pop().map(rustls::PrivateKey).unwrap()
+        let config = TlsConfig {
+            config: Arc::new(rustls_config),
        };

-        let key = sign::any_supported_type(&priv_key).context("invalid private key")?;
-
-        let cert_chain_bytes = std::fs::read(cert_path)
-            .context(format!("Failed to read TLS cert file at '{cert_path}.'"))?;
-
-        let cert_chain = {
-            rustls_pemfile::certs(&mut &cert_chain_bytes[..])
-                .context(format!(
-                    "Failed to read TLS certificate chain from bytes from file at '{cert_path}'."
-                ))?
-                .into_iter()
-                .map(rustls::Certificate)
-                .collect()
-        };
-
-        let common_name = {
-            let pem = x509_parser::pem::parse_x509_pem(&cert_chain_bytes)
-                .context(format!(
-                    "Failed to parse PEM object from bytes from file at '{cert_path}'."
-                ))?
-                .1;
-            let common_name = pem.parse_x509()?.subject().to_string();
-
-            // We only use non-wildcard certificates in link proxy so it seems okay to treat them the same as
-            // wildcard ones as we don't use SNI there. That treatment only affects certificate selection, so
-            // verify-full will still check wildcard match. Old coding here just ignored non-wildcard common names
-            // and passed None instead, which blows up number of cases downstream code should handle. Proper coding
-            // here should better avoid Option for common_names, and do wildcard-based certificate selection instead
-            // of cutting off '*.' parts.
-            if common_name.starts_with("CN=*.") {
-                common_name.strip_prefix("CN=*.").map(|s| s.to_string())
-            } else {
-                common_name.strip_prefix("CN=").map(|s| s.to_string())
-            }
-        }
-        .context(format!(
-            "Failed to parse common name from certificate at '{cert_path}'."
-        ))?;
-
-        self.certs.insert(
-            common_name,
-            Arc::new(rustls::sign::CertifiedKey::new(cert_chain, key)),
-        );
-
-        Ok(())
-    }
-
-    fn get_common_names(&self) -> HashSet<String> {
-        self.certs.keys().map(|s| s.to_string()).collect()
-    }
-}
-
-impl rustls::server::ResolvesServerCert for CertResolver {
-    fn resolve(
-        &self,
-        _client_hello: rustls::server::ClientHello,
-    ) -> Option<Arc<rustls::sign::CertifiedKey>> {
-        // loop here and cut off more and more subdomains until we find
-        // a match to get a proper wildcard support. OTOH, we now do not
-        // use nested domains, so keep this simple for now.
-        //
-        // With the current coding foo.com will match *.foo.com and that
-        // repeats behavior of the old code.
-        if let Some(mut sni_name) = _client_hello.server_name() {
-            loop {
-                if let Some(cert) = self.certs.get(sni_name) {
-                    return Some(cert.clone());
-                }
-                if let Some((_, rest)) = sni_name.split_once('.') {
-                    sni_name = rest;
-                } else {
-                    return None;
-                }
-            }
-        } else {
-            None
-        }
+        Ok(config)
    }
 }

--- a/proxy/src/main.rs
+++ b/proxy/src/main.rs
@@ -7,6 +7,7 @@
 mod auth;
 mod cache;
 mod cancellation;
+mod certs;
 mod compute;
 mod config;
 mod console;
@@ -23,10 +24,12 @@ mod url;
 mod waiters;

 use anyhow::{bail, Context};
+use auth::BackendType;
+use certs::config::TlsServers;
 use clap::{self, Arg};
-use config::ProxyConfig;
+use config::{MetricCollectionConfig, ProxyConfig, TlsConfig};
 use futures::FutureExt;
-use std::{borrow::Cow, future::Future, net::SocketAddr};
+use std::{borrow::Cow, future::Future, net::SocketAddr, path::PathBuf};
 use tokio::{net::TcpListener, task::JoinError};
 use tracing::{info, warn};
 use utils::{project_git_version, sentry_init::init_sentry};
@@ -126,27 +129,34 @@ async fn handle_signals() -> anyhow::Result<()> {
    }
 }

-/// ProxyConfig is created at proxy startup, and lives forever.
-fn build_config(args: &clap::ArgMatches) -> anyhow::Result<&'static ProxyConfig> {
-    let tls_config = match (
-        args.get_one::<String>("tls-key"),
-        args.get_one::<String>("tls-cert"),
-    ) {
-        (Some(key_path), Some(cert_path)) => Some(config::configure_tls(
-            key_path,
-            cert_path,
-            args.get_one::<String>("certs-dir"),
-        )?),
-        (None, None) => None,
+fn build_tls_config(args: &clap::ArgMatches) -> anyhow::Result<Option<TlsConfig>> {
+    let tls_config = args.get_one::<PathBuf>("tls-config");
+    let main = tls_config.map(TlsServers::from_config_file).transpose()?;
+
+    let tls_cert = args.get_one::<PathBuf>("tls-cert");
+    let tls_key = args.get_one::<PathBuf>("tls-key");
+
+    let _aux = match (tls_cert, tls_key) {
+        (Some(_key), Some(_cert)) => todo!("implement legacy TLS setup"),
+        (None, None) => None::<()>,
        _ => bail!("either both or neither tls-key and tls-cert must be specified"),
    };

-    let metric_collection = match (
-        args.get_one::<String>("metric-collection-endpoint"),
-        args.get_one::<String>("metric-collection-interval"),
-    ) {
+    // TODO: first merge `main` and `_aux` into one.
+    main.map(|servers| {
+        let resolver = certs::CertResolver::new(servers)?;
+        TlsConfig::new(resolver)
+    })
+    .transpose()
+}
+
+fn build_metrics_config(args: &clap::ArgMatches) -> anyhow::Result<Option<MetricCollectionConfig>> {
+    let endpoint = args.get_one::<String>("metric-collection-endpoint");
+    let interval = args.get_one::<String>("metric-collection-interval");
+
+    let config = match (endpoint, interval) {
        (Some(endpoint), Some(interval)) => Some(config::MetricCollectionConfig {
-            endpoint: endpoint.parse()?,
+            endpoint: endpoint.parse().context("bad metrics endpoint")?,
            interval: humantime::parse_duration(interval)?,
        }),
        (None, None) => None,
@@ -156,7 +166,11 @@ fn build_config(args: &clap::ArgMatches) -> anyhow::Result<&'static ProxyConfig>
        ),
    };

-    let auth_backend = match args.get_one::<String>("auth-backend").unwrap().as_str() {
+    Ok(config)
+}
+
+fn build_auth_config(args: &clap::ArgMatches) -> anyhow::Result<BackendType<'static, ()>> {
+    let config = match args.get_one::<String>("auth-backend").unwrap().as_str() {
        "console" => {
            let config::CacheOptions { size, ttl } = args
                .get_one::<String>("wake-compute-cache")
@@ -186,10 +200,15 @@ fn build_config(args: &clap::ArgMatches) -> anyhow::Result<&'static ProxyConfig>
        other => bail!("unsupported auth backend: {other}"),
    };

+    Ok(config)
+}
+
+/// ProxyConfig is created at proxy startup, and lives forever.
+fn build_config(args: &clap::ArgMatches) -> anyhow::Result<&'static ProxyConfig> {
    let config = Box::leak(Box::new(ProxyConfig {
-        tls_config,
-        auth_backend,
-        metric_collection,
+        tls_config: build_tls_config(args)?,
+        auth_backend: build_auth_config(args)?,
+        metric_collection: build_metrics_config(args)?,
    }));

    Ok(config)
@@ -249,20 +268,22 @@ fn cli() -> clap::Command {
                .short('k')
                .long("tls-key")
                .alias("ssl-key") // backwards compatibility
-                .help("path to TLS key for client postgres connections"),
+                .help("path to TLS key for client postgres connections")
+                .value_parser(clap::builder::PathBufValueParser::new()),
        )
        .arg(
            Arg::new("tls-cert")
                .short('c')
                .long("tls-cert")
                .alias("ssl-cert") // backwards compatibility
-                .help("path to TLS cert for client postgres connections"),
+                .help("path to TLS cert for client postgres connections")
+                .value_parser(clap::builder::PathBufValueParser::new()),
        )
-        // tls-key and tls-cert are for backwards compatibility, we can put all certs in one dir
        .arg(
-            Arg::new("certs-dir")
-                .long("certs-dir")
-                .help("path to directory with TLS certificates for client postgres connections"),
+            Arg::new("tls-config")
+                .long("tls-config")
+                .help("path to the TLS config file (example: config/servers.dhall)")
+                .value_parser(clap::builder::PathBufValueParser::new()),
        )
        .arg(
            Arg::new("metric-collection-endpoint")
--- a/proxy/src/metrics.rs
+++ b/proxy/src/metrics.rs
@@ -5,7 +5,7 @@ use chrono::{DateTime, Utc};
 use consumption_metrics::{idempotency_key, Event, EventChunk, EventType, CHUNK_SIZE};
 use serde::Serialize;
 use std::collections::HashMap;
-use tracing::{error, info, instrument, trace, warn};
+use tracing::{debug, error, info, instrument, trace};

 const PROXY_IO_BYTES_PER_CLIENT: &str = "proxy_io_bytes_per_client";

@@ -84,14 +84,10 @@ fn gather_proxy_io_bytes_per_client() -> Vec<(Ids, (u64, DateTime<Utc>))> {

                    let value = ms.get_counter().get_value() as u64;

-                    // Report if the metric value is suspiciously large
-                    if value > (1u64 << 40) {
-                        warn!(
-                            "potentially abnormal counter value: branch_id {} endpoint_id {} val: {}",
-                            branch_id, endpoint_id, value
-                        );
-                    }
-
+                    debug!(
+                        "branch_id {} endpoint_id {} val: {}",
+                        branch_id, endpoint_id, value
+                    );
                    current_metrics.push((
                        Ids {
                            endpoint_id: endpoint_id.to_string(),
@@ -128,15 +124,11 @@ async fn collect_metrics_iteration(
            let mut value = *curr_val;

            if let Some((prev_val, prev_time)) = cached_metrics.get(curr_key) {
-                // Only send metrics updates if the metric has increased
-                if curr_val > prev_val {
+                // Only send metrics updates if the metric has changed
+                if curr_val - prev_val > 0 {
                    value = curr_val - prev_val;
                    start_time = *prev_time;
                } else {
-                    if curr_val < prev_val {
-                        error!("proxy_io_bytes_per_client metric value decreased from {} to {} for key {:?}",
-                        prev_val, curr_val, curr_key);
-                    }
                    return None;
                }
            };
@@ -197,7 +189,7 @@ async fn collect_metrics_iteration(
                    })
                    // update cached value (add delta) and time
                    .and_modify(|e| {
-                        e.0 = e.0.saturating_add(send_metric.value);
+                        e.0 += send_metric.value;
                        e.1 = stop_time
                    })
                    // cache new metric
--- a/proxy/src/proxy.rs
+++ b/proxy/src/proxy.rs
@@ -112,7 +112,6 @@ pub async fn handle_ws_client(
        NUM_CONNECTIONS_CLOSED_COUNTER.inc();
    }

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

    // TLS is None here, because the connection is already encrypted.
@@ -124,11 +123,10 @@ pub async fn handle_ws_client(

    // Extract credentials which we're going to use for auth.
    let creds = {
-        let common_names = tls.and_then(|tls| tls.common_names.clone());
        let result = config
            .auth_backend
            .as_ref()
-            .map(|_| auth::ClientCredentials::parse(&params, hostname, common_names))
+            .map(|_| auth::ClientCredentials::parse(&params, hostname))
            .transpose();

        async { result }.or_else(|e| stream.throw_error(e)).await?
@@ -163,11 +161,10 @@ async fn handle_client(
    // Extract credentials which we're going to use for auth.
    let creds = {
        let sni = stream.get_ref().sni_hostname();
-        let common_names = tls.and_then(|tls| tls.common_names.clone());
        let result = config
            .auth_backend
            .as_ref()
-            .map(|_| auth::ClientCredentials::parse(&params, sni, common_names))
+            .map(|_| auth::ClientCredentials::parse(&params, sni))
            .transpose();

        async { result }.or_else(|e| stream.throw_error(e)).await?
--- a/proxy/src/proxy/tests.rs
+++ b/proxy/src/proxy/tests.rs
@@ -41,10 +41,7 @@ impl ClientConfig<'_> {
 }

 /// Generate TLS certificates and build rustls configs for client and server.
-fn generate_tls_config<'a>(
-    hostname: &'a str,
-    common_name: &'a str,
-) -> anyhow::Result<(ClientConfig<'a>, TlsConfig)> {
+fn generate_tls_config(hostname: &str) -> anyhow::Result<(ClientConfig<'_>, TlsConfig)> {
    let (ca, cert, key) = generate_certs(hostname)?;

    let tls_config = {
@@ -54,12 +51,7 @@ fn generate_tls_config<'a>(
            .with_single_cert(vec![cert], key)?
            .into();

-        let common_names = Some([common_name.to_owned()].iter().cloned().collect());
-
-        TlsConfig {
-            config,
-            common_names,
-        }
+        TlsConfig { config }
    };

    let client_config = {
@@ -152,7 +144,7 @@ async fn dummy_proxy(
 async fn handshake_tls_is_enforced_by_proxy() -> anyhow::Result<()> {
    let (client, server) = tokio::io::duplex(1024);

-    let (_, server_config) = generate_tls_config("generic-project-name.localhost", "localhost")?;
+    let (_, server_config) = generate_tls_config("generic-project-name.localhost")?;
    let proxy = tokio::spawn(dummy_proxy(client, Some(server_config), NoAuth));

    let client_err = tokio_postgres::Config::new()
@@ -180,8 +172,7 @@ async fn handshake_tls_is_enforced_by_proxy() -> anyhow::Result<()> {
 async fn handshake_tls() -> anyhow::Result<()> {
    let (client, server) = tokio::io::duplex(1024);

-    let (client_config, server_config) =
-        generate_tls_config("generic-project-name.localhost", "localhost")?;
+    let (client_config, server_config) = generate_tls_config("generic-project-name.localhost")?;
    let proxy = tokio::spawn(dummy_proxy(client, Some(server_config), NoAuth));

    let (_client, _conn) = tokio_postgres::Config::new()
@@ -239,8 +230,7 @@ async fn keepalive_is_inherited() -> anyhow::Result<()> {
 async fn scram_auth_good(#[case] password: &str) -> anyhow::Result<()> {
    let (client, server) = tokio::io::duplex(1024);

-    let (client_config, server_config) =
-        generate_tls_config("generic-project-name.localhost", "localhost")?;
+    let (client_config, server_config) = generate_tls_config("generic-project-name.localhost")?;
    let proxy = tokio::spawn(dummy_proxy(
        client,
        Some(server_config),
@@ -262,8 +252,7 @@ async fn scram_auth_good(#[case] password: &str) -> anyhow::Result<()> {
 async fn scram_auth_mock() -> anyhow::Result<()> {
    let (client, server) = tokio::io::duplex(1024);

-    let (client_config, server_config) =
-        generate_tls_config("generic-project-name.localhost", "localhost")?;
+    let (client_config, server_config) = generate_tls_config("generic-project-name.localhost")?;
    let proxy = tokio::spawn(dummy_proxy(
        client,
        Some(server_config),
--- a/pyproject.toml
+++ b/pyproject.toml
@@ -26,7 +26,7 @@ prometheus-client = "^0.14.1"
 pytest-timeout = "^2.1.0"
 Werkzeug = "^2.2.3"
 pytest-order = "^1.0.1"
-allure-pytest = "^2.13.1"
+allure-pytest = "^2.10.0"
 pytest-asyncio = "^0.19.0"
 toml = "^0.10.2"
 psutil = "^5.9.4"
@@ -34,7 +34,6 @@ types-psutil = "^5.9.5.4"
 types-toml = "^0.10.8"
 pytest-httpserver = "^1.0.6"
 aiohttp = "3.7.4"
-pytest-rerunfailures = "^11.1.2"

 [tool.poetry.group.dev.dependencies]
 black = "^23.1.0"
@@ -70,9 +69,6 @@ strict = true
 module = [
    "asyncpg.*",
    "pg8000.*",
-    "allure.*",
-    "allure_commons.*",
-    "allure_pytest.*",
 ]
 ignore_missing_imports = true

--- a/run_clippy.sh
+++ b/run_clippy.sh
@@ -8,7 +8,13 @@
 # warnings and errors right in the editor.
 # In vscode, this setting is Rust-analyzer>Check On Save:Command

+# manual-range-contains wants
+#   !(4..=MAX_STARTUP_PACKET_LENGTH).contains(&len)
+# instead of
+#   len < 4 || len > MAX_STARTUP_PACKET_LENGTH
+# , let's disagree.
+
 # * `-A unknown_lints` – do not warn about unknown lint suppressions
 #                        that people with newer toolchains might use
 # * `-D warnings`      - fail on any warnings (`cargo` returns non-zero exit status)
-cargo clippy --locked --all --all-targets --all-features -- -A unknown_lints -D warnings
+cargo clippy --locked --all --all-targets --all-features -- -A unknown_lints -A clippy::manual-range-contains -D warnings
--- a/safekeeper/src/timeline.rs
+++ b/safekeeper/src/timeline.rs
@@ -674,8 +674,7 @@ impl Timeline {
            bail!(TimelineError::Cancelled(self.ttid));
        }

-        let mut state = self.write_shared_state();
-        state.sk.inmem.backup_lsn = max(state.sk.inmem.backup_lsn, backup_lsn);
+        self.write_shared_state().sk.inmem.backup_lsn = backup_lsn;
        // we should check whether to shut down offloader, but this will be done
        // soon by peer communication anyway.
        Ok(())
--- a/safekeeper/src/wal_backup.rs
+++ b/safekeeper/src/wal_backup.rs
@@ -323,8 +323,7 @@ impl WalBackupTask {
            }

            match backup_lsn_range(
-                &self.timeline,
-                &mut backup_lsn,
+                backup_lsn,
                commit_lsn,
                self.wal_seg_size,
                &self.timeline_dir,
@@ -332,7 +331,13 @@ impl WalBackupTask {
            )
            .await
            {
-                Ok(()) => {
+                Ok(backup_lsn_result) => {
+                    backup_lsn = backup_lsn_result;
+                    let res = self.timeline.set_wal_backup_lsn(backup_lsn_result);
+                    if let Err(e) = res {
+                        error!("failed to set wal_backup_lsn: {}", e);
+                        return;
+                    }
                    retry_attempt = 0;
                }
                Err(e) => {
@@ -349,25 +354,20 @@ impl WalBackupTask {
 }

 pub async fn backup_lsn_range(
-    timeline: &Arc<Timeline>,
-    backup_lsn: &mut Lsn,
+    start_lsn: Lsn,
    end_lsn: Lsn,
    wal_seg_size: usize,
    timeline_dir: &Path,
    workspace_dir: &Path,
-) -> Result<()> {
-    let start_lsn = *backup_lsn;
+) -> Result<Lsn> {
+    let mut res = start_lsn;
    let segments = get_segments(start_lsn, end_lsn, wal_seg_size);
    for s in &segments {
        backup_single_segment(s, timeline_dir, workspace_dir)
            .await
            .with_context(|| format!("offloading segno {}", s.seg_no))?;

-        let new_backup_lsn = s.end_lsn;
-        timeline
-            .set_wal_backup_lsn(new_backup_lsn)
-            .context("setting wal_backup_lsn")?;
-        *backup_lsn = new_backup_lsn;
+        res = s.end_lsn;
    }
    info!(
        "offloaded segnos {:?} up to {}, previous backup_lsn {}",
@@ -375,7 +375,7 @@ pub async fn backup_lsn_range(
        end_lsn,
        start_lsn,
    );
-    Ok(())
+    Ok(res)
 }

 async fn backup_single_segment(
--- a/scripts/flaky_tests.py
+++ b/scripts/flaky_tests.py
@@ -1,87 +0,0 @@
-#! /usr/bin/env python3
-
-import argparse
-import json
-import logging
-from collections import defaultdict
-from typing import DefaultDict, Dict
-
-import psycopg2
-import psycopg2.extras
-
-# We call the test "flaky" if it failed at least once on the main branch in the last N=10 days.
-FLAKY_TESTS_QUERY = """
-    SELECT
-        DISTINCT parent_suite, suite, test
-    FROM
-        (
-            SELECT
-                revision,
-                jsonb_array_elements(data -> 'children') -> 'name' as parent_suite,
-                jsonb_array_elements(jsonb_array_elements(data -> 'children') -> 'children') -> 'name' as suite,
-                jsonb_array_elements(jsonb_array_elements(jsonb_array_elements(data -> 'children') -> 'children') -> 'children') -> 'name' as test,
-                jsonb_array_elements(jsonb_array_elements(jsonb_array_elements(data -> 'children') -> 'children') -> 'children') -> 'status' as status,
-                to_timestamp((jsonb_array_elements(jsonb_array_elements(jsonb_array_elements(data -> 'children') -> 'children') -> 'children') -> 'time' -> 'start')::bigint / 1000)::date as timestamp
-            FROM
-                regress_test_results
-            WHERE
-                reference = 'refs/heads/main'
-        ) data
-    WHERE
-        timestamp > CURRENT_DATE - INTERVAL '%s' day
-        AND status::text IN ('"failed"', '"broken"')
-    ;
-"""
-
-
-def main(args: argparse.Namespace):
-    connstr = args.connstr
-    interval_days = args.days
-    output = args.output
-
-    res: DefaultDict[str, DefaultDict[str, Dict[str, bool]]]
-    res = defaultdict(lambda: defaultdict(dict))
-
-    logging.info("connecting to the database...")
-    with psycopg2.connect(connstr, connect_timeout=10) as conn:
-        with conn.cursor(cursor_factory=psycopg2.extras.DictCursor) as cur:
-            logging.info("fetching flaky tests...")
-            cur.execute(FLAKY_TESTS_QUERY, (interval_days,))
-            rows = cur.fetchall()
-
-    for row in rows:
-        logging.info(f"\t{row['parent_suite'].replace('.', '/')}/{row['suite']}.py::{row['test']}")
-        res[row["parent_suite"]][row["suite"]][row["test"]] = True
-
-    logging.info(f"saving results to {output.name}")
-    json.dump(res, output, indent=2)
-
-
-if __name__ == "__main__":
-    parser = argparse.ArgumentParser(description="Detect flaky tests in the last N days")
-    parser.add_argument(
-        "--output",
-        type=argparse.FileType("w"),
-        default="flaky.json",
-        help="path to output json file (default: flaky.json)",
-    )
-    parser.add_argument(
-        "--days",
-        required=False,
-        default=10,
-        type=int,
-        help="how many days to look back for flaky tests (default: 10)",
-    )
-    parser.add_argument(
-        "connstr",
-        help="connection string to the test results database",
-    )
-    args = parser.parse_args()
-
-    level = logging.INFO
-    logging.basicConfig(
-        format="%(message)s",
-        level=level,
-    )
-
-    main(args)
--- a/scripts/pr-comment-test-report.js
+++ b/scripts/pr-comment-test-report.js
@@ -1,125 +0,0 @@
-//
-// The script parses Allure reports and posts a comment with a summary of the test results to the PR.
-// It accepts an array of items and creates a comment with a summary for each one (for "release" and "debug", together or separately if any of them failed to be generated).
-//
-// The comment is updated on each run with the latest results.
-//
-// It is designed to be used with actions/github-script from GitHub Workflows:
-// - uses: actions/github-script@v6
-//   with:
-//     script: |
-//       const script = require("./scripts/pr-comment-test-report.js")
-//       await script({
-//         github,
-//         context,
-//         fetch,
-//         reports: [{...}, ...], // each report is expected to have "buildType", "reportUrl", and "jsonUrl" properties
-//       })
-//
-
-module.exports = async ({ github, context, fetch, reports }) => {
-    // Marker to find the comment in the subsequent runs
-    const startMarker = `<!--AUTOMATIC COMMENT START #${context.payload.number}-->`
-    // GitHub bot id taken from (https://api.github.com/users/github-actions[bot])
-    const githubActionsBotId = 41898282
-    // The latest commit in the PR URL
-    const commitUrl = `${context.serverUrl}/${context.repo.owner}/${context.repo.repo}/pull/${context.payload.number}/commits/${context.payload.pull_request.head.sha}`
-    // Commend body itself
-    let commentBody = `${startMarker}\n### Test results for ${commitUrl}:\n___\n`
-
-    // Common parameters for GitHub API requests
-    const ownerRepoParams = {
-        owner: context.repo.owner,
-        repo: context.repo.repo,
-    }
-
-    for (const report of reports) {
-        const {buildType, reportUrl, jsonUrl} = report
-
-        if (!reportUrl || !jsonUrl) {
-            console.warn(`"reportUrl" or "jsonUrl" aren't set for ${buildType} build`)
-            continue
-        }
-
-        const suites = await (await fetch(jsonUrl)).json()
-
-        // Allure distinguishes "failed" (with an assertion error) and "broken" (with any other error) tests.
-        // For this report it's ok to treat them in the same way (as failed).
-        failedTests = []
-        passedTests = []
-        skippedTests = []
-
-        retriedTests = []
-        retriedStatusChangedTests = []
-
-        for (const parentSuite of suites.children) {
-            for (const suite of parentSuite.children) {
-                for (const test of suite.children) {
-                    pytestName = `${parentSuite.name.replace(".", "/")}/${suite.name}.py::${test.name}`
-                    test.pytestName = pytestName
-
-                    if (test.status === "passed") {
-                        passedTests.push(test);
-                    } else if (test.status === "failed" || test.status === "broken") {
-                        failedTests.push(test);
-                    } else if (test.status === "skipped") {
-                        skippedTests.push(test);
-                    }
-
-                    if (test.retriesCount > 0) {
-                        retriedTests.push(test);
-
-                        if (test.retriedStatusChangedTests) {
-                            retriedStatusChangedTests.push(test);
-                        }
-                    }
-                }
-            }
-        }
-
-        const totalTestsCount = failedTests.length + passedTests.length + skippedTests.length
-        commentBody += `#### ${buildType} build: ${totalTestsCount} tests run: ${passedTests.length} passed, ${failedTests.length} failed, ${skippedTests.length} ([full report](${reportUrl}))\n`
-        if (failedTests.length > 0) {
-            commentBody += `Failed tests:\n`
-            for (const test of failedTests) {
-                const allureLink = `${reportUrl}#suites/${test.parentUid}/${test.uid}`
-
-                commentBody += `- [\`${test.pytestName}\`](${allureLink})`
-                if (test.retriesCount > 0) {
-                    commentBody += ` (ran [${test.retriesCount + 1} times](${allureLink}/retries))`
-                }
-                commentBody += "\n"
-            }
-            commentBody += "\n"
-        }
-        if (retriedStatusChangedTests > 0) {
-            commentBody += `Flaky tests:\n`
-            for (const test of retriedStatusChangedTests) {
-                const status = test.status === "passed" ? ":white_check_mark:" : ":x:"
-                commentBody += `- ${status} [\`${test.pytestName}\`](${reportUrl}#suites/${test.parentUid}/${test.uid}/retries)\n`
-            }
-            commentBody += "\n"
-        }
-        commentBody += "___\n"
-    }
-
-    const { data: comments } = await github.rest.issues.listComments({
-        issue_number: context.payload.number,
-        ...ownerRepoParams,
-    })
-
-    const comment = comments.find(comment => comment.user.id === githubActionsBotId && comment.body.startsWith(startMarker))
-    if (comment) {
-        await github.rest.issues.updateComment({
-            comment_id: comment.id,
-            body: commentBody,
-            ...ownerRepoParams,
-        })
-    } else {
-        await github.rest.issues.createComment({
-            issue_number: context.payload.number,
-            body: commentBody,
-            ...ownerRepoParams,
-        })
-    }
-}
--- a/test_runner/conftest.py
+++ b/test_runner/conftest.py
@@ -4,5 +4,4 @@ pytest_plugins = (
    "fixtures.pg_stats",
    "fixtures.compare_fixtures",
    "fixtures.slow",
-    "fixtures.flaky",
 )
--- a/test_runner/fixtures/flaky.py
+++ b/test_runner/fixtures/flaky.py
@@ -1,58 +0,0 @@
-import json
-from pathlib import Path
-from typing import List
-
-import pytest
-from _pytest.config import Config
-from _pytest.config.argparsing import Parser
-from allure_commons.types import LabelType
-from allure_pytest.utils import allure_name, allure_suite_labels
-
-from fixtures.log_helper import log
-
-"""
-The plugin reruns flaky tests.
-It uses `pytest.mark.flaky` provided by `pytest-rerunfailures` plugin and flaky tests detected by `scripts/flaky_tests.py`
-
-Note: the logic of getting flaky tests is extracted to a separate script to avoid running it for each of N xdist workers
-"""
-
-
-def pytest_addoption(parser: Parser):
-    parser.addoption(
-        "--flaky-tests-json",
-        action="store",
-        type=Path,
-        help="Path to json file with flaky tests generated by scripts/flaky_tests.py",
-    )
-
-
-def pytest_collection_modifyitems(config: Config, items: List[pytest.Item]):
-    if not config.getoption("--flaky-tests-json"):
-        return
-
-    # Any error with getting flaky tests aren't critical, so just do not rerun any tests
-    flaky_json = config.getoption("--flaky-tests-json")
-    if not flaky_json.exists():
-        return
-
-    content = flaky_json.read_text()
-    try:
-        flaky_tests = json.loads(content)
-    except ValueError:
-        log.error(f"Can't parse {content} as json")
-        return
-
-    for item in items:
-        # Use the same logic for constructing test name as Allure does (we store allure-provided data in DB)
-        # Ref https://github.com/allure-framework/allure-python/blob/2.13.1/allure-pytest/src/listener.py#L98-L100
-        allure_labels = dict(allure_suite_labels(item))
-        parent_suite = str(allure_labels.get(LabelType.PARENT_SUITE))
-        suite = str(allure_labels.get(LabelType.SUITE))
-        params = item.callspec.params if hasattr(item, "callspec") else {}
-        name = allure_name(item, params)
-
-        if flaky_tests.get(parent_suite, {}).get(suite, {}).get(name, False):
-            # Rerun 3 times = 1 original run + 2 reruns
-            log.info(f"Marking {item.nodeid} as flaky. It will be rerun up to 3 times")
-            item.add_marker(pytest.mark.flaky(reruns=2))
--- a/test_runner/fixtures/neon_fixtures.py
+++ b/test_runner/fixtures/neon_fixtures.py
@@ -14,9 +14,9 @@ import tempfile
 import textwrap
 import time
 import uuid
+from collections import defaultdict
 from contextlib import closing, contextmanager
 from dataclasses import dataclass, field
-from datetime import datetime
 from enum import Flag, auto
 from functools import cached_property
 from itertools import chain, product
@@ -43,12 +43,11 @@ from psycopg2.extensions import make_dsn, parse_dsn
 from typing_extensions import Literal

 from fixtures.log_helper import log
-from fixtures.pageserver.http import PageserverHttpClient
-from fixtures.pageserver.utils import wait_for_last_record_lsn, wait_for_upload
+from fixtures.metrics import Metrics, parse_metrics
 from fixtures.types import Lsn, TenantId, TimelineId
 from fixtures.utils import (
    ATTACHMENT_NAME_REGEX,
-    allure_add_grafana_links,
+    Fn,
    allure_attach_from_dir,
    get_self_dir,
    subprocess_capture,
@@ -1119,6 +1118,538 @@ def neon_env_builder(
        yield builder


+class PageserverApiException(Exception):
+    def __init__(self, message, status_code: int):
+        super().__init__(message)
+        self.status_code = status_code
+
+
+class PageserverHttpClient(requests.Session):
+    def __init__(self, port: int, is_testing_enabled_or_skip: Fn, auth_token: Optional[str] = None):
+        super().__init__()
+        self.port = port
+        self.auth_token = auth_token
+        self.is_testing_enabled_or_skip = is_testing_enabled_or_skip
+
+        if auth_token is not None:
+            self.headers["Authorization"] = f"Bearer {auth_token}"
+
+    def verbose_error(self, res: requests.Response):
+        try:
+            res.raise_for_status()
+        except requests.RequestException as e:
+            try:
+                msg = res.json()["msg"]
+            except:  # noqa: E722
+                msg = ""
+            raise PageserverApiException(msg, res.status_code) from e
+
+    def check_status(self):
+        self.get(f"http://localhost:{self.port}/v1/status").raise_for_status()
+
+    def configure_failpoints(self, config_strings: Tuple[str, str] | List[Tuple[str, str]]):
+        self.is_testing_enabled_or_skip()
+
+        if isinstance(config_strings, tuple):
+            pairs = [config_strings]
+        else:
+            pairs = config_strings
+
+        log.info(f"Requesting config failpoints: {repr(pairs)}")
+
+        res = self.put(
+            f"http://localhost:{self.port}/v1/failpoints",
+            json=[{"name": name, "actions": actions} for name, actions in pairs],
+        )
+        log.info(f"Got failpoints request response code {res.status_code}")
+        self.verbose_error(res)
+        res_json = res.json()
+        assert res_json is None
+        return res_json
+
+    def tenant_list(self) -> List[Dict[Any, Any]]:
+        res = self.get(f"http://localhost:{self.port}/v1/tenant")
+        self.verbose_error(res)
+        res_json = res.json()
+        assert isinstance(res_json, list)
+        return res_json
+
+    def tenant_create(self, new_tenant_id: Optional[TenantId] = None) -> TenantId:
+        res = self.post(
+            f"http://localhost:{self.port}/v1/tenant",
+            json={
+                "new_tenant_id": str(new_tenant_id) if new_tenant_id else None,
+            },
+        )
+        self.verbose_error(res)
+        if res.status_code == 409:
+            raise Exception(f"could not create tenant: already exists for id {new_tenant_id}")
+        new_tenant_id = res.json()
+        assert isinstance(new_tenant_id, str)
+        return TenantId(new_tenant_id)
+
+    def tenant_attach(self, tenant_id: TenantId):
+        res = self.post(f"http://localhost:{self.port}/v1/tenant/{tenant_id}/attach")
+        self.verbose_error(res)
+
+    def tenant_detach(self, tenant_id: TenantId, detach_ignored=False):
+        params = {}
+        if detach_ignored:
+            params["detach_ignored"] = "true"
+
+        res = self.post(f"http://localhost:{self.port}/v1/tenant/{tenant_id}/detach", params=params)
+        self.verbose_error(res)
+
+    def tenant_load(self, tenant_id: TenantId):
+        res = self.post(f"http://localhost:{self.port}/v1/tenant/{tenant_id}/load")
+        self.verbose_error(res)
+
+    def tenant_ignore(self, tenant_id: TenantId):
+        res = self.post(f"http://localhost:{self.port}/v1/tenant/{tenant_id}/ignore")
+        self.verbose_error(res)
+
+    def tenant_status(self, tenant_id: TenantId) -> Dict[Any, Any]:
+        res = self.get(f"http://localhost:{self.port}/v1/tenant/{tenant_id}")
+        self.verbose_error(res)
+        res_json = res.json()
+        assert isinstance(res_json, dict)
+        return res_json
+
+    def tenant_config(self, tenant_id: TenantId) -> TenantConfig:
+        res = self.get(f"http://localhost:{self.port}/v1/tenant/{tenant_id}/config")
+        self.verbose_error(res)
+        return TenantConfig.from_json(res.json())
+
+    def set_tenant_config(self, tenant_id: TenantId, config: dict[str, Any]):
+        assert "tenant_id" not in config.keys()
+        res = self.put(
+            f"http://localhost:{self.port}/v1/tenant/config",
+            json={**config, "tenant_id": str(tenant_id)},
+        )
+        self.verbose_error(res)
+
+    def patch_tenant_config_client_side(
+        self,
+        tenant_id: TenantId,
+        inserts: Optional[Dict[str, Any]] = None,
+        removes: Optional[List[str]] = None,
+    ):
+        current = self.tenant_config(tenant_id).tenant_specific_overrides
+        if inserts is not None:
+            current.update(inserts)
+        if removes is not None:
+            for key in removes:
+                del current[key]
+        self.set_tenant_config(tenant_id, current)
+
+    def tenant_size(self, tenant_id: TenantId) -> int:
+        return self.tenant_size_and_modelinputs(tenant_id)[0]
+
+    def tenant_size_and_modelinputs(self, tenant_id: TenantId) -> Tuple[int, Dict[str, Any]]:
+        """
+        Returns the tenant size, together with the model inputs as the second tuple item.
+        """
+        res = self.get(f"http://localhost:{self.port}/v1/tenant/{tenant_id}/synthetic_size")
+        self.verbose_error(res)
+        res = res.json()
+        assert isinstance(res, dict)
+        assert TenantId(res["id"]) == tenant_id
+        size = res["size"]
+        assert type(size) == int
+        inputs = res["inputs"]
+        assert type(inputs) is dict
+        return (size, inputs)
+
+    def tenant_size_debug(self, tenant_id: TenantId) -> str:
+        """
+        Returns the tenant size debug info, as an HTML string
+        """
+        res = self.get(
+            f"http://localhost:{self.port}/v1/tenant/{tenant_id}/synthetic_size",
+            headers={"Accept": "text/html"},
+        )
+        return res.text
+
+    def timeline_list(
+        self,
+        tenant_id: TenantId,
+        include_non_incremental_logical_size: bool = False,
+        include_timeline_dir_layer_file_size_sum: bool = False,
+    ) -> List[Dict[str, Any]]:
+        params = {}
+        if include_non_incremental_logical_size:
+            params["include-non-incremental-logical-size"] = "true"
+        if include_timeline_dir_layer_file_size_sum:
+            params["include-timeline-dir-layer-file-size-sum"] = "true"
+
+        res = self.get(
+            f"http://localhost:{self.port}/v1/tenant/{tenant_id}/timeline", params=params
+        )
+        self.verbose_error(res)
+        res_json = res.json()
+        assert isinstance(res_json, list)
+        return res_json
+
+    def timeline_create(
+        self,
+        tenant_id: TenantId,
+        new_timeline_id: Optional[TimelineId] = None,
+        ancestor_timeline_id: Optional[TimelineId] = None,
+        ancestor_start_lsn: Optional[Lsn] = None,
+    ) -> Dict[Any, Any]:
+        res = self.post(
+            f"http://localhost:{self.port}/v1/tenant/{tenant_id}/timeline",
+            json={
+                "new_timeline_id": str(new_timeline_id) if new_timeline_id else None,
+                "ancestor_start_lsn": str(ancestor_start_lsn) if ancestor_start_lsn else None,
+                "ancestor_timeline_id": str(ancestor_timeline_id) if ancestor_timeline_id else None,
+            },
+        )
+        self.verbose_error(res)
+        if res.status_code == 409:
+            raise Exception(f"could not create timeline: already exists for id {new_timeline_id}")
+
+        res_json = res.json()
+        assert isinstance(res_json, dict)
+        return res_json
+
+    def timeline_detail(
+        self,
+        tenant_id: TenantId,
+        timeline_id: TimelineId,
+        include_non_incremental_logical_size: bool = False,
+        include_timeline_dir_layer_file_size_sum: bool = False,
+        **kwargs,
+    ) -> Dict[Any, Any]:
+        params = {}
+        if include_non_incremental_logical_size:
+            params["include-non-incremental-logical-size"] = "true"
+        if include_timeline_dir_layer_file_size_sum:
+            params["include-timeline-dir-layer-file-size-sum"] = "true"
+
+        res = self.get(
+            f"http://localhost:{self.port}/v1/tenant/{tenant_id}/timeline/{timeline_id}",
+            params=params,
+            **kwargs,
+        )
+        self.verbose_error(res)
+        res_json = res.json()
+        assert isinstance(res_json, dict)
+        return res_json
+
+    def timeline_delete(self, tenant_id: TenantId, timeline_id: TimelineId):
+        res = self.delete(
+            f"http://localhost:{self.port}/v1/tenant/{tenant_id}/timeline/{timeline_id}"
+        )
+        self.verbose_error(res)
+        res_json = res.json()
+        assert res_json is None
+
+    def timeline_gc(
+        self, tenant_id: TenantId, timeline_id: TimelineId, gc_horizon: Optional[int]
+    ) -> dict[str, Any]:
+        self.is_testing_enabled_or_skip()
+
+        log.info(
+            f"Requesting GC: tenant {tenant_id}, timeline {timeline_id}, gc_horizon {repr(gc_horizon)}"
+        )
+        res = self.put(
+            f"http://localhost:{self.port}/v1/tenant/{tenant_id}/timeline/{timeline_id}/do_gc",
+            json={"gc_horizon": gc_horizon},
+        )
+        log.info(f"Got GC request response code: {res.status_code}")
+        self.verbose_error(res)
+        res_json = res.json()
+        assert res_json is not None
+        assert isinstance(res_json, dict)
+        return res_json
+
+    def timeline_compact(self, tenant_id: TenantId, timeline_id: TimelineId):
+        self.is_testing_enabled_or_skip()
+
+        log.info(f"Requesting compact: tenant {tenant_id}, timeline {timeline_id}")
+        res = self.put(
+            f"http://localhost:{self.port}/v1/tenant/{tenant_id}/timeline/{timeline_id}/compact"
+        )
+        log.info(f"Got compact request response code: {res.status_code}")
+        self.verbose_error(res)
+        res_json = res.json()
+        assert res_json is None
+
+    def timeline_get_lsn_by_timestamp(
+        self, tenant_id: TenantId, timeline_id: TimelineId, timestamp
+    ):
+        log.info(
+            f"Requesting lsn by timestamp {timestamp}, tenant {tenant_id}, timeline {timeline_id}"
+        )
+        res = self.get(
+            f"http://localhost:{self.port}/v1/tenant/{tenant_id}/timeline/{timeline_id}/get_lsn_by_timestamp?timestamp={timestamp}",
+        )
+        self.verbose_error(res)
+        res_json = res.json()
+        return res_json
+
+    def timeline_checkpoint(self, tenant_id: TenantId, timeline_id: TimelineId):
+        self.is_testing_enabled_or_skip()
+
+        log.info(f"Requesting checkpoint: tenant {tenant_id}, timeline {timeline_id}")
+        res = self.put(
+            f"http://localhost:{self.port}/v1/tenant/{tenant_id}/timeline/{timeline_id}/checkpoint"
+        )
+        log.info(f"Got checkpoint request response code: {res.status_code}")
+        self.verbose_error(res)
+        res_json = res.json()
+        assert res_json is None
+
+    def timeline_spawn_download_remote_layers(
+        self,
+        tenant_id: TenantId,
+        timeline_id: TimelineId,
+        max_concurrent_downloads: int,
+    ) -> dict[str, Any]:
+        body = {
+            "max_concurrent_downloads": max_concurrent_downloads,
+        }
+        res = self.post(
+            f"http://localhost:{self.port}/v1/tenant/{tenant_id}/timeline/{timeline_id}/download_remote_layers",
+            json=body,
+        )
+        self.verbose_error(res)
+        res_json = res.json()
+        assert res_json is not None
+        assert isinstance(res_json, dict)
+        return res_json
+
+    def timeline_poll_download_remote_layers_status(
+        self,
+        tenant_id: TenantId,
+        timeline_id: TimelineId,
+        spawn_response: dict[str, Any],
+        poll_state=None,
+    ) -> None | dict[str, Any]:
+        res = self.get(
+            f"http://localhost:{self.port}/v1/tenant/{tenant_id}/timeline/{timeline_id}/download_remote_layers",
+        )
+        self.verbose_error(res)
+        res_json = res.json()
+        assert res_json is not None
+        assert isinstance(res_json, dict)
+
+        # assumption in this API client here is that nobody else spawns the task
+        assert res_json["task_id"] == spawn_response["task_id"]
+
+        if poll_state is None or res_json["state"] == poll_state:
+            return res_json
+        return None
+
+    def timeline_download_remote_layers(
+        self,
+        tenant_id: TenantId,
+        timeline_id: TimelineId,
+        max_concurrent_downloads: int,
+        errors_ok=False,
+        at_least_one_download=True,
+    ):
+        res = self.timeline_spawn_download_remote_layers(
+            tenant_id, timeline_id, max_concurrent_downloads
+        )
+        while True:
+            completed = self.timeline_poll_download_remote_layers_status(
+                tenant_id, timeline_id, res, poll_state="Completed"
+            )
+            if not completed:
+                time.sleep(0.1)
+                continue
+            if not errors_ok:
+                assert completed["failed_download_count"] == 0
+            if at_least_one_download:
+                assert completed["successful_download_count"] > 0
+            return completed
+
+    def get_metrics_str(self) -> str:
+        """You probably want to use get_metrics() instead."""
+        res = self.get(f"http://localhost:{self.port}/metrics")
+        self.verbose_error(res)
+        return res.text
+
+    def get_metrics(self) -> Metrics:
+        res = self.get_metrics_str()
+        return parse_metrics(res)
+
+    def get_timeline_metric(
+        self, tenant_id: TenantId, timeline_id: TimelineId, metric_name: str
+    ) -> float:
+        metrics = self.get_metrics()
+        return metrics.query_one(
+            metric_name,
+            filter={
+                "tenant_id": str(tenant_id),
+                "timeline_id": str(timeline_id),
+            },
+        ).value
+
+    def get_remote_timeline_client_metric(
+        self,
+        metric_name: str,
+        tenant_id: TenantId,
+        timeline_id: TimelineId,
+        file_kind: str,
+        op_kind: str,
+    ) -> Optional[float]:
+        metrics = self.get_metrics()
+        matches = metrics.query_all(
+            name=metric_name,
+            filter={
+                "tenant_id": str(tenant_id),
+                "timeline_id": str(timeline_id),
+                "file_kind": str(file_kind),
+                "op_kind": str(op_kind),
+            },
+        )
+        if len(matches) == 0:
+            value = None
+        elif len(matches) == 1:
+            value = matches[0].value
+            assert value is not None
+        else:
+            assert len(matches) < 2, "above filter should uniquely identify metric"
+        return value
+
+    def get_metric_value(
+        self, name: str, filter: Optional[Dict[str, str]] = None
+    ) -> Optional[float]:
+        metrics = self.get_metrics()
+        results = metrics.query_all(name, filter=filter)
+        if not results:
+            log.info(f'could not find metric "{name}"')
+            return None
+        assert len(results) == 1, f"metric {name} with given filters is not unique, got: {results}"
+        return results[0].value
+
+    def layer_map_info(
+        self,
+        tenant_id: TenantId,
+        timeline_id: TimelineId,
+    ) -> LayerMapInfo:
+        res = self.get(
+            f"http://localhost:{self.port}/v1/tenant/{tenant_id}/timeline/{timeline_id}/layer/",
+        )
+        self.verbose_error(res)
+        return LayerMapInfo.from_json(res.json())
+
+    def download_layer(self, tenant_id: TenantId, timeline_id: TimelineId, layer_name: str):
+        res = self.get(
+            f"http://localhost:{self.port}/v1/tenant/{tenant_id}/timeline/{timeline_id}/layer/{layer_name}",
+        )
+        self.verbose_error(res)
+
+        assert res.status_code == 200
+
+    def evict_layer(self, tenant_id: TenantId, timeline_id: TimelineId, layer_name: str):
+        res = self.delete(
+            f"http://localhost:{self.port}/v1/tenant/{tenant_id}/timeline/{timeline_id}/layer/{layer_name}",
+        )
+        self.verbose_error(res)
+
+        assert res.status_code == 200
+
+    def evict_all_layers(self, tenant_id: TenantId, timeline_id: TimelineId):
+        info = self.layer_map_info(tenant_id, timeline_id)
+        for layer in info.historic_layers:
+            self.evict_layer(tenant_id, timeline_id, layer.layer_file_name)
+
+    def disk_usage_eviction_run(self, request: dict[str, Any]):
+        res = self.put(
+            f"http://localhost:{self.port}/v1/disk_usage_eviction/run",
+            json=request,
+        )
+        self.verbose_error(res)
+        return res.json()
+
+    def tenant_break(self, tenant_id: TenantId):
+        res = self.put(f"http://localhost:{self.port}/v1/tenant/{tenant_id}/break")
+        self.verbose_error(res)
+
+
+@dataclass
+class TenantConfig:
+    tenant_specific_overrides: Dict[str, Any]
+    effective_config: Dict[str, Any]
+
+    @classmethod
+    def from_json(cls, d: Dict[str, Any]) -> TenantConfig:
+        return TenantConfig(
+            tenant_specific_overrides=d["tenant_specific_overrides"],
+            effective_config=d["effective_config"],
+        )
+
+
+@dataclass
+class LayerMapInfo:
+    in_memory_layers: List[InMemoryLayerInfo]
+    historic_layers: List[HistoricLayerInfo]
+
+    @classmethod
+    def from_json(cls, d: Dict[str, Any]) -> LayerMapInfo:
+        info = LayerMapInfo(in_memory_layers=[], historic_layers=[])
+
+        json_in_memory_layers = d["in_memory_layers"]
+        assert isinstance(json_in_memory_layers, List)
+        for json_in_memory_layer in json_in_memory_layers:
+            info.in_memory_layers.append(InMemoryLayerInfo.from_json(json_in_memory_layer))
+
+        json_historic_layers = d["historic_layers"]
+        assert isinstance(json_historic_layers, List)
+        for json_historic_layer in json_historic_layers:
+            info.historic_layers.append(HistoricLayerInfo.from_json(json_historic_layer))
+
+        return info
+
+    def kind_count(self) -> Dict[str, int]:
+        counts: Dict[str, int] = defaultdict(int)
+        for inmem_layer in self.in_memory_layers:
+            counts[inmem_layer.kind] += 1
+        for hist_layer in self.historic_layers:
+            counts[hist_layer.kind] += 1
+        return counts
+
+
+@dataclass
+class InMemoryLayerInfo:
+    kind: str
+    lsn_start: str
+    lsn_end: Optional[str]
+
+    @classmethod
+    def from_json(cls, d: Dict[str, Any]) -> InMemoryLayerInfo:
+        return InMemoryLayerInfo(
+            kind=d["kind"],
+            lsn_start=d["lsn_start"],
+            lsn_end=d.get("lsn_end"),
+        )
+
+
+@dataclass(frozen=True)
+class HistoricLayerInfo:
+    kind: str
+    layer_file_name: str
+    layer_file_size: Optional[int]
+    lsn_start: str
+    lsn_end: Optional[str]
+    remote: bool
+
+    @classmethod
+    def from_json(cls, d: Dict[str, Any]) -> HistoricLayerInfo:
+        return HistoricLayerInfo(
+            kind=d["kind"],
+            layer_file_name=d["layer_file_name"],
+            layer_file_size=d.get("layer_file_size"),
+            lsn_start=d["lsn_start"],
+            lsn_end=d.get("lsn_end"),
+            remote=d["remote"],
+        )
+
+
@dataclass
 class PageserverPort:
    pg: int
@@ -1905,16 +2436,10 @@ def remote_pg(
    connstr = os.getenv("BENCHMARK_CONNSTR")
    if connstr is None:
        raise ValueError("no connstr provided, use BENCHMARK_CONNSTR environment variable")
-    start_ms = int(datetime.utcnow().timestamp() * 1000)
+
    with RemotePostgres(pg_bin, connstr) as remote_pg:
        yield remote_pg

-    end_ms = int(datetime.utcnow().timestamp() * 1000)
-    host = parse_dsn(connstr).get("host", "")
-    if host.endswith(".neon.build"):
-        # Add 10s margin to the start and end times
-        allure_add_grafana_links(host, start_ms - 10_000, end_ms + 10_000)
-

 class PSQL:
    """
@@ -2853,6 +3378,151 @@ def check_restored_datadir_content(
    assert (mismatch, error) == ([], [])


+def wait_until(number_of_iterations: int, interval: float, func):
+    """
+    Wait until 'func' returns successfully, without exception. Returns the
+    last return value from the function.
+    """
+    last_exception = None
+    for i in range(number_of_iterations):
+        try:
+            res = func()
+        except Exception as e:
+            log.info("waiting for %s iteration %s failed", func, i + 1)
+            last_exception = e
+            time.sleep(interval)
+            continue
+        return res
+    raise Exception("timed out while waiting for %s" % func) from last_exception
+
+
+def wait_while(number_of_iterations: int, interval: float, func):
+    """
+    Wait until 'func' returns false, or throws an exception.
+    """
+    for i in range(number_of_iterations):
+        try:
+            if not func():
+                return
+            log.info("waiting for %s iteration %s failed", func, i + 1)
+            time.sleep(interval)
+            continue
+        except Exception:
+            return
+    raise Exception("timed out while waiting for %s" % func)
+
+
+def assert_tenant_status(
+    pageserver_http_client: PageserverHttpClient, tenant: TenantId, expected_status: str
+):
+    tenant_status = pageserver_http_client.tenant_status(tenant)
+    log.info(f"tenant_status: {tenant_status}")
+    assert tenant_status["state"] == expected_status, tenant_status
+
+
+def tenant_exists(ps_http: PageserverHttpClient, tenant_id: TenantId):
+    tenants = ps_http.tenant_list()
+    matching = [t for t in tenants if TenantId(t["id"]) == tenant_id]
+    assert len(matching) < 2
+    if len(matching) == 0:
+        return None
+    return matching[0]
+
+
+def remote_consistent_lsn(
+    pageserver_http_client: PageserverHttpClient, tenant: TenantId, timeline: TimelineId
+) -> Lsn:
+    detail = pageserver_http_client.timeline_detail(tenant, timeline)
+
+    if detail["remote_consistent_lsn"] is None:
+        # No remote information at all. This happens right after creating
+        # a timeline, before any part of it has been uploaded to remote
+        # storage yet.
+        return Lsn(0)
+    else:
+        lsn_str = detail["remote_consistent_lsn"]
+        assert isinstance(lsn_str, str)
+        return Lsn(lsn_str)
+
+
+def wait_for_upload(
+    pageserver_http_client: PageserverHttpClient,
+    tenant: TenantId,
+    timeline: TimelineId,
+    lsn: Lsn,
+):
+    """waits for local timeline upload up to specified lsn"""
+    for i in range(20):
+        current_lsn = remote_consistent_lsn(pageserver_http_client, tenant, timeline)
+        if current_lsn >= lsn:
+            log.info("wait finished")
+            return
+        log.info(
+            "waiting for remote_consistent_lsn to reach {}, now {}, iteration {}".format(
+                lsn, current_lsn, i + 1
+            )
+        )
+        time.sleep(1)
+    raise Exception(
+        "timed out while waiting for remote_consistent_lsn to reach {}, was {}".format(
+            lsn, current_lsn
+        )
+    )
+
+
+# Does not use `wait_until` for debugging purposes
+def wait_until_tenant_state(
+    pageserver_http: PageserverHttpClient,
+    tenant_id: TenantId,
+    expected_state: str,
+    iterations: int,
+) -> bool:
+    for _ in range(iterations):
+        try:
+            tenant = pageserver_http.tenant_status(tenant_id=tenant_id)
+            log.debug(f"Tenant {tenant_id} data: {tenant}")
+            if tenant["state"] == expected_state:
+                return True
+        except Exception as e:
+            log.debug(f"Tenant {tenant_id} state retrieval failure: {e}")
+
+        time.sleep(1)
+
+    raise Exception(f"Tenant {tenant_id} did not become {expected_state} in {iterations} seconds")
+
+
+def last_record_lsn(
+    pageserver_http_client: PageserverHttpClient, tenant: TenantId, timeline: TimelineId
+) -> Lsn:
+    detail = pageserver_http_client.timeline_detail(tenant, timeline)
+
+    lsn_str = detail["last_record_lsn"]
+    assert isinstance(lsn_str, str)
+    return Lsn(lsn_str)
+
+
+def wait_for_last_record_lsn(
+    pageserver_http_client: PageserverHttpClient,
+    tenant: TenantId,
+    timeline: TimelineId,
+    lsn: Lsn,
+) -> Lsn:
+    """waits for pageserver to catch up to a certain lsn, returns the last observed lsn."""
+    for i in range(10):
+        current_lsn = last_record_lsn(pageserver_http_client, tenant, timeline)
+        if current_lsn >= lsn:
+            return current_lsn
+        log.info(
+            "waiting for last_record_lsn to reach {}, now {}, iteration {}".format(
+                lsn, current_lsn, i + 1
+            )
+        )
+        time.sleep(1)
+    raise Exception(
+        "timed out while waiting for last_record_lsn to reach {}, was {}".format(lsn, current_lsn)
+    )
+
+
 def wait_for_last_flush_lsn(
    env: NeonEnv, pg: Postgres, tenant: TenantId, timeline: TimelineId
 ) -> Lsn:
@@ -2914,3 +3584,23 @@ def wait_for_sk_commit_lsn_to_reach_remote_storage(
    ps_http.timeline_checkpoint(tenant_id, timeline_id)
    wait_for_upload(ps_http, tenant_id, timeline_id, lsn)
    return lsn
+
+
+def wait_for_upload_queue_empty(
+    pageserver: NeonPageserver, tenant_id: TenantId, timeline_id: TimelineId
+):
+    ps_http = pageserver.http_client()
+    while True:
+        all_metrics = ps_http.get_metrics()
+        tl = all_metrics.query_all(
+            "pageserver_remote_timeline_client_calls_unfinished",
+            {
+                "tenant_id": str(tenant_id),
+                "timeline_id": str(timeline_id),
+            },
+        )
+        assert len(tl) > 0
+        log.info(f"upload queue for {tenant_id}/{timeline_id}: {tl}")
+        if all(m.value == 0 for m in tl):
+            return
+        time.sleep(0.2)
--- a/test_runner/fixtures/pageserver/http.py
+++ b/test_runner/fixtures/pageserver/http.py
@@ -1,545 +0,0 @@
-from __future__ import annotations
-
-import time
-from collections import defaultdict
-from dataclasses import dataclass
-from typing import Any, Dict, List, Optional, Tuple
-
-import requests
-
-from fixtures.log_helper import log
-from fixtures.metrics import Metrics, parse_metrics
-from fixtures.types import Lsn, TenantId, TimelineId
-from fixtures.utils import Fn
-
-
-class PageserverApiException(Exception):
-    def __init__(self, message, status_code: int):
-        super().__init__(message)
-        self.status_code = status_code
-
-
-@dataclass
-class InMemoryLayerInfo:
-    kind: str
-    lsn_start: str
-    lsn_end: Optional[str]
-
-    @classmethod
-    def from_json(cls, d: Dict[str, Any]) -> InMemoryLayerInfo:
-        return InMemoryLayerInfo(
-            kind=d["kind"],
-            lsn_start=d["lsn_start"],
-            lsn_end=d.get("lsn_end"),
-        )
-
-
-@dataclass(frozen=True)
-class HistoricLayerInfo:
-    kind: str
-    layer_file_name: str
-    layer_file_size: Optional[int]
-    lsn_start: str
-    lsn_end: Optional[str]
-    remote: bool
-
-    @classmethod
-    def from_json(cls, d: Dict[str, Any]) -> HistoricLayerInfo:
-        return HistoricLayerInfo(
-            kind=d["kind"],
-            layer_file_name=d["layer_file_name"],
-            layer_file_size=d.get("layer_file_size"),
-            lsn_start=d["lsn_start"],
-            lsn_end=d.get("lsn_end"),
-            remote=d["remote"],
-        )
-
-
-@dataclass
-class LayerMapInfo:
-    in_memory_layers: List[InMemoryLayerInfo]
-    historic_layers: List[HistoricLayerInfo]
-
-    @classmethod
-    def from_json(cls, d: Dict[str, Any]) -> LayerMapInfo:
-        info = LayerMapInfo(in_memory_layers=[], historic_layers=[])
-
-        json_in_memory_layers = d["in_memory_layers"]
-        assert isinstance(json_in_memory_layers, List)
-        for json_in_memory_layer in json_in_memory_layers:
-            info.in_memory_layers.append(InMemoryLayerInfo.from_json(json_in_memory_layer))
-
-        json_historic_layers = d["historic_layers"]
-        assert isinstance(json_historic_layers, List)
-        for json_historic_layer in json_historic_layers:
-            info.historic_layers.append(HistoricLayerInfo.from_json(json_historic_layer))
-
-        return info
-
-    def kind_count(self) -> Dict[str, int]:
-        counts: Dict[str, int] = defaultdict(int)
-        for inmem_layer in self.in_memory_layers:
-            counts[inmem_layer.kind] += 1
-        for hist_layer in self.historic_layers:
-            counts[hist_layer.kind] += 1
-        return counts
-
-
-@dataclass
-class TenantConfig:
-    tenant_specific_overrides: Dict[str, Any]
-    effective_config: Dict[str, Any]
-
-    @classmethod
-    def from_json(cls, d: Dict[str, Any]) -> TenantConfig:
-        return TenantConfig(
-            tenant_specific_overrides=d["tenant_specific_overrides"],
-            effective_config=d["effective_config"],
-        )
-
-
-class PageserverHttpClient(requests.Session):
-    def __init__(self, port: int, is_testing_enabled_or_skip: Fn, auth_token: Optional[str] = None):
-        super().__init__()
-        self.port = port
-        self.auth_token = auth_token
-        self.is_testing_enabled_or_skip = is_testing_enabled_or_skip
-
-        if auth_token is not None:
-            self.headers["Authorization"] = f"Bearer {auth_token}"
-
-    def verbose_error(self, res: requests.Response):
-        try:
-            res.raise_for_status()
-        except requests.RequestException as e:
-            try:
-                msg = res.json()["msg"]
-            except:  # noqa: E722
-                msg = ""
-            raise PageserverApiException(msg, res.status_code) from e
-
-    def check_status(self):
-        self.get(f"http://localhost:{self.port}/v1/status").raise_for_status()
-
-    def configure_failpoints(self, config_strings: Tuple[str, str] | List[Tuple[str, str]]):
-        self.is_testing_enabled_or_skip()
-
-        if isinstance(config_strings, tuple):
-            pairs = [config_strings]
-        else:
-            pairs = config_strings
-
-        log.info(f"Requesting config failpoints: {repr(pairs)}")
-
-        res = self.put(
-            f"http://localhost:{self.port}/v1/failpoints",
-            json=[{"name": name, "actions": actions} for name, actions in pairs],
-        )
-        log.info(f"Got failpoints request response code {res.status_code}")
-        self.verbose_error(res)
-        res_json = res.json()
-        assert res_json is None
-        return res_json
-
-    def tenant_list(self) -> List[Dict[Any, Any]]:
-        res = self.get(f"http://localhost:{self.port}/v1/tenant")
-        self.verbose_error(res)
-        res_json = res.json()
-        assert isinstance(res_json, list)
-        return res_json
-
-    def tenant_create(self, new_tenant_id: Optional[TenantId] = None) -> TenantId:
-        res = self.post(
-            f"http://localhost:{self.port}/v1/tenant",
-            json={
-                "new_tenant_id": str(new_tenant_id) if new_tenant_id else None,
-            },
-        )
-        self.verbose_error(res)
-        if res.status_code == 409:
-            raise Exception(f"could not create tenant: already exists for id {new_tenant_id}")
-        new_tenant_id = res.json()
-        assert isinstance(new_tenant_id, str)
-        return TenantId(new_tenant_id)
-
-    def tenant_attach(self, tenant_id: TenantId):
-        res = self.post(f"http://localhost:{self.port}/v1/tenant/{tenant_id}/attach")
-        self.verbose_error(res)
-
-    def tenant_detach(self, tenant_id: TenantId, detach_ignored=False):
-        params = {}
-        if detach_ignored:
-            params["detach_ignored"] = "true"
-
-        res = self.post(f"http://localhost:{self.port}/v1/tenant/{tenant_id}/detach", params=params)
-        self.verbose_error(res)
-
-    def tenant_load(self, tenant_id: TenantId):
-        res = self.post(f"http://localhost:{self.port}/v1/tenant/{tenant_id}/load")
-        self.verbose_error(res)
-
-    def tenant_ignore(self, tenant_id: TenantId):
-        res = self.post(f"http://localhost:{self.port}/v1/tenant/{tenant_id}/ignore")
-        self.verbose_error(res)
-
-    def tenant_status(self, tenant_id: TenantId) -> Dict[Any, Any]:
-        res = self.get(f"http://localhost:{self.port}/v1/tenant/{tenant_id}")
-        self.verbose_error(res)
-        res_json = res.json()
-        assert isinstance(res_json, dict)
-        return res_json
-
-    def tenant_config(self, tenant_id: TenantId) -> TenantConfig:
-        res = self.get(f"http://localhost:{self.port}/v1/tenant/{tenant_id}/config")
-        self.verbose_error(res)
-        return TenantConfig.from_json(res.json())
-
-    def set_tenant_config(self, tenant_id: TenantId, config: dict[str, Any]):
-        assert "tenant_id" not in config.keys()
-        res = self.put(
-            f"http://localhost:{self.port}/v1/tenant/config",
-            json={**config, "tenant_id": str(tenant_id)},
-        )
-        self.verbose_error(res)
-
-    def patch_tenant_config_client_side(
-        self,
-        tenant_id: TenantId,
-        inserts: Optional[Dict[str, Any]] = None,
-        removes: Optional[List[str]] = None,
-    ):
-        current = self.tenant_config(tenant_id).tenant_specific_overrides
-        if inserts is not None:
-            current.update(inserts)
-        if removes is not None:
-            for key in removes:
-                del current[key]
-        self.set_tenant_config(tenant_id, current)
-
-    def tenant_size(self, tenant_id: TenantId) -> int:
-        return self.tenant_size_and_modelinputs(tenant_id)[0]
-
-    def tenant_size_and_modelinputs(self, tenant_id: TenantId) -> Tuple[int, Dict[str, Any]]:
-        """
-        Returns the tenant size, together with the model inputs as the second tuple item.
-        """
-        res = self.get(f"http://localhost:{self.port}/v1/tenant/{tenant_id}/synthetic_size")
-        self.verbose_error(res)
-        res = res.json()
-        assert isinstance(res, dict)
-        assert TenantId(res["id"]) == tenant_id
-        size = res["size"]
-        assert type(size) == int
-        inputs = res["inputs"]
-        assert type(inputs) is dict
-        return (size, inputs)
-
-    def tenant_size_debug(self, tenant_id: TenantId) -> str:
-        """
-        Returns the tenant size debug info, as an HTML string
-        """
-        res = self.get(
-            f"http://localhost:{self.port}/v1/tenant/{tenant_id}/synthetic_size",
-            headers={"Accept": "text/html"},
-        )
-        return res.text
-
-    def timeline_list(
-        self,
-        tenant_id: TenantId,
-        include_non_incremental_logical_size: bool = False,
-        include_timeline_dir_layer_file_size_sum: bool = False,
-    ) -> List[Dict[str, Any]]:
-        params = {}
-        if include_non_incremental_logical_size:
-            params["include-non-incremental-logical-size"] = "true"
-        if include_timeline_dir_layer_file_size_sum:
-            params["include-timeline-dir-layer-file-size-sum"] = "true"
-
-        res = self.get(
-            f"http://localhost:{self.port}/v1/tenant/{tenant_id}/timeline", params=params
-        )
-        self.verbose_error(res)
-        res_json = res.json()
-        assert isinstance(res_json, list)
-        return res_json
-
-    def timeline_create(
-        self,
-        tenant_id: TenantId,
-        new_timeline_id: Optional[TimelineId] = None,
-        ancestor_timeline_id: Optional[TimelineId] = None,
-        ancestor_start_lsn: Optional[Lsn] = None,
-    ) -> Dict[Any, Any]:
-        res = self.post(
-            f"http://localhost:{self.port}/v1/tenant/{tenant_id}/timeline",
-            json={
-                "new_timeline_id": str(new_timeline_id) if new_timeline_id else None,
-                "ancestor_start_lsn": str(ancestor_start_lsn) if ancestor_start_lsn else None,
-                "ancestor_timeline_id": str(ancestor_timeline_id) if ancestor_timeline_id else None,
-            },
-        )
-        self.verbose_error(res)
-        if res.status_code == 409:
-            raise Exception(f"could not create timeline: already exists for id {new_timeline_id}")
-
-        res_json = res.json()
-        assert isinstance(res_json, dict)
-        return res_json
-
-    def timeline_detail(
-        self,
-        tenant_id: TenantId,
-        timeline_id: TimelineId,
-        include_non_incremental_logical_size: bool = False,
-        include_timeline_dir_layer_file_size_sum: bool = False,
-        **kwargs,
-    ) -> Dict[Any, Any]:
-        params = {}
-        if include_non_incremental_logical_size:
-            params["include-non-incremental-logical-size"] = "true"
-        if include_timeline_dir_layer_file_size_sum:
-            params["include-timeline-dir-layer-file-size-sum"] = "true"
-
-        res = self.get(
-            f"http://localhost:{self.port}/v1/tenant/{tenant_id}/timeline/{timeline_id}",
-            params=params,
-            **kwargs,
-        )
-        self.verbose_error(res)
-        res_json = res.json()
-        assert isinstance(res_json, dict)
-        return res_json
-
-    def timeline_delete(self, tenant_id: TenantId, timeline_id: TimelineId):
-        res = self.delete(
-            f"http://localhost:{self.port}/v1/tenant/{tenant_id}/timeline/{timeline_id}"
-        )
-        self.verbose_error(res)
-        res_json = res.json()
-        assert res_json is None
-
-    def timeline_gc(
-        self, tenant_id: TenantId, timeline_id: TimelineId, gc_horizon: Optional[int]
-    ) -> dict[str, Any]:
-        self.is_testing_enabled_or_skip()
-
-        log.info(
-            f"Requesting GC: tenant {tenant_id}, timeline {timeline_id}, gc_horizon {repr(gc_horizon)}"
-        )
-        res = self.put(
-            f"http://localhost:{self.port}/v1/tenant/{tenant_id}/timeline/{timeline_id}/do_gc",
-            json={"gc_horizon": gc_horizon},
-        )
-        log.info(f"Got GC request response code: {res.status_code}")
-        self.verbose_error(res)
-        res_json = res.json()
-        assert res_json is not None
-        assert isinstance(res_json, dict)
-        return res_json
-
-    def timeline_compact(self, tenant_id: TenantId, timeline_id: TimelineId):
-        self.is_testing_enabled_or_skip()
-
-        log.info(f"Requesting compact: tenant {tenant_id}, timeline {timeline_id}")
-        res = self.put(
-            f"http://localhost:{self.port}/v1/tenant/{tenant_id}/timeline/{timeline_id}/compact"
-        )
-        log.info(f"Got compact request response code: {res.status_code}")
-        self.verbose_error(res)
-        res_json = res.json()
-        assert res_json is None
-
-    def timeline_get_lsn_by_timestamp(
-        self, tenant_id: TenantId, timeline_id: TimelineId, timestamp
-    ):
-        log.info(
-            f"Requesting lsn by timestamp {timestamp}, tenant {tenant_id}, timeline {timeline_id}"
-        )
-        res = self.get(
-            f"http://localhost:{self.port}/v1/tenant/{tenant_id}/timeline/{timeline_id}/get_lsn_by_timestamp?timestamp={timestamp}",
-        )
-        self.verbose_error(res)
-        res_json = res.json()
-        return res_json
-
-    def timeline_checkpoint(self, tenant_id: TenantId, timeline_id: TimelineId):
-        self.is_testing_enabled_or_skip()
-
-        log.info(f"Requesting checkpoint: tenant {tenant_id}, timeline {timeline_id}")
-        res = self.put(
-            f"http://localhost:{self.port}/v1/tenant/{tenant_id}/timeline/{timeline_id}/checkpoint"
-        )
-        log.info(f"Got checkpoint request response code: {res.status_code}")
-        self.verbose_error(res)
-        res_json = res.json()
-        assert res_json is None
-
-    def timeline_spawn_download_remote_layers(
-        self,
-        tenant_id: TenantId,
-        timeline_id: TimelineId,
-        max_concurrent_downloads: int,
-    ) -> dict[str, Any]:
-        body = {
-            "max_concurrent_downloads": max_concurrent_downloads,
-        }
-        res = self.post(
-            f"http://localhost:{self.port}/v1/tenant/{tenant_id}/timeline/{timeline_id}/download_remote_layers",
-            json=body,
-        )
-        self.verbose_error(res)
-        res_json = res.json()
-        assert res_json is not None
-        assert isinstance(res_json, dict)
-        return res_json
-
-    def timeline_poll_download_remote_layers_status(
-        self,
-        tenant_id: TenantId,
-        timeline_id: TimelineId,
-        spawn_response: dict[str, Any],
-        poll_state=None,
-    ) -> None | dict[str, Any]:
-        res = self.get(
-            f"http://localhost:{self.port}/v1/tenant/{tenant_id}/timeline/{timeline_id}/download_remote_layers",
-        )
-        self.verbose_error(res)
-        res_json = res.json()
-        assert res_json is not None
-        assert isinstance(res_json, dict)
-
-        # assumption in this API client here is that nobody else spawns the task
-        assert res_json["task_id"] == spawn_response["task_id"]
-
-        if poll_state is None or res_json["state"] == poll_state:
-            return res_json
-        return None
-
-    def timeline_download_remote_layers(
-        self,
-        tenant_id: TenantId,
-        timeline_id: TimelineId,
-        max_concurrent_downloads: int,
-        errors_ok=False,
-        at_least_one_download=True,
-    ):
-        res = self.timeline_spawn_download_remote_layers(
-            tenant_id, timeline_id, max_concurrent_downloads
-        )
-        while True:
-            completed = self.timeline_poll_download_remote_layers_status(
-                tenant_id, timeline_id, res, poll_state="Completed"
-            )
-            if not completed:
-                time.sleep(0.1)
-                continue
-            if not errors_ok:
-                assert completed["failed_download_count"] == 0
-            if at_least_one_download:
-                assert completed["successful_download_count"] > 0
-            return completed
-
-    def get_metrics_str(self) -> str:
-        """You probably want to use get_metrics() instead."""
-        res = self.get(f"http://localhost:{self.port}/metrics")
-        self.verbose_error(res)
-        return res.text
-
-    def get_metrics(self) -> Metrics:
-        res = self.get_metrics_str()
-        return parse_metrics(res)
-
-    def get_timeline_metric(
-        self, tenant_id: TenantId, timeline_id: TimelineId, metric_name: str
-    ) -> float:
-        metrics = self.get_metrics()
-        return metrics.query_one(
-            metric_name,
-            filter={
-                "tenant_id": str(tenant_id),
-                "timeline_id": str(timeline_id),
-            },
-        ).value
-
-    def get_remote_timeline_client_metric(
-        self,
-        metric_name: str,
-        tenant_id: TenantId,
-        timeline_id: TimelineId,
-        file_kind: str,
-        op_kind: str,
-    ) -> Optional[float]:
-        metrics = self.get_metrics()
-        matches = metrics.query_all(
-            name=metric_name,
-            filter={
-                "tenant_id": str(tenant_id),
-                "timeline_id": str(timeline_id),
-                "file_kind": str(file_kind),
-                "op_kind": str(op_kind),
-            },
-        )
-        if len(matches) == 0:
-            value = None
-        elif len(matches) == 1:
-            value = matches[0].value
-            assert value is not None
-        else:
-            assert len(matches) < 2, "above filter should uniquely identify metric"
-        return value
-
-    def get_metric_value(
-        self, name: str, filter: Optional[Dict[str, str]] = None
-    ) -> Optional[float]:
-        metrics = self.get_metrics()
-        results = metrics.query_all(name, filter=filter)
-        if not results:
-            log.info(f'could not find metric "{name}"')
-            return None
-        assert len(results) == 1, f"metric {name} with given filters is not unique, got: {results}"
-        return results[0].value
-
-    def layer_map_info(
-        self,
-        tenant_id: TenantId,
-        timeline_id: TimelineId,
-    ) -> LayerMapInfo:
-        res = self.get(
-            f"http://localhost:{self.port}/v1/tenant/{tenant_id}/timeline/{timeline_id}/layer/",
-        )
-        self.verbose_error(res)
-        return LayerMapInfo.from_json(res.json())
-
-    def download_layer(self, tenant_id: TenantId, timeline_id: TimelineId, layer_name: str):
-        res = self.get(
-            f"http://localhost:{self.port}/v1/tenant/{tenant_id}/timeline/{timeline_id}/layer/{layer_name}",
-        )
-        self.verbose_error(res)
-
-        assert res.status_code == 200
-
-    def evict_layer(self, tenant_id: TenantId, timeline_id: TimelineId, layer_name: str):
-        res = self.delete(
-            f"http://localhost:{self.port}/v1/tenant/{tenant_id}/timeline/{timeline_id}/layer/{layer_name}",
-        )
-        self.verbose_error(res)
-
-        assert res.status_code == 200
-
-    def evict_all_layers(self, tenant_id: TenantId, timeline_id: TimelineId):
-        info = self.layer_map_info(tenant_id, timeline_id)
-        for layer in info.historic_layers:
-            self.evict_layer(tenant_id, timeline_id, layer.layer_file_name)
-
-    def disk_usage_eviction_run(self, request: dict[str, Any]):
-        res = self.put(
-            f"http://localhost:{self.port}/v1/disk_usage_eviction/run",
-            json=request,
-        )
-        self.verbose_error(res)
-        return res.json()
-
-    def tenant_break(self, tenant_id: TenantId):
-        res = self.put(f"http://localhost:{self.port}/v1/tenant/{tenant_id}/break")
-        self.verbose_error(res)
--- a/test_runner/fixtures/pageserver/utils.py
+++ b/test_runner/fixtures/pageserver/utils.py
@@ -1,145 +0,0 @@
-import time
-
-from fixtures.log_helper import log
-from fixtures.pageserver.http import PageserverHttpClient
-from fixtures.types import Lsn, TenantId, TimelineId
-
-
-def assert_tenant_status(
-    pageserver_http: PageserverHttpClient, tenant: TenantId, expected_status: str
-):
-    tenant_status = pageserver_http.tenant_status(tenant)
-    log.info(f"tenant_status: {tenant_status}")
-    assert tenant_status["state"] == expected_status, tenant_status
-
-
-def tenant_exists(pageserver_http: PageserverHttpClient, tenant_id: TenantId):
-    tenants = pageserver_http.tenant_list()
-    matching = [t for t in tenants if TenantId(t["id"]) == tenant_id]
-    assert len(matching) < 2
-    if len(matching) == 0:
-        return None
-    return matching[0]
-
-
-def remote_consistent_lsn(
-    pageserver_http: PageserverHttpClient, tenant: TenantId, timeline: TimelineId
-) -> Lsn:
-    detail = pageserver_http.timeline_detail(tenant, timeline)
-
-    if detail["remote_consistent_lsn"] is None:
-        # No remote information at all. This happens right after creating
-        # a timeline, before any part of it has been uploaded to remote
-        # storage yet.
-        return Lsn(0)
-    else:
-        lsn_str = detail["remote_consistent_lsn"]
-        assert isinstance(lsn_str, str)
-        return Lsn(lsn_str)
-
-
-def wait_for_upload(
-    pageserver_http: PageserverHttpClient,
-    tenant: TenantId,
-    timeline: TimelineId,
-    lsn: Lsn,
-):
-    """waits for local timeline upload up to specified lsn"""
-    for i in range(20):
-        current_lsn = remote_consistent_lsn(pageserver_http, tenant, timeline)
-        if current_lsn >= lsn:
-            log.info("wait finished")
-            return
-        log.info(
-            "waiting for remote_consistent_lsn to reach {}, now {}, iteration {}".format(
-                lsn, current_lsn, i + 1
-            )
-        )
-        time.sleep(1)
-    raise Exception(
-        "timed out while waiting for remote_consistent_lsn to reach {}, was {}".format(
-            lsn, current_lsn
-        )
-    )
-
-
-def wait_until_tenant_state(
-    pageserver_http: PageserverHttpClient,
-    tenant_id: TenantId,
-    expected_state: str,
-    iterations: int,
-) -> bool:
-    """
-    Does not use `wait_until` for debugging purposes
-    """
-    for _ in range(iterations):
-        try:
-            tenant = pageserver_http.tenant_status(tenant_id=tenant_id)
-            log.debug(f"Tenant {tenant_id} data: {tenant}")
-            if tenant["state"] == expected_state:
-                return True
-        except Exception as e:
-            log.debug(f"Tenant {tenant_id} state retrieval failure: {e}")
-
-        time.sleep(1)
-
-    raise Exception(f"Tenant {tenant_id} did not become {expected_state} in {iterations} seconds")
-
-
-def wait_until_tenant_active(
-    pageserver_http: PageserverHttpClient, tenant_id: TenantId, iterations: int = 30
-):
-    wait_until_tenant_state(
-        pageserver_http, tenant_id, expected_state="Active", iterations=iterations
-    )
-
-
-def last_record_lsn(
-    pageserver_http_client: PageserverHttpClient, tenant: TenantId, timeline: TimelineId
-) -> Lsn:
-    detail = pageserver_http_client.timeline_detail(tenant, timeline)
-
-    lsn_str = detail["last_record_lsn"]
-    assert isinstance(lsn_str, str)
-    return Lsn(lsn_str)
-
-
-def wait_for_last_record_lsn(
-    pageserver_http: PageserverHttpClient,
-    tenant: TenantId,
-    timeline: TimelineId,
-    lsn: Lsn,
-) -> Lsn:
-    """waits for pageserver to catch up to a certain lsn, returns the last observed lsn."""
-    for i in range(10):
-        current_lsn = last_record_lsn(pageserver_http, tenant, timeline)
-        if current_lsn >= lsn:
-            return current_lsn
-        log.info(
-            "waiting for last_record_lsn to reach {}, now {}, iteration {}".format(
-                lsn, current_lsn, i + 1
-            )
-        )
-        time.sleep(1)
-    raise Exception(
-        "timed out while waiting for last_record_lsn to reach {}, was {}".format(lsn, current_lsn)
-    )
-
-
-def wait_for_upload_queue_empty(
-    pageserver_http: PageserverHttpClient, tenant_id: TenantId, timeline_id: TimelineId
-):
-    while True:
-        all_metrics = pageserver_http.get_metrics()
-        tl = all_metrics.query_all(
-            "pageserver_remote_timeline_client_calls_unfinished",
-            {
-                "tenant_id": str(tenant_id),
-                "timeline_id": str(timeline_id),
-            },
-        )
-        assert len(tl) > 0
-        log.info(f"upload queue for {tenant_id}/{timeline_id}: {tl}")
-        if all(m.value == 0 for m in tl):
-            return
-        time.sleep(0.2)
--- a/test_runner/fixtures/utils.py
+++ b/test_runner/fixtures/utils.py
@@ -1,5 +1,4 @@
 import contextlib
-import json
 import os
 import re
 import subprocess
@@ -7,9 +6,8 @@ import tarfile
 import time
 from pathlib import Path
 from typing import Any, Callable, Dict, List, Tuple, TypeVar
-from urllib.parse import urlencode

-import allure
+import allure  # type: ignore
 from psycopg2.extensions import cursor

 from fixtures.log_helper import log
@@ -186,46 +184,6 @@ def allure_attach_from_dir(dir: Path):
            allure.attach.file(source, name, attachment_type, extension)


-DATASOURCE_ID = "xHHYY0dVz"
-
-
-def allure_add_grafana_links(host: str, start_ms: int, end_ms: int):
-    """Add links to server logs in Grafana to Allure report"""
-    # We expect host to be in format like ep-divine-night-159320.us-east-2.aws.neon.build
-    endpoint_id, region_id, _ = host.split(".", 2)
-
-    expressions = {
-        "compute logs": f'{{app="compute-node-{endpoint_id}", neon_region="{region_id}"}}',
-        "k8s events": f'{{job="integrations/kubernetes/eventhandler"}} |~ "name=compute-node-{endpoint_id}-"',
-        "console logs": f'{{neon_service="console", neon_region="{region_id}"}} | json | endpoint_id = "{endpoint_id}"',
-        "proxy logs": f'{{neon_service="proxy-scram", neon_region="{region_id}"}}',
-    }
-
-    params: Dict[str, Any] = {
-        "datasource": DATASOURCE_ID,
-        "queries": [
-            {
-                "expr": "<PUT AN EXPRESSION HERE>",
-                "refId": "A",
-                "datasource": {"type": "loki", "uid": DATASOURCE_ID},
-                "editorMode": "code",
-                "queryType": "range",
-            }
-        ],
-        "range": {
-            "from": str(start_ms),
-            "to": str(end_ms),
-        },
-    }
-    for name, expr in expressions.items():
-        params["queries"][0]["expr"] = expr
-        query_string = urlencode({"orgId": 1, "left": json.dumps(params)})
-        link = f"https://neonprod.grafana.net/explore?{query_string}"
-
-        allure.dynamic.link(link, name=name)
-        log.info(f"{name}: {link}")
-
-
 def start_in_background(
    command: list[str], cwd: Path, log_file_name: str, is_started: Fn
 ) -> subprocess.Popen[bytes]:
@@ -278,19 +236,3 @@ def wait_until(number_of_iterations: int, interval: float, func: Fn):
            continue
        return res
    raise Exception("timed out while waiting for %s" % func) from last_exception
-
-
-def wait_while(number_of_iterations: int, interval: float, func):
-    """
-    Wait until 'func' returns false, or throws an exception.
-    """
-    for i in range(number_of_iterations):
-        try:
-            if not func():
-                return
-            log.info("waiting for %s iteration %s failed", func, i + 1)
-            time.sleep(interval)
-            continue
-        except Exception:
-            return
-    raise Exception("timed out while waiting for %s" % func)
--- a/test_runner/performance/test_branch_creation.py
+++ b/test_runner/performance/test_branch_creation.py
@@ -10,7 +10,7 @@ import pytest
 from fixtures.benchmark_fixture import MetricReport
 from fixtures.compare_fixtures import NeonCompare
 from fixtures.log_helper import log
-from fixtures.pageserver.utils import wait_for_last_record_lsn
+from fixtures.neon_fixtures import wait_for_last_record_lsn
 from fixtures.types import Lsn


--- a/test_runner/regress/test_auth.py
+++ b/test_runner/regress/test_auth.py
@@ -1,8 +1,7 @@
 from contextlib import closing

 import pytest
-from fixtures.neon_fixtures import NeonEnvBuilder, PgProtocol
-from fixtures.pageserver.http import PageserverApiException
+from fixtures.neon_fixtures import NeonEnvBuilder, PageserverApiException, PgProtocol
 from fixtures.types import TenantId


--- a/test_runner/regress/test_compatibility.py
+++ b/test_runner/regress/test_compatibility.py
@@ -10,11 +10,12 @@ from fixtures.log_helper import log
 from fixtures.neon_fixtures import (
    NeonCli,
    NeonEnvBuilder,
+    PageserverHttpClient,
    PgBin,
    PortDistributor,
+    wait_for_last_record_lsn,
+    wait_for_upload,
 )
-from fixtures.pageserver.http import PageserverHttpClient
-from fixtures.pageserver.utils import wait_for_last_record_lsn, wait_for_upload
 from fixtures.types import Lsn
 from pytest import FixtureRequest

@@ -33,15 +34,9 @@ from pytest import FixtureRequest
 # - check_neon_works performs the test itself, feel free to add more checks there.
 #

-check_ondisk_data_compatibility_if_enabled = pytest.mark.skipif(
-    os.environ.get("CHECK_ONDISK_DATA_COMPATIBILITY") is None,
-    reason="CHECK_ONDISK_DATA_COMPATIBILITY env is not set",
-)
-

 # Note: if renaming this test, don't forget to update a reference to it in a workflow file:
 # "Upload compatibility snapshot" step in .github/actions/run-python-test-set/action.yml
-@check_ondisk_data_compatibility_if_enabled
@pytest.mark.xdist_group("compatibility")
@pytest.mark.order(before="test_forward_compatibility")
 def test_create_snapshot(neon_env_builder: NeonEnvBuilder, pg_bin: PgBin, test_output_dir: Path):
@@ -86,7 +81,6 @@ def test_create_snapshot(neon_env_builder: NeonEnvBuilder, pg_bin: PgBin, test_o
    # Directory `test_output_dir / "compatibility_snapshot_pg14"` is uploaded to S3 in a workflow, keep the name in sync with it


-@check_ondisk_data_compatibility_if_enabled
@pytest.mark.xdist_group("compatibility")
@pytest.mark.order(after="test_create_snapshot")
 def test_backward_compatibility(
@@ -140,7 +134,6 @@ def test_backward_compatibility(
    ), "Breaking changes are allowed by ALLOW_BACKWARD_COMPATIBILITY_BREAKAGE, but the test has passed without any breakage"


-@check_ondisk_data_compatibility_if_enabled
@pytest.mark.xdist_group("compatibility")
@pytest.mark.order(after="test_create_snapshot")
 def test_forward_compatibility(
--- a/test_runner/regress/test_disk_usage_eviction.py
+++ b/test_runner/regress/test_disk_usage_eviction.py
@@ -11,14 +11,14 @@ from fixtures.neon_fixtures import (
    LocalFsStorage,
    NeonEnv,
    NeonEnvBuilder,
+    PageserverHttpClient,
    PgBin,
    RemoteStorageKind,
    wait_for_last_flush_lsn,
+    wait_for_upload_queue_empty,
+    wait_until,
 )
-from fixtures.pageserver.http import PageserverHttpClient
-from fixtures.pageserver.utils import wait_for_upload_queue_empty
 from fixtures.types import Lsn, TenantId, TimelineId
-from fixtures.utils import wait_until

 GLOBAL_LRU_LOG_LINE = "tenant_min_resident_size-respecting LRU would not relieve pressure, evicting more following global LRU policy"

@@ -138,7 +138,7 @@ def eviction_env(request, neon_env_builder: NeonEnvBuilder, pg_bin: PgBin) -> Ev
    # remove the initial tenant
    ## why wait for upload queue? => https://github.com/neondatabase/neon/issues/3865
    assert env.initial_timeline
-    wait_for_upload_queue_empty(pageserver_http, env.initial_tenant, env.initial_timeline)
+    wait_for_upload_queue_empty(env.pageserver, env.initial_tenant, env.initial_timeline)
    pageserver_http.tenant_detach(env.initial_tenant)
    assert isinstance(env.remote_storage, LocalFsStorage)
    tenant_remote_storage = env.remote_storage.root / "tenants" / str(env.initial_tenant)
@@ -182,7 +182,7 @@ def eviction_env(request, neon_env_builder: NeonEnvBuilder, pg_bin: PgBin) -> Ev
    # after stopping the safekeepers, we know that no new WAL will be coming in
    for tenant_id, timeline_id in timelines:
        pageserver_http.timeline_checkpoint(tenant_id, timeline_id)
-        wait_for_upload_queue_empty(pageserver_http, tenant_id, timeline_id)
+        wait_for_upload_queue_empty(env.pageserver, tenant_id, timeline_id)
        tl_info = pageserver_http.timeline_detail(tenant_id, timeline_id)
        assert tl_info["last_record_lsn"] == tl_info["disk_consistent_lsn"]
        assert tl_info["disk_consistent_lsn"] == tl_info["remote_consistent_lsn"]
--- a/test_runner/regress/test_import.py
+++ b/test_runner/regress/test_import.py
@@ -13,8 +13,9 @@ from fixtures.neon_fixtures import (
    NeonEnvBuilder,
    PgBin,
    Postgres,
+    wait_for_last_record_lsn,
+    wait_for_upload,
 )
-from fixtures.pageserver.utils import wait_for_last_record_lsn, wait_for_upload
 from fixtures.types import Lsn, TenantId, TimelineId
 from fixtures.utils import subprocess_capture

--- a/test_runner/regress/test_layer_eviction.py
+++ b/test_runner/regress/test_layer_eviction.py
@@ -6,9 +6,10 @@ from fixtures.neon_fixtures import (
    NeonEnvBuilder,
    RemoteStorageKind,
    wait_for_last_flush_lsn,
+    wait_for_last_record_lsn,
    wait_for_sk_commit_lsn_to_reach_remote_storage,
+    wait_for_upload,
 )
-from fixtures.pageserver.utils import wait_for_last_record_lsn, wait_for_upload
 from fixtures.types import Lsn, TenantId, TimelineId
 from fixtures.utils import query_scalar

--- a/test_runner/regress/test_neon_cli.py
+++ b/test_runner/regress/test_neon_cli.py
@@ -5,8 +5,8 @@ from fixtures.neon_fixtures import (
    DEFAULT_BRANCH_NAME,
    NeonEnv,
    NeonEnvBuilder,
+    PageserverHttpClient,
 )
-from fixtures.pageserver.http import PageserverHttpClient
 from fixtures.types import TenantId, TimelineId


--- a/test_runner/regress/test_normal_work.py
+++ b/test_runner/regress/test_normal_work.py
@@ -1,7 +1,6 @@
 import pytest
 from fixtures.log_helper import log
-from fixtures.neon_fixtures import NeonEnv, NeonEnvBuilder
-from fixtures.pageserver.http import PageserverHttpClient
+from fixtures.neon_fixtures import NeonEnv, NeonEnvBuilder, PageserverHttpClient


 def check_tenant(env: NeonEnv, pageserver_http: PageserverHttpClient):
--- a/test_runner/regress/test_ondemand_download.py
+++ b/test_runner/regress/test_ondemand_download.py
@@ -10,20 +10,20 @@ import pytest
 from fixtures.log_helper import log
 from fixtures.neon_fixtures import (
    NeonEnvBuilder,
+    PageserverApiException,
+    PageserverHttpClient,
    RemoteStorageKind,
+    assert_tenant_status,
    available_remote_storages,
    wait_for_last_flush_lsn,
-    wait_for_sk_commit_lsn_to_reach_remote_storage,
-)
-from fixtures.pageserver.http import PageserverApiException, PageserverHttpClient
-from fixtures.pageserver.utils import (
-    assert_tenant_status,
    wait_for_last_record_lsn,
+    wait_for_sk_commit_lsn_to_reach_remote_storage,
    wait_for_upload,
+    wait_until,
    wait_until_tenant_state,
 )
 from fixtures.types import Lsn
-from fixtures.utils import query_scalar, wait_until
+from fixtures.utils import query_scalar


 def get_num_downloaded_layers(client: PageserverHttpClient, tenant_id, timeline_id):
@@ -63,7 +63,7 @@ def test_ondemand_download_large_rel(
    tenant, _ = env.neon_cli.create_tenant(
        conf={
            # disable background GC
-            "gc_period": "0s",
+            "gc_period": "10 m",
            "gc_horizon": f"{10 * 1024 ** 3}",  # 10 GB
            # small checkpoint distance to create more delta layer files
            "checkpoint_distance": f"{10 * 1024 ** 2}",  # 10 MB
--- a/test_runner/regress/test_pageserver_api.py
+++ b/test_runner/regress/test_pageserver_api.py
@@ -6,8 +6,8 @@ from fixtures.neon_fixtures import (
    DEFAULT_BRANCH_NAME,
    NeonEnv,
    NeonEnvBuilder,
+    PageserverHttpClient,
 )
-from fixtures.pageserver.http import PageserverHttpClient
 from fixtures.types import Lsn, TenantId, TimelineId
 from fixtures.utils import wait_until

--- a/test_runner/regress/test_read_trace.py
+++ b/test_runner/regress/test_read_trace.py
@@ -1,7 +1,6 @@
 from contextlib import closing

-from fixtures.neon_fixtures import NeonEnvBuilder
-from fixtures.pageserver.utils import wait_for_last_record_lsn
+from fixtures.neon_fixtures import NeonEnvBuilder, wait_for_last_record_lsn
 from fixtures.types import Lsn, TenantId, TimelineId
 from fixtures.utils import query_scalar

--- a/test_runner/regress/test_readonly_node.py
+++ b/test_runner/regress/test_readonly_node.py
@@ -1,7 +1,6 @@
 import pytest
 from fixtures.log_helper import log
-from fixtures.neon_fixtures import NeonEnv
-from fixtures.pageserver.utils import wait_for_last_record_lsn
+from fixtures.neon_fixtures import NeonEnv, wait_for_last_record_lsn
 from fixtures.types import Lsn
 from fixtures.utils import query_scalar

--- a/test_runner/regress/test_remote_storage.py
+++ b/test_runner/regress/test_remote_storage.py
@@ -13,15 +13,13 @@ from fixtures.log_helper import log
 from fixtures.neon_fixtures import (
    LocalFsStorage,
    NeonEnvBuilder,
+    PageserverApiException,
+    PageserverHttpClient,
    RemoteStorageKind,
    available_remote_storages,
    wait_for_last_flush_lsn,
-)
-from fixtures.pageserver.http import PageserverApiException, PageserverHttpClient
-from fixtures.pageserver.utils import (
    wait_for_last_record_lsn,
    wait_for_upload,
-    wait_until_tenant_active,
    wait_until_tenant_state,
 )
 from fixtures.types import Lsn, TenantId, TimelineId
@@ -174,10 +172,15 @@ def test_remote_storage_backup_and_restore(
        client.tenant_attach(tenant_id)
    log.info("waiting for tenant to become active. this should be quick with on-demand download")

-    wait_until_tenant_active(
-        pageserver_http=client,
-        tenant_id=tenant_id,
-        iterations=5,
+    def tenant_active():
+        all_states = client.tenant_list()
+        [tenant] = [t for t in all_states if TenantId(t["id"]) == tenant_id]
+        assert tenant["state"] == "Active"
+
+    wait_until(
+        number_of_iterations=5,
+        interval=1,
+        func=tenant_active,
    )

    detail = client.timeline_detail(tenant_id, timeline_id)
@@ -354,7 +357,12 @@ def test_remote_storage_upload_queue_retries(

    client.tenant_attach(tenant_id)

-    wait_until_tenant_active(client, tenant_id)
+    def tenant_active():
+        all_states = client.tenant_list()
+        [tenant] = [t for t in all_states if TenantId(t["id"]) == tenant_id]
+        assert tenant["state"] == "Active"
+
+    wait_until(30, 1, tenant_active)

    log.info("restarting postgres to validate")
    pg = env.postgres.create_start("main", tenant_id=tenant_id)
@@ -489,7 +497,12 @@ def test_remote_timeline_client_calls_started_metric(

    client.tenant_attach(tenant_id)

-    wait_until_tenant_active(client, tenant_id)
+    def tenant_active():
+        all_states = client.tenant_list()
+        [tenant] = [t for t in all_states if TenantId(t["id"]) == tenant_id]
+        assert tenant["state"] == "Active"
+
+    wait_until(30, 1, tenant_active)

    log.info("restarting postgres to validate")
    pg = env.postgres.create_start("main", tenant_id=tenant_id)
--- a/test_runner/regress/test_tenant_conf.py
+++ b/test_runner/regress/test_tenant_conf.py
@@ -6,8 +6,9 @@ from fixtures.neon_fixtures import (
    LocalFsStorage,
    NeonEnvBuilder,
    RemoteStorageKind,
+    assert_tenant_status,
+    wait_for_upload,
 )
-from fixtures.pageserver.utils import assert_tenant_status, wait_for_upload
 from fixtures.types import Lsn
 from fixtures.utils import wait_until

--- a/test_runner/regress/test_tenant_detach.py
+++ b/test_runner/regress/test_tenant_detach.py
@@ -9,18 +9,18 @@ from fixtures.log_helper import log
 from fixtures.neon_fixtures import (
    NeonEnv,
    NeonEnvBuilder,
+    PageserverApiException,
+    PageserverHttpClient,
    Postgres,
    RemoteStorageKind,
    available_remote_storages,
-)
-from fixtures.pageserver.http import PageserverApiException, PageserverHttpClient
-from fixtures.pageserver.utils import (
    wait_for_last_record_lsn,
    wait_for_upload,
+    wait_until,
    wait_until_tenant_state,
 )
 from fixtures.types import Lsn, TenantId, TimelineId
-from fixtures.utils import query_scalar, wait_until
+from fixtures.utils import query_scalar


 def do_gc_target(
--- a/test_runner/regress/test_tenant_relocation.py
+++ b/test_runner/regress/test_tenant_relocation.py
@@ -10,24 +10,18 @@ from fixtures.neon_fixtures import (
    NeonBroker,
    NeonEnv,
    NeonEnvBuilder,
+    PageserverHttpClient,
    PortDistributor,
    Postgres,
-)
-from fixtures.pageserver.http import PageserverHttpClient
-from fixtures.pageserver.utils import (
    assert_tenant_status,
    tenant_exists,
    wait_for_last_record_lsn,
    wait_for_upload,
-)
-from fixtures.types import Lsn, TenantId, TimelineId
-from fixtures.utils import (
-    query_scalar,
-    start_in_background,
-    subprocess_capture,
    wait_until,
    wait_while,
 )
+from fixtures.types import Lsn, TenantId, TimelineId
+from fixtures.utils import query_scalar, start_in_background, subprocess_capture


 def assert_abs_margin_ratio(a: float, b: float, margin_ratio: float):
--- a/test_runner/regress/test_tenant_size.py
+++ b/test_runner/regress/test_tenant_size.py
@@ -6,11 +6,11 @@ from fixtures.log_helper import log
 from fixtures.neon_fixtures import (
    NeonEnv,
    NeonEnvBuilder,
+    PageserverHttpClient,
    Postgres,
    wait_for_last_flush_lsn,
    wait_for_wal_insert_lsn,
 )
-from fixtures.pageserver.http import PageserverHttpClient
 from fixtures.types import Lsn, TenantId, TimelineId


--- a/test_runner/regress/test_tenants_with_remote_storage.py
+++ b/test_runner/regress/test_tenants_with_remote_storage.py
@@ -20,12 +20,10 @@ from fixtures.neon_fixtures import (
    NeonEnvBuilder,
    Postgres,
    RemoteStorageKind,
-    available_remote_storages,
-    wait_for_sk_commit_lsn_to_reach_remote_storage,
-)
-from fixtures.pageserver.utils import (
    assert_tenant_status,
+    available_remote_storages,
    wait_for_last_record_lsn,
+    wait_for_sk_commit_lsn_to_reach_remote_storage,
    wait_for_upload,
 )
 from fixtures.types import Lsn, TenantId, TimelineId
--- a/test_runner/regress/test_timeline_delete.py
+++ b/test_runner/regress/test_timeline_delete.py
@@ -1,6 +1,5 @@
 import pytest
-from fixtures.neon_fixtures import NeonEnv
-from fixtures.pageserver.http import PageserverApiException
+from fixtures.neon_fixtures import NeonEnv, PageserverApiException
 from fixtures.types import TenantId, TimelineId
 from fixtures.utils import wait_until

--- a/test_runner/regress/test_timeline_size.py
+++ b/test_runner/regress/test_timeline_size.py
@@ -14,21 +14,20 @@ from fixtures.log_helper import log
 from fixtures.neon_fixtures import (
    NeonEnv,
    NeonEnvBuilder,
+    PageserverApiException,
+    PageserverHttpClient,
    PgBin,
    PortDistributor,
    Postgres,
    RemoteStorageKind,
    VanillaPostgres,
-    wait_for_last_flush_lsn,
-)
-from fixtures.pageserver.http import PageserverApiException, PageserverHttpClient
-from fixtures.pageserver.utils import (
    assert_tenant_status,
+    wait_for_last_flush_lsn,
    wait_for_upload_queue_empty,
-    wait_until_tenant_active,
+    wait_until,
 )
 from fixtures.types import TenantId, TimelineId
-from fixtures.utils import get_timeline_dir_size, wait_until
+from fixtures.utils import get_timeline_dir_size


 def test_timeline_size(neon_simple_env: NeonEnv):
@@ -247,7 +246,12 @@ def test_timeline_initial_logical_size_calculation_cancellation(
        extra_env_vars={"FAILPOINTS": "timeline-calculate-logical-size-pause=pause"}
    )

-    wait_until_tenant_active(client, tenant_id)
+    def tenant_active():
+        all_states = client.tenant_list()
+        [tenant] = [t for t in all_states if TenantId(t["id"]) == tenant_id]
+        assert tenant["state"] == "Active"
+
+    wait_until(30, 1, tenant_active)

    # kick off initial size calculation task (the response we get here is the estimated size)
    def assert_size_calculation_not_done():
@@ -421,7 +425,7 @@ def test_timeline_physical_size_post_compaction(
    pageserver_http.timeline_compact(env.initial_tenant, new_timeline_id)

    if remote_storage_kind is not None:
-        wait_for_upload_queue_empty(pageserver_http, env.initial_tenant, new_timeline_id)
+        wait_for_upload_queue_empty(env.pageserver, env.initial_tenant, new_timeline_id)

    assert_physical_size_invariants(
        get_physical_size_values(env, env.initial_tenant, new_timeline_id, remote_storage_kind),
@@ -474,7 +478,7 @@ def test_timeline_physical_size_post_gc(
    pageserver_http.timeline_gc(env.initial_tenant, new_timeline_id, gc_horizon=None)

    if remote_storage_kind is not None:
-        wait_for_upload_queue_empty(pageserver_http, env.initial_tenant, new_timeline_id)
+        wait_for_upload_queue_empty(env.pageserver, env.initial_tenant, new_timeline_id)

    assert_physical_size_invariants(
        get_physical_size_values(env, env.initial_tenant, new_timeline_id, remote_storage_kind),
@@ -580,7 +584,7 @@ def test_tenant_physical_size(

    tenant, timeline = env.neon_cli.create_tenant()
    if remote_storage_kind is not None:
-        wait_for_upload_queue_empty(pageserver_http, tenant, timeline)
+        wait_for_upload_queue_empty(env.pageserver, tenant, timeline)

    def get_timeline_resident_physical_size(timeline: TimelineId):
        sizes = get_physical_size_values(env, tenant, timeline, remote_storage_kind)
@@ -605,7 +609,7 @@ def test_tenant_physical_size(
        pageserver_http.timeline_checkpoint(tenant, timeline)

        if remote_storage_kind is not None:
-            wait_for_upload_queue_empty(pageserver_http, tenant, timeline)
+            wait_for_upload_queue_empty(env.pageserver, tenant, timeline)

        timeline_total_resident_physical_size += get_timeline_resident_physical_size(timeline)

--- a/test_runner/regress/test_wal_acceptor.py
+++ b/test_runner/regress/test_wal_acceptor.py
@@ -30,8 +30,9 @@ from fixtures.neon_fixtures import (
    SafekeeperHttpClient,
    SafekeeperPort,
    available_remote_storages,
+    wait_for_last_record_lsn,
+    wait_for_upload,
 )
-from fixtures.pageserver.utils import wait_for_last_record_lsn, wait_for_upload
 from fixtures.types import Lsn, TenantId, TimelineId
 from fixtures.utils import get_dir_size, query_scalar, start_in_background

--- a/test_runner/regress/test_walredo_not_left_behind_on_detach.py
+++ b/test_runner/regress/test_walredo_not_left_behind_on_detach.py
@@ -3,8 +3,7 @@ import time
 import psutil
 import pytest
 from fixtures.log_helper import log
-from fixtures.neon_fixtures import NeonEnvBuilder
-from fixtures.pageserver.http import PageserverApiException
+from fixtures.neon_fixtures import NeonEnvBuilder, PageserverApiException
 from fixtures.types import TenantId
Author	SHA1	Message	Date
Dmitry Ivanov	3f8751191b	Simplify SNI parsing	2023-04-05 13:03:16 +03:00
Dmitry Ivanov	aba8cec279	More logging	2023-04-04 21:14:30 +03:00
Dmitry Ivanov	67e1d6f6fc	Finally, build TLS config in proxy's main	2023-04-04 21:02:23 +03:00
Dmitry Ivanov	febce3903a	Implement GlobMap for cert resolution	2023-04-04 19:58:56 +03:00
Dmitry Ivanov	a271ca6c8c	Properly extract cert names	2023-04-03 22:22:48 +03:00
Dmitry Ivanov	cee9c726d2	Implement proper parsing	2023-04-03 20:27:42 +03:00
Dmitry Ivanov	a12c85449a	WIP	2023-04-03 20:27:42 +03:00