Fix calculation of logical delta layer space

Co-authored-by: Heikki Linnakangas <heikki@neon.tech>

.cargo/config.toml

@@ -14,4 +14,3 @@ opt-level = 1
 
 [alias]
 build_testing = ["build", "--features", "testing"]
-neon = ["run", "--bin", "neon_local"]

.github/actions/allure-report-generate/action.yml

@@ -147,8 +147,6 @@ runs:
         echo "report-url=${REPORT_URL}"                                   >> $GITHUB_OUTPUT
         echo "report-json-url=${REPORT_URL%/index.html}/data/suites.json" >> $GITHUB_OUTPUT
 
-        echo "[Allure Report](${REPORT_URL})" >> ${GITHUB_STEP_SUMMARY}
-
     - name: Release lock
       if: always()
       shell: bash -euxo pipefail {0}

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

@@ -48,10 +48,6 @@ inputs:
     description: 'Whether to rerun flaky tests'
     required: false
     default: 'false'
-  pg_version:
-    description: 'Postgres version to use for tests'
-    required: false
-    default: 'v14'
 
 runs:
   using: "composite"
@@ -71,12 +67,12 @@ runs:
         path: /tmp/neon-previous
         prefix: latest
 
-    - name: Download compatibility snapshot
+    - name: Download compatibility snapshot for Postgres 14
       if: inputs.build_type != 'remote'
       uses: ./.github/actions/download
       with:
-        name: compatibility-snapshot-${{ inputs.build_type }}-pg${{ inputs.pg_version }}
-        path: /tmp/compatibility_snapshot_pg${{ inputs.pg_version }}
+        name: compatibility-snapshot-${{ inputs.build_type }}-pg14
+        path: /tmp/compatibility_snapshot_pg14
         prefix: latest
 
     - name: Checkout
@@ -106,18 +102,17 @@ runs:
         BUILD_TYPE: ${{ inputs.build_type }}
         AWS_ACCESS_KEY_ID: ${{ inputs.real_s3_access_key_id }}
         AWS_SECRET_ACCESS_KEY: ${{ inputs.real_s3_secret_access_key }}
-        COMPATIBILITY_SNAPSHOT_DIR: /tmp/compatibility_snapshot_pg${{ inputs.pg_version }}
+        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 }}
-        PG_VERSION: ${{ inputs.pg_version }}
       shell: bash -euxo pipefail {0}
       run: |
         # PLATFORM will be embedded in the perf test report
         # and it is needed to distinguish different environments
         export PLATFORM=${PLATFORM:-github-actions-selfhosted}
         export POSTGRES_DISTRIB_DIR=${POSTGRES_DISTRIB_DIR:-/tmp/neon/pg_install}
-        export DEFAULT_PG_VERSION=${PG_VERSION#v}
+        export DEFAULT_PG_VERSION=${DEFAULT_PG_VERSION:-14}
 
         if [ "${BUILD_TYPE}" = "remote" ]; then
           export REMOTE_ENV=1
@@ -197,13 +192,13 @@ runs:
           scripts/generate_and_push_perf_report.sh
         fi
 
-    - name: Upload compatibility snapshot
+    - name: Upload compatibility snapshot for Postgres 14
       if: github.ref_name == 'release'
       uses: ./.github/actions/upload
       with:
-        name: compatibility-snapshot-${{ inputs.build_type }}-pg${{ inputs.pg_version }}-${{ github.run_id }}
+        name: compatibility-snapshot-${{ inputs.build_type }}-pg14-${{ github.run_id }}
         # Directory is created by test_compatibility.py::test_create_snapshot, keep the path in sync with the test
-        path: /tmp/test_output/compatibility_snapshot_pg${{ inputs.pg_version }}/
+        path: /tmp/test_output/compatibility_snapshot_pg14/
         prefix: latest
 
     - name: Upload test results

.github/pull_request_template.md

@@ -1,6 +1,6 @@
-## Problem
+## Describe your changes
 
-## Summary of changes
+## Issue ticket number and link
 
 ## Checklist before requesting a review
 

.github/workflows/benchmarking.yml

@@ -16,12 +16,12 @@ on:
   workflow_dispatch: # adds ability to run this manually
     inputs:
       region_id:
-        description: 'Project region id. If not set, the default region will be used'
+        description: 'Use a particular region. If not set the default region will be used'
         required: false
         default: 'aws-us-east-2'
       save_perf_report:
         type: boolean
-        description: 'Publish perf report. If not set, the report will be published only for the main branch'
+        description: 'Publish perf report or not. If not set, the report is published only for the main branch'
         required: false
 
 defaults:
@@ -125,14 +125,13 @@ jobs:
         matrix='{
           "platform": [
             "neon-captest-new",
-            "neon-captest-reuse",
-            "neonvm-captest-new"
+            "neon-captest-reuse"
           ],
           "db_size": [ "10gb" ],
-          "include": [{ "platform": "neon-captest-freetier",   "db_size": "3gb"  },
-                      { "platform": "neon-captest-new",        "db_size": "50gb" },
-                      { "platform": "neonvm-captest-freetier", "db_size": "3gb"  },
-                      { "platform": "neonvm-captest-new",      "db_size": "50gb" }]
+          "include": [
+            { "platform": "neon-captest-freetier", "db_size": "3gb"  },
+            { "platform": "neon-captest-new",      "db_size": "50gb" }
+          ]
         }'
 
         if [ "$(date +%A)" = "Saturday" ]; then
@@ -198,7 +197,7 @@ jobs:
         echo "${POSTGRES_DISTRIB_DIR}/v${DEFAULT_PG_VERSION}/bin" >> $GITHUB_PATH
 
     - name: Create Neon Project
-      if: contains(fromJson('["neon-captest-new", "neon-captest-freetier", "neonvm-captest-new", "neonvm-captest-freetier"]'), matrix.platform)
+      if: contains(fromJson('["neon-captest-new", "neon-captest-freetier"]'), matrix.platform)
       id: create-neon-project
       uses: ./.github/actions/neon-project-create
       with:
@@ -206,7 +205,6 @@ jobs:
         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]' }}
-        provisioner: ${{ (contains(matrix.platform, 'neonvm-') && 'k8s-neonvm') || 'k8s-pod' }}
 
     - name: Set up Connection String
       id: set-up-connstr
@@ -215,7 +213,7 @@ jobs:
           neon-captest-reuse)
             CONNSTR=${{ secrets.BENCHMARK_CAPTEST_CONNSTR }}
             ;;
-          neon-captest-new | neon-captest-freetier | neonvm-captest-new | neonvm-captest-freetier)
+          neon-captest-new | neon-captest-freetier)
             CONNSTR=${{ steps.create-neon-project.outputs.dsn }}
             ;;
           rds-aurora)
@@ -225,7 +223,7 @@ jobs:
             CONNSTR=${{ secrets.BENCHMARK_RDS_POSTGRES_CONNSTR }}
             ;;
           *)
-            echo >&2 "Unknown PLATFORM=${PLATFORM}"
+            echo >&2 "Unknown PLATFORM=${PLATFORM}. Allowed only 'neon-captest-reuse', 'neon-captest-new', 'neon-captest-freetier', 'rds-aurora', or 'rds-postgres'"
             exit 1
             ;;
         esac

.github/workflows/build_and_test.yml

@@ -324,8 +324,7 @@ jobs:
     runs-on: [ self-hosted, gen3, large ]
     container:
       image: 369495373322.dkr.ecr.eu-central-1.amazonaws.com/rust:pinned
-      # Default shared memory is 64mb
-      options: --init --shm-size=512mb
+      options: --init
     needs: [ build-neon ]
     strategy:
       fail-fast: false
@@ -351,8 +350,8 @@ jobs:
           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
-          pg_version: ${{ matrix.pg_version }}
         env:
+          DEFAULT_PG_VERSION: ${{ matrix.pg_version }}
           TEST_RESULT_CONNSTR: ${{ secrets.REGRESS_TEST_RESULT_CONNSTR }}
           CHECK_ONDISK_DATA_COMPATIBILITY: nonempty
 
@@ -364,8 +363,7 @@ jobs:
     runs-on: [ self-hosted, gen3, small ]
     container:
       image: 369495373322.dkr.ecr.eu-central-1.amazonaws.com/rust:pinned
-      # Default shared memory is 64mb
-      options: --init --shm-size=512mb
+      options: --init
     needs: [ build-neon ]
     if: github.ref_name == 'main' || contains(github.event.pull_request.labels.*.name, 'run-benchmarks')
     strategy:
@@ -492,43 +490,37 @@ jobs:
       - name: Merge coverage data
         run: scripts/coverage "--profraw-prefix=$GITHUB_JOB" --dir=/tmp/coverage merge
 
-      - name: Build coverage report
-        env:
-          COMMIT_URL: ${{ github.server_url }}/${{ github.repository }}/commit/${{ github.event.pull_request.head.sha || github.sha }}
+      - name: Build and upload coverage report
         run: |
+          COMMIT_SHA=${{ github.event.pull_request.head.sha }}
+          COMMIT_SHA=${COMMIT_SHA:-${{ github.sha }}}
+          COMMIT_URL=https://github.com/${{ github.repository }}/commit/$COMMIT_SHA
+
           scripts/coverage \
             --dir=/tmp/coverage report \
             --input-objects=/tmp/coverage/binaries.list \
-            --commit-url=${COMMIT_URL} \
+            --commit-url=$COMMIT_URL \
             --format=github
 
-      - name: Upload coverage report
-        id: upload-coverage-report
-        env:
-          BUCKET: neon-github-public-dev
-          COMMIT_SHA: ${{ github.event.pull_request.head.sha || github.sha }}
-        run: |
-          aws s3 cp --only-show-errors --recursive /tmp/coverage/report s3://neon-github-public-dev/code-coverage/${COMMIT_SHA}
+          REPORT_URL=https://${{ github.repository_owner }}.github.io/zenith-coverage-data/$COMMIT_SHA
 
-          REPORT_URL=https://${BUCKET}.s3.amazonaws.com/code-coverage/${COMMIT_SHA}/index.html
-          echo "report-url=${REPORT_URL}" >> $GITHUB_OUTPUT
+          scripts/git-upload \
+            --repo=https://${{ secrets.VIP_VAP_ACCESS_TOKEN }}@github.com/${{ github.repository_owner }}/zenith-coverage-data.git \
+            --message="Add code coverage for $COMMIT_URL" \
+            copy /tmp/coverage/report $COMMIT_SHA # COPY FROM TO_RELATIVE
 
-      - uses: actions/github-script@v6
-        env:
-          REPORT_URL: ${{ steps.upload-coverage-report.outputs.report-url }}
-          COMMIT_SHA: ${{ github.event.pull_request.head.sha || github.sha }}
-        with:
-          script: |
-            const { REPORT_URL, COMMIT_SHA } = process.env
-
-            await github.rest.repos.createCommitStatus({
-              owner: context.repo.owner,
-              repo: context.repo.repo,
-              sha: `${COMMIT_SHA}`,
-              state: 'success',
-              target_url: `${REPORT_URL}`,
-              context: 'Code coverage report',
-            })
+          # Add link to the coverage report to the commit
+          curl -f -X POST \
+          https://api.github.com/repos/${{ github.repository }}/statuses/$COMMIT_SHA \
+          -H "Accept: application/vnd.github.v3+json" \
+          --user "${{ secrets.CI_ACCESS_TOKEN }}" \
+          --data \
+            "{
+              \"state\": \"success\",
+              \"context\": \"neon-coverage\",
+              \"description\": \"Coverage report is ready\",
+              \"target_url\": \"$REPORT_URL\"
+            }"
 
   trigger-e2e-tests:
     runs-on: [ self-hosted, gen3, small ]
@@ -711,11 +703,7 @@ jobs:
 
   compute-node-image:
     runs-on: [ self-hosted, gen3, large ]
-    container:
-      image: gcr.io/kaniko-project/executor:v1.9.2-debug
-      # Workaround for "Resolving download.osgeo.org (download.osgeo.org)... failed: Temporary failure in name resolution.""
-      # Should be prevented by https://github.com/neondatabase/neon/issues/4281
-      options: --add-host=download.osgeo.org:140.211.15.30
+    container: gcr.io/kaniko-project/executor:v1.9.2-debug
     needs: [ tag ]
     strategy:
       fail-fast: false
@@ -777,7 +765,7 @@ jobs:
       run:
         shell: sh -eu {0}
     env:
-      VM_BUILDER_VERSION: v0.7.3-alpha3
+      VM_BUILDER_VERSION: v0.4.6
 
     steps:
       - name: Checkout
@@ -787,18 +775,21 @@ jobs:
 
       - name: Downloading vm-builder
         run: |
-          curl -fL https://github.com/neondatabase/autoscaling/releases/download/$VM_BUILDER_VERSION/vm-builder -o vm-builder
+          curl -L https://github.com/neondatabase/neonvm/releases/download/$VM_BUILDER_VERSION/vm-builder -o vm-builder
           chmod +x vm-builder
 
-      # Note: we need a separate pull step here because otherwise vm-builder will try to pull, and
-      # it won't have the proper authentication (written at v0.6.0)
       - name: Pulling compute-node image
         run: |
           docker pull 369495373322.dkr.ecr.eu-central-1.amazonaws.com/compute-node-${{ matrix.version }}:${{needs.tag.outputs.build-tag}}
 
+      - name: Building VM compute-node rootfs
+        run: |
+          docker build -t temp-vm-compute-node --build-arg SRC_IMAGE=369495373322.dkr.ecr.eu-central-1.amazonaws.com/compute-node-${{ matrix.version }}:${{needs.tag.outputs.build-tag}} -f Dockerfile.vm-compute-node .
+
       - name: Build vm image
         run: |
-          ./vm-builder -src=369495373322.dkr.ecr.eu-central-1.amazonaws.com/compute-node-${{ matrix.version }}:${{needs.tag.outputs.build-tag}} -dst=369495373322.dkr.ecr.eu-central-1.amazonaws.com/vm-compute-node-${{ matrix.version }}:${{needs.tag.outputs.build-tag}}
+          # note: as of 2023-01-12, vm-builder requires a trailing ":latest" for local images
+          ./vm-builder -use-inittab -src=temp-vm-compute-node:latest -dst=369495373322.dkr.ecr.eu-central-1.amazonaws.com/vm-compute-node-${{ matrix.version }}:${{needs.tag.outputs.build-tag}}
 
       - name: Pushing vm-compute-node image
         run: |
@@ -958,7 +949,7 @@ jobs:
   promote-compatibility-data:
     runs-on: [ self-hosted, gen3, small ]
     container:
-      image: 369495373322.dkr.ecr.eu-central-1.amazonaws.com/base:pinned
+      image: 369495373322.dkr.ecr.eu-central-1.amazonaws.com/rust:pinned
       options: --init
     needs: [ promote-images, tag, regress-tests ]
     if: github.ref_name == 'release' && github.event_name != 'workflow_dispatch'
@@ -969,13 +960,11 @@ jobs:
           PREFIX: artifacts/latest
         run: |
           # Update compatibility snapshot for the release
-          for pg_version in v14 v15; do
-            for build_type in debug release; do
-              OLD_FILENAME=compatibility-snapshot-${build_type}-pg${pg_version}-${GITHUB_RUN_ID}.tar.zst
-              NEW_FILENAME=compatibility-snapshot-${build_type}-pg${pg_version}.tar.zst
+          for build_type in debug release; do
+            OLD_FILENAME=compatibility-snapshot-${build_type}-pg14-${GITHUB_RUN_ID}.tar.zst
+            NEW_FILENAME=compatibility-snapshot-${build_type}-pg14.tar.zst
 
-              time aws s3 mv --only-show-errors s3://${BUCKET}/${PREFIX}/${OLD_FILENAME} s3://${BUCKET}/${PREFIX}/${NEW_FILENAME}
-            done
+            time aws s3 mv --only-show-errors s3://${BUCKET}/${PREFIX}/${OLD_FILENAME} s3://${BUCKET}/${PREFIX}/${NEW_FILENAME}
           done
 
           # Update Neon artifact for the release (reuse already uploaded artifact)

CONTRIBUTING.md

@@ -2,7 +2,7 @@
 
 Howdy! Usual good software engineering practices apply. Write
 tests. Write comments. Follow standard Rust coding practices where
-possible. Use `cargo fmt` and `cargo clippy` to tidy up formatting.
+possible. Use 'cargo fmt' and 'clippy' to tidy up formatting.
 
 There are soft spots in the code, which could use cleanup,
 refactoring, additional comments, and so forth. Let's try to raise the

Cargo.lock

@@ -230,38 +230,40 @@ checksum = "d468802bab17cbc0cc575e9b053f41e72aa36bfa6b7f55e3529ffa43161b97fa"
 
 [[package]]
 name = "aws-config"
-version = "0.55.2"
+version = "0.51.0"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "fc00553f5f3c06ffd4510a9d576f92143618706c45ea6ff81e84ad9be9588abd"
+checksum = "56a636c44c77fa18bdba56126a34d30cfe5538fe88f7d34988fa731fee143ddd"
 dependencies = [
- "aws-credential-types",
  "aws-http",
+ "aws-sdk-sso",
  "aws-sdk-sts",
- "aws-smithy-async",
- "aws-smithy-client",
- "aws-smithy-http",
- "aws-smithy-http-tower",
+ "aws-smithy-async 0.51.0",
+ "aws-smithy-client 0.51.0",
+ "aws-smithy-http 0.51.0",
+ "aws-smithy-http-tower 0.51.0",
  "aws-smithy-json",
- "aws-smithy-types",
- "aws-types",
+ "aws-smithy-types 0.51.0",
+ "aws-types 0.51.0",
  "bytes",
- "fastrand",
+ "hex",
  "http",
  "hyper",
+ "ring",
  "time",
  "tokio",
  "tower",
  "tracing",
+ "zeroize",
 ]
 
 [[package]]
 name = "aws-credential-types"
-version = "0.55.2"
+version = "0.55.1"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "4cb57ac6088805821f78d282c0ba8aec809f11cbee10dda19a97b03ab040ccc2"
+checksum = "f4232d3729eefc287adc0d5a8adc97b7d94eefffe6bbe94312cc86c7ab6b06ce"
 dependencies = [
- "aws-smithy-async",
- "aws-smithy-types",
+ "aws-smithy-async 0.55.1",
+ "aws-smithy-types 0.55.1",
  "fastrand",
  "tokio",
  "tracing",
@@ -270,13 +272,13 @@ dependencies = [
 
 [[package]]
 name = "aws-endpoint"
-version = "0.55.2"
+version = "0.51.0"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "9c5f6f84a4f46f95a9bb71d9300b73cd67eb868bc43ae84f66ad34752299f4ac"
+checksum = "6ca8f374874f6459aaa88dc861d7f5d834ca1ff97668eae190e97266b5f6c3fb"
 dependencies = [
- "aws-smithy-http",
- "aws-smithy-types",
- "aws-types",
+ "aws-smithy-http 0.51.0",
+ "aws-smithy-types 0.51.0",
+ "aws-types 0.51.0",
  "http",
  "regex",
  "tracing",
@@ -284,14 +286,13 @@ dependencies = [
 
 [[package]]
 name = "aws-http"
-version = "0.55.2"
+version = "0.51.0"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "a754683c322f7dc5167484266489fdebdcd04d26e53c162cad1f3f949f2c5671"
+checksum = "78d41e19e779b73463f5f0c21b3aacc995f4ba783ab13a7ae9f5dfb159a551b4"
 dependencies = [
- "aws-credential-types",
- "aws-smithy-http",
- "aws-smithy-types",
- "aws-types",
+ "aws-smithy-http 0.51.0",
+ "aws-smithy-types 0.51.0",
+ "aws-types 0.51.0",
  "bytes",
  "http",
  "http-body",
@@ -303,104 +304,127 @@ dependencies = [
 
 [[package]]
 name = "aws-sdk-s3"
-version = "0.25.1"
+version = "0.21.0"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "392b9811ca489747ac84349790e49deaa1f16631949e7dd4156000251c260eae"
+checksum = "a9f08665c8e03aca8cb092ef01e617436ebfa977fddc1240e1b062488ab5d48a"
 dependencies = [
- "aws-credential-types",
  "aws-endpoint",
  "aws-http",
  "aws-sig-auth",
  "aws-sigv4",
- "aws-smithy-async",
+ "aws-smithy-async 0.51.0",
  "aws-smithy-checksums",
- "aws-smithy-client",
+ "aws-smithy-client 0.51.0",
  "aws-smithy-eventstream",
- "aws-smithy-http",
- "aws-smithy-http-tower",
- "aws-smithy-json",
- "aws-smithy-types",
+ "aws-smithy-http 0.51.0",
+ "aws-smithy-http-tower 0.51.0",
+ "aws-smithy-types 0.51.0",
  "aws-smithy-xml",
- "aws-types",
+ "aws-types 0.51.0",
  "bytes",
+ "bytes-utils",
  "http",
  "http-body",
- "once_cell",
- "percent-encoding",
- "regex",
  "tokio-stream",
  "tower",
  "tracing",
- "url",
+]
+
+[[package]]
+name = "aws-sdk-sso"
+version = "0.21.0"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "86dcb1cb71aa8763b327542ead410424515cff0cde5b753eedd2917e09c63734"
+dependencies = [
+ "aws-endpoint",
+ "aws-http",
+ "aws-sig-auth",
+ "aws-smithy-async 0.51.0",
+ "aws-smithy-client 0.51.0",
+ "aws-smithy-http 0.51.0",
+ "aws-smithy-http-tower 0.51.0",
+ "aws-smithy-json",
+ "aws-smithy-types 0.51.0",
+ "aws-types 0.51.0",
+ "bytes",
+ "http",
+ "tokio-stream",
+ "tower",
 ]
 
 [[package]]
 name = "aws-sdk-sts"
-version = "0.27.0"
+version = "0.21.0"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "2d0fbe3c2c342bc8dfea4bb43937405a8ec06f99140a0dcb9c7b59e54dfa93a1"
+checksum = "fdfcf584297c666f6b472d5368a78de3bc714b6e0a53d7fbf76c3e347c292ab1"
 dependencies = [
- "aws-credential-types",
  "aws-endpoint",
  "aws-http",
  "aws-sig-auth",
- "aws-smithy-async",
- "aws-smithy-client",
- "aws-smithy-http",
- "aws-smithy-http-tower",
- "aws-smithy-json",
+ "aws-smithy-async 0.51.0",
+ "aws-smithy-client 0.51.0",
+ "aws-smithy-http 0.51.0",
+ "aws-smithy-http-tower 0.51.0",
  "aws-smithy-query",
- "aws-smithy-types",
+ "aws-smithy-types 0.51.0",
  "aws-smithy-xml",
- "aws-types",
+ "aws-types 0.51.0",
  "bytes",
  "http",
- "regex",
  "tower",
- "tracing",
 ]
 
 [[package]]
 name = "aws-sig-auth"
-version = "0.55.2"
+version = "0.51.0"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "84dc92a63ede3c2cbe43529cb87ffa58763520c96c6a46ca1ced80417afba845"
+checksum = "12cbe7b2be9e185c1fbce27fc9c41c66b195b32d89aa099f98768d9544221308"
 dependencies = [
- "aws-credential-types",
  "aws-sigv4",
  "aws-smithy-eventstream",
- "aws-smithy-http",
- "aws-types",
+ "aws-smithy-http 0.51.0",
+ "aws-types 0.51.0",
  "http",
  "tracing",
 ]
 
 [[package]]
 name = "aws-sigv4"
-version = "0.55.2"
+version = "0.51.1"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "392fefab9d6fcbd76d518eb3b1c040b84728ab50f58df0c3c53ada4bea9d327e"
+checksum = "5c0b2658d2cb66dbf02f0e8dee80810ef1e0ca3530ede463e0ef994c301087d1"
 dependencies = [
  "aws-smithy-eventstream",
- "aws-smithy-http",
+ "aws-smithy-http 0.51.0",
  "bytes",
  "form_urlencoded",
  "hex",
- "hmac",
  "http",
  "once_cell",
  "percent-encoding",
  "regex",
- "sha2",
+ "ring",
  "time",
  "tracing",
 ]
 
 [[package]]
 name = "aws-smithy-async"
-version = "0.55.2"
+version = "0.51.0"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "ae23b9fe7a07d0919000116c4c5c0578303fbce6fc8d32efca1f7759d4c20faf"
+checksum = "7b3442b4c5d3fc39891a2e5e625735fba6b24694887d49c6518460fde98247a9"
+dependencies = [
+ "futures-util",
+ "pin-project-lite",
+ "tokio",
+ "tokio-stream",
+]
+
+[[package]]
+name = "aws-smithy-async"
+version = "0.55.1"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "88573bcfbe1dcfd54d4912846df028b42d6255cbf9ce07be216b1bbfd11fc4b9"
 dependencies = [
  "futures-util",
  "pin-project-lite",
@@ -410,12 +434,12 @@ dependencies = [
 
 [[package]]
 name = "aws-smithy-checksums"
-version = "0.55.2"
+version = "0.51.0"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "a6367acbd6849b8c7c659e166955531274ae147bf83ab4312885991f6b6706cb"
+checksum = "cc227e36e346f45298288359f37123e1a92628d1cec6b11b5eb335553278bd9e"
 dependencies = [
- "aws-smithy-http",
- "aws-smithy-types",
+ "aws-smithy-http 0.51.0",
+ "aws-smithy-types 0.51.0",
  "bytes",
  "crc32c",
  "crc32fast",
@@ -431,14 +455,14 @@ dependencies = [
 
 [[package]]
 name = "aws-smithy-client"
-version = "0.55.2"
+version = "0.51.0"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "5230d25d244a51339273b8870f0f77874cd4449fb4f8f629b21188ae10cfc0ba"
+checksum = "ff28d553714f8f54cd921227934fc13a536a1c03f106e56b362fd57e16d450ad"
 dependencies = [
- "aws-smithy-async",
- "aws-smithy-http",
- "aws-smithy-http-tower",
- "aws-smithy-types",
+ "aws-smithy-async 0.51.0",
+ "aws-smithy-http 0.51.0",
+ "aws-smithy-http-tower 0.51.0",
+ "aws-smithy-types 0.51.0",
  "bytes",
  "fastrand",
  "http",
@@ -447,7 +471,26 @@ dependencies = [
  "hyper-rustls",
  "lazy_static",
  "pin-project-lite",
- "rustls 0.20.8",
+ "tokio",
+ "tower",
+ "tracing",
+]
+
+[[package]]
+name = "aws-smithy-client"
+version = "0.55.1"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "b2f52352bae50d3337d5d6151b695d31a8c10ebea113eca5bead531f8301b067"
+dependencies = [
+ "aws-smithy-async 0.55.1",
+ "aws-smithy-http 0.55.1",
+ "aws-smithy-http-tower 0.55.1",
+ "aws-smithy-types 0.55.1",
+ "bytes",
+ "fastrand",
+ "http",
+ "http-body",
+ "pin-project-lite",
  "tokio",
  "tower",
  "tracing",
@@ -455,23 +498,23 @@ dependencies = [
 
 [[package]]
 name = "aws-smithy-eventstream"
-version = "0.55.2"
+version = "0.51.0"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "22d2a2bcc16e5c4d949ffd2b851da852b9bbed4bb364ed4ae371b42137ca06d9"
+checksum = "d7ea0df7161ce65b5c8ca6eb709a1a907376fa18226976e41c748ce02ccccf24"
 dependencies = [
- "aws-smithy-types",
+ "aws-smithy-types 0.51.0",
  "bytes",
  "crc32fast",
 ]
 
 [[package]]
 name = "aws-smithy-http"
-version = "0.55.2"
+version = "0.51.0"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "b60e2133beb9fe6ffe0b70deca57aaeff0a35ad24a9c6fab2fd3b4f45b99fdb5"
+checksum = "bf58ed4fefa61dbf038e5421a521cbc2c448ef69deff0ab1d915d8a10eda5664"
 dependencies = [
  "aws-smithy-eventstream",
- "aws-smithy-types",
+ "aws-smithy-types 0.51.0",
  "bytes",
  "bytes-utils",
  "futures-core",
@@ -488,13 +531,48 @@ dependencies = [
 ]
 
 [[package]]
-name = "aws-smithy-http-tower"
-version = "0.55.2"
+name = "aws-smithy-http"
+version = "0.55.1"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "3a4d94f556c86a0dd916a5d7c39747157ea8cb909ca469703e20fee33e448b67"
+checksum = "03bcc02d7ed9649d855c8ce4a735e9848d7b8f7568aad0504c158e3baa955df8"
 dependencies = [
- "aws-smithy-http",
- "aws-smithy-types",
+ "aws-smithy-types 0.55.1",
+ "bytes",
+ "bytes-utils",
+ "futures-core",
+ "http",
+ "http-body",
+ "hyper",
+ "once_cell",
+ "percent-encoding",
+ "pin-project-lite",
+ "pin-utils",
+ "tracing",
+]
+
+[[package]]
+name = "aws-smithy-http-tower"
+version = "0.51.0"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "20c96d7bd35e7cf96aca1134b2f81b1b59ffe493f7c6539c051791cbbf7a42d3"
+dependencies = [
+ "aws-smithy-http 0.51.0",
+ "bytes",
+ "http",
+ "http-body",
+ "pin-project-lite",
+ "tower",
+ "tracing",
+]
+
+[[package]]
+name = "aws-smithy-http-tower"
+version = "0.55.1"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "da88b3a860f65505996c29192d800f1aeb9480440f56d63aad33a3c12045017a"
+dependencies = [
+ "aws-smithy-http 0.55.1",
+ "aws-smithy-types 0.55.1",
  "bytes",
  "http",
  "http-body",
@@ -505,28 +583,40 @@ dependencies = [
 
 [[package]]
 name = "aws-smithy-json"
-version = "0.55.2"
+version = "0.51.0"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "5ce3d6e6ebb00b2cce379f079ad5ec508f9bcc3a9510d9b9c1840ed1d6f8af39"
+checksum = "d8324ba98c8a94187723cc16c37aefa09504646ee65c3d2c3af495bab5ea701b"
 dependencies = [
- "aws-smithy-types",
+ "aws-smithy-types 0.51.0",
 ]
 
 [[package]]
 name = "aws-smithy-query"
-version = "0.55.2"
+version = "0.51.0"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "d58edfca32ef9bfbc1ca394599e17ea329cb52d6a07359827be74235b64b3298"
+checksum = "83834ed2ff69ea6f6657baf205267dc2c0abe940703503a3e5d60ce23be3d306"
 dependencies = [
- "aws-smithy-types",
+ "aws-smithy-types 0.51.0",
  "urlencoding",
 ]
 
 [[package]]
 name = "aws-smithy-types"
-version = "0.55.2"
+version = "0.51.0"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "58db46fc1f4f26be01ebdb821751b4e2482cd43aa2b64a0348fb89762defaffa"
+checksum = "8b02e06ea63498c43bc0217ea4d16605d4e58d85c12fc23f6572ff6d0a840c61"
+dependencies = [
+ "itoa",
+ "num-integer",
+ "ryu",
+ "time",
+]
+
+[[package]]
+name = "aws-smithy-types"
+version = "0.55.1"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "cd0afc731fd1417d791f9145a1e0c30e23ae0beaab9b4814017708ead2fc20f1"
 dependencies = [
  "base64-simd",
  "itoa",
@@ -537,24 +627,40 @@ dependencies = [
 
 [[package]]
 name = "aws-smithy-xml"
-version = "0.55.2"
+version = "0.51.0"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "fb557fe4995bd9ec87fb244bbb254666a971dc902a783e9da8b7711610e9664c"
+checksum = "246e9f83dd1fdf5d347fa30ae4ad30a9d1d42ce4cd74a93d94afa874646f94cd"
 dependencies = [
  "xmlparser",
 ]
 
 [[package]]
 name = "aws-types"
-version = "0.55.2"
+version = "0.51.0"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "de0869598bfe46ec44ffe17e063ed33336e59df90356ca8ff0e8da6f7c1d994b"
+checksum = "05701d32da168b44f7ee63147781aed8723e792cc131cb9b18363b5393f17f70"
+dependencies = [
+ "aws-smithy-async 0.51.0",
+ "aws-smithy-client 0.51.0",
+ "aws-smithy-http 0.51.0",
+ "aws-smithy-types 0.51.0",
+ "http",
+ "rustc_version",
+ "tracing",
+ "zeroize",
+]
+
+[[package]]
+name = "aws-types"
+version = "0.55.1"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "b9b082e329d9a304d39e193ad5c7ab363a0d6507aca6965e0673a746686fb0cc"
 dependencies = [
  "aws-credential-types",
- "aws-smithy-async",
- "aws-smithy-client",
- "aws-smithy-http",
- "aws-smithy-types",
+ "aws-smithy-async 0.55.1",
+ "aws-smithy-client 0.55.1",
+ "aws-smithy-http 0.55.1",
+ "aws-smithy-types 0.55.1",
  "http",
  "rustc_version",
  "tracing",
@@ -905,20 +1011,6 @@ version = "1.0.0"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "acbf1af155f9b9ef647e42cdc158db4b64a1b61f743629225fde6f3e0be2a7c7"
 
-[[package]]
-name = "combine"
-version = "4.6.6"
-source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "35ed6e9d84f0b51a7f52daf1c7d71dd136fd7a3f41a8462b8cdb8c78d920fad4"
-dependencies = [
- "bytes",
- "futures-core",
- "memchr",
- "pin-project-lite",
- "tokio",
- "tokio-util",
-]
-
 [[package]]
 name = "comfy-table"
 version = "6.1.4"
@@ -2834,7 +2926,7 @@ dependencies = [
 [[package]]
 name = "postgres"
 version = "0.19.4"
-source = "git+https://github.com/neondatabase/rust-postgres.git?rev=2e9b5f1ddc481d1a98fa79f6b9378ac4f170b7c9#2e9b5f1ddc481d1a98fa79f6b9378ac4f170b7c9"
+source = "git+https://github.com/neondatabase/rust-postgres.git?rev=0bc41d8503c092b040142214aac3cf7d11d0c19f#0bc41d8503c092b040142214aac3cf7d11d0c19f"
 dependencies = [
  "bytes",
  "fallible-iterator",
@@ -2847,7 +2939,7 @@ dependencies = [
 [[package]]
 name = "postgres-native-tls"
 version = "0.5.0"
-source = "git+https://github.com/neondatabase/rust-postgres.git?rev=2e9b5f1ddc481d1a98fa79f6b9378ac4f170b7c9#2e9b5f1ddc481d1a98fa79f6b9378ac4f170b7c9"
+source = "git+https://github.com/neondatabase/rust-postgres.git?rev=0bc41d8503c092b040142214aac3cf7d11d0c19f#0bc41d8503c092b040142214aac3cf7d11d0c19f"
 dependencies = [
  "native-tls",
  "tokio",
@@ -2858,7 +2950,7 @@ dependencies = [
 [[package]]
 name = "postgres-protocol"
 version = "0.6.4"
-source = "git+https://github.com/neondatabase/rust-postgres.git?rev=2e9b5f1ddc481d1a98fa79f6b9378ac4f170b7c9#2e9b5f1ddc481d1a98fa79f6b9378ac4f170b7c9"
+source = "git+https://github.com/neondatabase/rust-postgres.git?rev=0bc41d8503c092b040142214aac3cf7d11d0c19f#0bc41d8503c092b040142214aac3cf7d11d0c19f"
 dependencies = [
  "base64 0.20.0",
  "byteorder",
@@ -2876,7 +2968,7 @@ dependencies = [
 [[package]]
 name = "postgres-types"
 version = "0.2.4"
-source = "git+https://github.com/neondatabase/rust-postgres.git?rev=2e9b5f1ddc481d1a98fa79f6b9378ac4f170b7c9#2e9b5f1ddc481d1a98fa79f6b9378ac4f170b7c9"
+source = "git+https://github.com/neondatabase/rust-postgres.git?rev=0bc41d8503c092b040142214aac3cf7d11d0c19f#0bc41d8503c092b040142214aac3cf7d11d0c19f"
 dependencies = [
  "bytes",
  "fallible-iterator",
@@ -3118,8 +3210,6 @@ dependencies = [
  "prometheus",
  "rand",
  "rcgen",
- "redis",
- "ref-cast",
  "regex",
  "reqwest",
  "reqwest-middleware",
@@ -3226,29 +3316,6 @@ dependencies = [
  "yasna",
 ]
 
-[[package]]
-name = "redis"
-version = "0.23.0"
-source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "3ea8c51b5dc1d8e5fd3350ec8167f464ec0995e79f2e90a075b63371500d557f"
-dependencies = [
- "async-trait",
- "bytes",
- "combine",
- "futures-util",
- "itoa",
- "percent-encoding",
- "pin-project-lite",
- "rustls 0.21.0",
- "rustls-native-certs",
- "ryu",
- "sha1_smol",
- "tokio",
- "tokio-rustls 0.24.0",
- "tokio-util",
- "url",
-]
-
 [[package]]
 name = "redox_syscall"
 version = "0.2.16"
@@ -3267,26 +3334,6 @@ dependencies = [
  "bitflags",
 ]
 
-[[package]]
-name = "ref-cast"
-version = "1.0.16"
-source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "f43faa91b1c8b36841ee70e97188a869d37ae21759da6846d4be66de5bf7b12c"
-dependencies = [
- "ref-cast-impl",
-]
-
-[[package]]
-name = "ref-cast-impl"
-version = "1.0.16"
-source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "8d2275aab483050ab2a7364c1a46604865ee7d6906684e08db0f090acf74f9e7"
-dependencies = [
- "proc-macro2",
- "quote",
- "syn 2.0.15",
-]
-
 [[package]]
 name = "regex"
 version = "1.7.3"
@@ -3320,10 +3367,9 @@ dependencies = [
  "anyhow",
  "async-trait",
  "aws-config",
- "aws-credential-types",
  "aws-sdk-s3",
- "aws-smithy-http",
- "aws-types",
+ "aws-smithy-http 0.51.0",
+ "aws-types 0.55.1",
  "hyper",
  "metrics",
  "once_cell",
@@ -3903,12 +3949,6 @@ dependencies = [
  "digest",
 ]
 
-[[package]]
-name = "sha1_smol"
-version = "1.0.0"
-source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "ae1a47186c03a32177042e55dbc5fd5aee900b8e0069a8d70fba96a9375cd012"
-
 [[package]]
 name = "sha2"
 version = "0.10.6"
@@ -4336,9 +4376,9 @@ dependencies = [
 
 [[package]]
 name = "tokio"
-version = "1.28.1"
+version = "1.27.0"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "0aa32867d44e6f2ce3385e89dceb990188b8bb0fb25b0cf576647a6f98ac5105"
+checksum = "d0de47a4eecbe11f498978a9b29d792f0d2692d1dd003650c24c76510e3bc001"
 dependencies = [
  "autocfg",
  "bytes",
@@ -4349,7 +4389,7 @@ dependencies = [
  "signal-hook-registry",
  "socket2 0.4.9",
  "tokio-macros",
- "windows-sys 0.48.0",
+ "windows-sys 0.45.0",
 ]
 
 [[package]]
@@ -4364,9 +4404,9 @@ dependencies = [
 
 [[package]]
 name = "tokio-macros"
-version = "2.1.0"
+version = "2.0.0"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "630bdcf245f78637c13ec01ffae6187cca34625e8c63150d424b59e55af2675e"
+checksum = "61a573bdc87985e9d6ddeed1b3d864e8a302c847e40d647746df2f1de209d1ce"
 dependencies = [
  "proc-macro2",
  "quote",
@@ -4386,7 +4426,7 @@ dependencies = [
 [[package]]
 name = "tokio-postgres"
 version = "0.7.7"
-source = "git+https://github.com/neondatabase/rust-postgres.git?rev=2e9b5f1ddc481d1a98fa79f6b9378ac4f170b7c9#2e9b5f1ddc481d1a98fa79f6b9378ac4f170b7c9"
+source = "git+https://github.com/neondatabase/rust-postgres.git?rev=0bc41d8503c092b040142214aac3cf7d11d0c19f#0bc41d8503c092b040142214aac3cf7d11d0c19f"
 dependencies = [
  "async-trait",
  "byteorder",
@@ -5391,7 +5431,6 @@ dependencies = [
  "reqwest",
  "ring",
  "rustls 0.20.8",
- "rustls 0.21.0",
  "scopeguard",
  "serde",
  "serde_json",

Cargo.toml

@@ -21,10 +21,9 @@ anyhow = { version = "1.0", features = ["backtrace"] }
 async-stream = "0.3"
 async-trait = "0.1"
 atty = "0.2.14"
-aws-config = { version = "0.55", default-features = false, features=["rustls"] }
-aws-sdk-s3 = "0.25"
-aws-smithy-http = "0.55"
-aws-credential-types = "0.55"
+aws-config = { version = "0.51.0", default-features = false, features=["rustls"] }
+aws-sdk-s3 = "0.21.0"
+aws-smithy-http = "0.51.0"
 aws-types = "0.55"
 base64 = "0.13.0"
 bincode = "1.3"
@@ -77,12 +76,10 @@ pin-project-lite = "0.2"
 prometheus = {version = "0.13", default_features=false, features = ["process"]} # removes protobuf dependency
 prost = "0.11"
 rand = "0.8"
-ref-cast = "1.0"
-redis = { version = "0.23.0", features = ["tokio-rustls-comp"] }
 regex = "1.4"
 reqwest = { version = "0.11", default-features = false, features = ["rustls-tls"] }
-reqwest-middleware = "0.2.0"
 reqwest-tracing = { version = "0.4.0", features = ["opentelemetry_0_18"] }
+reqwest-middleware = "0.2.0"
 routerify = "3"
 rpds = "0.13"
 rustls = "0.20"
@@ -128,11 +125,11 @@ env_logger = "0.10"
 log = "0.4"
 
 ## Libraries from neondatabase/ git forks, ideally with changes to be upstreamed
-postgres = { git = "https://github.com/neondatabase/rust-postgres.git", rev="2e9b5f1ddc481d1a98fa79f6b9378ac4f170b7c9" }
-postgres-native-tls = { git = "https://github.com/neondatabase/rust-postgres.git", rev="2e9b5f1ddc481d1a98fa79f6b9378ac4f170b7c9" }
-postgres-protocol = { git = "https://github.com/neondatabase/rust-postgres.git", rev="2e9b5f1ddc481d1a98fa79f6b9378ac4f170b7c9" }
-postgres-types = { git = "https://github.com/neondatabase/rust-postgres.git", rev="2e9b5f1ddc481d1a98fa79f6b9378ac4f170b7c9" }
-tokio-postgres = { git = "https://github.com/neondatabase/rust-postgres.git", rev="2e9b5f1ddc481d1a98fa79f6b9378ac4f170b7c9" }
+postgres = { git = "https://github.com/neondatabase/rust-postgres.git", rev="0bc41d8503c092b040142214aac3cf7d11d0c19f" }
+postgres-native-tls = { git = "https://github.com/neondatabase/rust-postgres.git", rev="0bc41d8503c092b040142214aac3cf7d11d0c19f" }
+postgres-protocol = { git = "https://github.com/neondatabase/rust-postgres.git", rev="0bc41d8503c092b040142214aac3cf7d11d0c19f" }
+postgres-types = { git = "https://github.com/neondatabase/rust-postgres.git", rev="0bc41d8503c092b040142214aac3cf7d11d0c19f" }
+tokio-postgres = { git = "https://github.com/neondatabase/rust-postgres.git", rev="0bc41d8503c092b040142214aac3cf7d11d0c19f" }
 tokio-tar = { git = "https://github.com/neondatabase/tokio-tar.git", rev="404df61437de0feef49ba2ccdbdd94eb8ad6e142" }
 
 ## Other git libraries
@@ -168,7 +165,7 @@ tonic-build = "0.9"
 
 # This is only needed for proxy's tests.
 # TODO: we should probably fork `tokio-postgres-rustls` instead.
-tokio-postgres = { git = "https://github.com/neondatabase/rust-postgres.git", rev="2e9b5f1ddc481d1a98fa79f6b9378ac4f170b7c9" }
+tokio-postgres = { git = "https://github.com/neondatabase/rust-postgres.git", rev="0bc41d8503c092b040142214aac3cf7d11d0c19f" }
 
 # Changes the MAX_THREADS limit from 4096 to 32768.
 # This is a temporary workaround for using tracing from many threads in safekeepers code,

Dockerfile.compute-node

@@ -415,23 +415,6 @@ RUN apt-get update && \
     make -j $(getconf _NPROCESSORS_ONLN) install && \
     echo 'trusted = true' >> /usr/local/pgsql/share/extension/kq_imcx.control
 
-#########################################################################################
-#
-# Layer "pg-cron-pg-build"
-# compile pg_cron extension
-#
-#########################################################################################
-FROM build-deps AS pg-cron-pg-build
-COPY --from=pg-build /usr/local/pgsql/ /usr/local/pgsql/
-
-ENV PATH "/usr/local/pgsql/bin/:$PATH"
-RUN wget https://github.com/citusdata/pg_cron/archive/refs/tags/v1.5.2.tar.gz -O pg_cron.tar.gz && \
-    echo "6f7f0980c03f1e2a6a747060e67bf4a303ca2a50e941e2c19daeed2b44dec744 pg_cron.tar.gz" | sha256sum --check && \
-    mkdir pg_cron-src && cd pg_cron-src && tar xvzf ../pg_cron.tar.gz --strip-components=1 -C . && \
-    make -j $(getconf _NPROCESSORS_ONLN) && \
-    make -j $(getconf _NPROCESSORS_ONLN) install && \
-    echo 'trusted = true' >> /usr/local/pgsql/share/extension/pg_cron.control
-
 #########################################################################################
 #
 # Layer "rust extensions"
@@ -546,7 +529,6 @@ COPY --from=plpgsql-check-pg-build /usr/local/pgsql/ /usr/local/pgsql/
 COPY --from=timescaledb-pg-build /usr/local/pgsql/ /usr/local/pgsql/
 COPY --from=pg-hint-plan-pg-build /usr/local/pgsql/ /usr/local/pgsql/
 COPY --from=kq-imcx-pg-build /usr/local/pgsql/ /usr/local/pgsql/
-COPY --from=pg-cron-pg-build /usr/local/pgsql/ /usr/local/pgsql/
 COPY pgxn/ pgxn/
 
 RUN make -j $(getconf _NPROCESSORS_ONLN) \

Dockerfile.vm-compute-node

@@ -0,0 +1,70 @@
+# Note: this file *mostly* just builds on Dockerfile.compute-node
+
+ARG SRC_IMAGE
+ARG VM_INFORMANT_VERSION=v0.1.14
+# on libcgroup update, make sure to check bootstrap.sh for changes
+ARG LIBCGROUP_VERSION=v2.0.3
+
+# Pull VM informant, to copy from later
+FROM neondatabase/vm-informant:$VM_INFORMANT_VERSION as informant
+
+# Build cgroup-tools
+#
+# At time of writing (2023-03-14), debian bullseye has a version of cgroup-tools (technically
+# libcgroup) that doesn't support cgroup v2 (version 0.41-11). Unfortunately, the vm-informant
+# requires cgroup v2, so we'll build cgroup-tools ourselves.
+FROM debian:bullseye-slim as libcgroup-builder
+ARG LIBCGROUP_VERSION
+
+RUN set -exu \
+	&& apt update \
+	&& apt install --no-install-recommends -y \
+		git \
+		ca-certificates \
+		automake \
+		cmake \
+		make \
+		gcc \
+		byacc \
+		flex \
+		libtool \
+		libpam0g-dev \
+	&& git clone --depth 1 -b $LIBCGROUP_VERSION https://github.com/libcgroup/libcgroup \
+	&& INSTALL_DIR="/libcgroup-install" \
+	&& mkdir -p "$INSTALL_DIR/bin" "$INSTALL_DIR/include" \
+	&& cd libcgroup \
+	# extracted from bootstrap.sh, with modified flags:
+	&& (test -d m4 || mkdir m4) \
+	&& autoreconf -fi \
+	&& rm -rf autom4te.cache \
+	&& CFLAGS="-O3" ./configure --prefix="$INSTALL_DIR" --sysconfdir=/etc --localstatedir=/var --enable-opaque-hierarchy="name=systemd" \
+	# actually build the thing...
+	&& make install
+
+# Combine, starting from non-VM compute node image.
+FROM $SRC_IMAGE as base
+
+# Temporarily set user back to root so we can run adduser, set inittab
+USER root
+RUN adduser vm-informant --disabled-password --no-create-home
+
+RUN set -e \
+	&& rm -f /etc/inittab \
+	&& touch /etc/inittab
+
+RUN set -e \
+	&& echo "::sysinit:cgconfigparser -l /etc/cgconfig.conf -s 1664" >> /etc/inittab \
+	&& CONNSTR="dbname=postgres user=cloud_admin sslmode=disable" \
+	&& ARGS="--auto-restart --cgroup=neon-postgres --pgconnstr=\"$CONNSTR\"" \
+	&& echo "::respawn:su vm-informant -c '/usr/local/bin/vm-informant $ARGS'" >> /etc/inittab
+
+USER postgres
+
+ADD vm-cgconfig.conf /etc/cgconfig.conf
+COPY --from=informant /usr/bin/vm-informant /usr/local/bin/vm-informant
+
+COPY --from=libcgroup-builder /libcgroup-install/bin/* /usr/bin/
+COPY --from=libcgroup-builder /libcgroup-install/lib/* /usr/lib/
+COPY --from=libcgroup-builder /libcgroup-install/sbin/* /usr/sbin/
+
+ENTRYPOINT ["/usr/sbin/cgexec", "-g", "*:neon-postgres", "/usr/local/bin/compute_ctl"]

README.md

@@ -17,7 +17,7 @@ The Neon storage engine consists of two major components:
 - Pageserver. Scalable storage backend for the compute nodes.
 - Safekeepers. The safekeepers form a redundant WAL service that received WAL from the compute node, and stores it durably until it has been processed by the pageserver and uploaded to cloud storage.
 
-See developer documentation in [SUMMARY.md](/docs/SUMMARY.md) for more information.
+See developer documentation in [/docs/SUMMARY.md](/docs/SUMMARY.md) for more information.
 
 ## Running local installation
 
@@ -130,11 +130,11 @@ Python (3.9 or higher), and install python3 packages using `./scripts/pysync` (r
 ```sh
 # Create repository in .neon with proper paths to binaries and data
 # Later that would be responsibility of a package install script
-> cargo neon init
+> ./target/debug/neon_local init
 Starting pageserver at '127.0.0.1:64000' in '.neon'.
 
 # start pageserver, safekeeper, and broker for their intercommunication
-> cargo neon start
+> ./target/debug/neon_local start
 Starting neon broker at 127.0.0.1:50051
 storage_broker started, pid: 2918372
 Starting pageserver at '127.0.0.1:64000' in '.neon'.
@@ -143,19 +143,19 @@ Starting safekeeper at '127.0.0.1:5454' in '.neon/safekeepers/sk1'.
 safekeeper 1 started, pid: 2918437
 
 # create initial tenant and use it as a default for every future neon_local invocation
-> cargo neon tenant create --set-default
+> ./target/debug/neon_local tenant create --set-default
 tenant 9ef87a5bf0d92544f6fafeeb3239695c successfully created on the pageserver
 Created an initial timeline 'de200bd42b49cc1814412c7e592dd6e9' at Lsn 0/16B5A50 for tenant: 9ef87a5bf0d92544f6fafeeb3239695c
 Setting tenant 9ef87a5bf0d92544f6fafeeb3239695c as a default one
 
 # start postgres compute node
-> cargo neon endpoint start main
+> ./target/debug/neon_local endpoint start main
 Starting new endpoint main (PostgreSQL v14) on timeline de200bd42b49cc1814412c7e592dd6e9 ...
 Extracting base backup to create postgres instance: path=.neon/pgdatadirs/tenants/9ef87a5bf0d92544f6fafeeb3239695c/main port=55432
 Starting postgres at 'host=127.0.0.1 port=55432 user=cloud_admin dbname=postgres'
 
 # check list of running postgres instances
-> cargo neon endpoint list
+> ./target/debug/neon_local endpoint list
  ENDPOINT  ADDRESS          TIMELINE                          BRANCH NAME  LSN        STATUS
  main      127.0.0.1:55432  de200bd42b49cc1814412c7e592dd6e9  main         0/16B5BA8  running
 ```
@@ -177,22 +177,22 @@ postgres=# select * from t;
 3. And create branches and run postgres on them:
 ```sh
 # create branch named migration_check
-> cargo neon timeline branch --branch-name migration_check
+> ./target/debug/neon_local timeline branch --branch-name migration_check
 Created timeline 'b3b863fa45fa9e57e615f9f2d944e601' at Lsn 0/16F9A00 for tenant: 9ef87a5bf0d92544f6fafeeb3239695c. Ancestor timeline: 'main'
 
 # check branches tree
-> cargo neon timeline list
+> ./target/debug/neon_local timeline list
 (L) main [de200bd42b49cc1814412c7e592dd6e9]
 (L) ┗━ @0/16F9A00: migration_check [b3b863fa45fa9e57e615f9f2d944e601]
 
 # start postgres on that branch
-> cargo neon endpoint start migration_check --branch-name migration_check
+> ./target/debug/neon_local endpoint start migration_check --branch-name migration_check
 Starting new endpoint migration_check (PostgreSQL v14) on timeline b3b863fa45fa9e57e615f9f2d944e601 ...
 Extracting base backup to create postgres instance: path=.neon/pgdatadirs/tenants/9ef87a5bf0d92544f6fafeeb3239695c/migration_check port=55433
 Starting postgres at 'host=127.0.0.1 port=55433 user=cloud_admin dbname=postgres'
 
 # check the new list of running postgres instances
-> cargo neon endpoint list
+> ./target/debug/neon_local endpoint list
  ENDPOINT         ADDRESS          TIMELINE                          BRANCH NAME      LSN        STATUS
  main             127.0.0.1:55432  de200bd42b49cc1814412c7e592dd6e9  main             0/16F9A38  running
  migration_check  127.0.0.1:55433  b3b863fa45fa9e57e615f9f2d944e601  migration_check  0/16F9A70  running
@@ -221,7 +221,7 @@ postgres=# select * from t;
 4. If you want to run tests afterward (see below), you must stop all the running of the pageserver, safekeeper, and postgres instances
    you have just started. You can terminate them all with one command:
 ```sh
-> cargo neon stop
+> ./target/debug/neon_local stop
 ```
 
 ## Running tests
@@ -238,9 +238,9 @@ CARGO_BUILD_FLAGS="--features=testing" make
 
 ## Documentation
 
-[docs](/docs) Contains a top-level overview of all available markdown documentation.
+[/docs/](/docs/) Contains a top-level overview of all available markdown documentation.
 
-- [sourcetree.md](/docs/sourcetree.md) contains overview of source tree layout.
+- [/docs/sourcetree.md](/docs/sourcetree.md) contains overview of source tree layout.
 
 To view your `rustdoc` documentation in a browser, try running `cargo doc --no-deps --open`
 
@@ -265,6 +265,6 @@ To get more familiar with this aspect, refer to:
 
 ## Join the development
 
-- Read [CONTRIBUTING.md](/CONTRIBUTING.md) to learn about project code style and practices.
-- To get familiar with a source tree layout, use [sourcetree.md](/docs/sourcetree.md).
+- Read `CONTRIBUTING.md` to learn about project code style and practices.
+- To get familiar with a source tree layout, use [/docs/sourcetree.md](/docs/sourcetree.md).
 - To learn more about PostgreSQL internals, check http://www.interdb.jp/pg/index.html

control_plane/src/endpoint.rs

@@ -134,7 +134,6 @@ pub struct Endpoint {
 
     // port and address of the Postgres server
     pub address: SocketAddr,
-    // postgres major version in the format: 14, 15, etc.
     pg_version: u32,
 
     // These are not part of the endpoint as such, but the environment
@@ -382,11 +381,6 @@ impl Endpoint {
                 conf.append("primary_conninfo", connstr.as_str());
                 conf.append("primary_slot_name", slot_name.as_str());
                 conf.append("hot_standby", "on");
-                // prefetching of blocks referenced in WAL doesn't make sense for us
-                // Neon hot standby ignores pages that are not in the shared_buffers
-                if self.pg_version >= 15 {
-                    conf.append("recovery_prefetch", "off");
-                }
             }
         }
 

control_plane/src/pageserver.rs

@@ -8,7 +8,9 @@ use std::process::{Child, Command};
 use std::{io, result};
 
 use anyhow::{bail, Context};
-use pageserver_api::models::{self, TenantInfo, TimelineInfo};
+use pageserver_api::models::{
+    TenantConfigRequest, TenantCreateRequest, TenantInfo, TimelineCreateRequest, TimelineInfo,
+};
 use postgres_backend::AuthType;
 use postgres_connection::{parse_host_port, PgConnectionConfig};
 use reqwest::blocking::{Client, RequestBuilder, Response};
@@ -314,8 +316,8 @@ impl PageServerNode {
         settings: HashMap<&str, &str>,
     ) -> anyhow::Result<TenantId> {
         let mut settings = settings.clone();
-
-        let config = models::TenantConfig {
+        let request = TenantCreateRequest {
+            new_tenant_id,
             checkpoint_distance: settings
                 .remove("checkpoint_distance")
                 .map(|x| x.parse::<u64>())
@@ -370,10 +372,6 @@ impl PageServerNode {
                 .remove("evictions_low_residence_duration_metric_threshold")
                 .map(|x| x.to_string()),
         };
-        let request = models::TenantCreateRequest {
-            new_tenant_id,
-            config,
-        };
         if !settings.is_empty() {
             bail!("Unrecognized tenant settings: {settings:?}")
         }
@@ -393,81 +391,67 @@ impl PageServerNode {
             })
     }
 
-    pub fn tenant_config(
-        &self,
-        tenant_id: TenantId,
-        mut settings: HashMap<&str, &str>,
-    ) -> anyhow::Result<()> {
-        let config = {
-            // Braces to make the diff easier to read
-            models::TenantConfig {
+    pub fn tenant_config(&self, tenant_id: TenantId, settings: HashMap<&str, &str>) -> Result<()> {
+        self.http_request(Method::PUT, format!("{}/tenant/config", self.http_base_url))?
+            .json(&TenantConfigRequest {
+                tenant_id,
                 checkpoint_distance: settings
-                    .remove("checkpoint_distance")
+                    .get("checkpoint_distance")
                     .map(|x| x.parse::<u64>())
                     .transpose()
                     .context("Failed to parse 'checkpoint_distance' as an integer")?,
-                checkpoint_timeout: settings.remove("checkpoint_timeout").map(|x| x.to_string()),
+                checkpoint_timeout: settings.get("checkpoint_timeout").map(|x| x.to_string()),
                 compaction_target_size: settings
-                    .remove("compaction_target_size")
+                    .get("compaction_target_size")
                     .map(|x| x.parse::<u64>())
                     .transpose()
                     .context("Failed to parse 'compaction_target_size' as an integer")?,
-                compaction_period: settings.remove("compaction_period").map(|x| x.to_string()),
+                compaction_period: settings.get("compaction_period").map(|x| x.to_string()),
                 compaction_threshold: settings
-                    .remove("compaction_threshold")
+                    .get("compaction_threshold")
                     .map(|x| x.parse::<usize>())
                     .transpose()
                     .context("Failed to parse 'compaction_threshold' as an integer")?,
                 gc_horizon: settings
-                    .remove("gc_horizon")
+                    .get("gc_horizon")
                     .map(|x| x.parse::<u64>())
                     .transpose()
                     .context("Failed to parse 'gc_horizon' as an integer")?,
-                gc_period: settings.remove("gc_period").map(|x| x.to_string()),
+                gc_period: settings.get("gc_period").map(|x| x.to_string()),
                 image_creation_threshold: settings
-                    .remove("image_creation_threshold")
+                    .get("image_creation_threshold")
                     .map(|x| x.parse::<usize>())
                     .transpose()
                     .context("Failed to parse 'image_creation_threshold' as non zero integer")?,
-                pitr_interval: settings.remove("pitr_interval").map(|x| x.to_string()),
+                pitr_interval: settings.get("pitr_interval").map(|x| x.to_string()),
                 walreceiver_connect_timeout: settings
-                    .remove("walreceiver_connect_timeout")
-                    .map(|x| x.to_string()),
-                lagging_wal_timeout: settings
-                    .remove("lagging_wal_timeout")
+                    .get("walreceiver_connect_timeout")
                     .map(|x| x.to_string()),
+                lagging_wal_timeout: settings.get("lagging_wal_timeout").map(|x| x.to_string()),
                 max_lsn_wal_lag: settings
-                    .remove("max_lsn_wal_lag")
+                    .get("max_lsn_wal_lag")
                     .map(|x| x.parse::<NonZeroU64>())
                     .transpose()
                     .context("Failed to parse 'max_lsn_wal_lag' as non zero integer")?,
                 trace_read_requests: settings
-                    .remove("trace_read_requests")
+                    .get("trace_read_requests")
                     .map(|x| x.parse::<bool>())
                     .transpose()
                     .context("Failed to parse 'trace_read_requests' as bool")?,
                 eviction_policy: settings
-                    .remove("eviction_policy")
-                    .map(serde_json::from_str)
+                    .get("eviction_policy")
+                    .map(|x| serde_json::from_str(x))
                     .transpose()
                     .context("Failed to parse 'eviction_policy' json")?,
                 min_resident_size_override: settings
-                    .remove("min_resident_size_override")
+                    .get("min_resident_size_override")
                     .map(|x| x.parse::<u64>())
                     .transpose()
                     .context("Failed to parse 'min_resident_size_override' as an integer")?,
                 evictions_low_residence_duration_metric_threshold: settings
-                    .remove("evictions_low_residence_duration_metric_threshold")
+                    .get("evictions_low_residence_duration_metric_threshold")
                     .map(|x| x.to_string()),
-            }
-        };
-
-        if !settings.is_empty() {
-            bail!("Unrecognized tenant settings: {settings:?}")
-        }
-
-        self.http_request(Method::PUT, format!("{}/tenant/config", self.http_base_url))?
-            .json(&models::TenantConfigRequest { tenant_id, config })
+            })
             .send()?
             .error_from_body()?;
 
@@ -499,7 +483,7 @@ impl PageServerNode {
             Method::POST,
             format!("{}/tenant/{}/timeline", self.http_base_url, tenant_id),
         )?
-        .json(&models::TimelineCreateRequest {
+        .json(&TimelineCreateRequest {
             new_timeline_id,
             ancestor_start_lsn,
             ancestor_timeline_id,

docs/pageserver-thread-mgmt.md

@@ -4,11 +4,6 @@ The pageserver uses Tokio for handling concurrency. Everything runs in
 Tokio tasks, although some parts are written in blocking style and use
 spawn_blocking().
 
-We currently use std blocking functions for disk I/O, however.  The
-current model is that we consider disk I/Os to be short enough that we
-perform them while running in a Tokio task. Changing all the disk I/O
-calls to async is a TODO.
-
 Each Tokio task is tracked by the `task_mgr` module. It maintains a
 registry of tasks, and which tenant or timeline they are operating
 on.
@@ -26,86 +21,19 @@ also a `shudown_watcher()` Future that can be used with `tokio::select!`
 or similar, to wake up on shutdown.
 
 
-### Async cancellation safety
+### Sync vs async
 
-In async Rust, futures can be "cancelled" at any await point, by
-dropping the Future. For example, `tokio::select!` returns as soon as
-one of the Futures returns, and drops the others. `tokio::timeout!` is
-another example. In the Rust ecosystem, some functions are
-cancellation-safe, meaning they can be safely dropped without
-side-effects, while others are not. See documentation of
-`tokio::select!` for examples.
+We use async to wait for incoming data on network connections, and to
+perform other long-running operations. For example, each WAL receiver
+connection is handled by a tokio Task. Once a piece of WAL has been
+received from the network, the task calls the blocking functions in
+the Repository to process the WAL.
 
-In the pageserver and safekeeper, async code is *not*
-cancellation-safe by default. Unless otherwise marked, any async
-function that you call cannot be assumed to be async
-cancellation-safe, and must be polled to completion.
+The core storage code in `layered_repository/` is synchronous, with
+blocking locks and I/O calls. The current model is that we consider
+disk I/Os to be short enough that we perform them while running in a
+Tokio task. If that becomes a problem, we should use `spawn_blocking`
+before entering the synchronous parts of the code, or switch to using
+tokio I/O functions.
 
-The downside of non-cancellation safe code is that you have to be very
-careful when using `tokio::select!`, `tokio::timeout!`, and other such
-functions that can cause a Future to be dropped. They can only be used
-with functions that are explicitly documented to be cancellation-safe,
-or you need to spawn a separate task to shield from the cancellation.
-
-At the entry points to the code, we also take care to poll futures to
-completion, or shield the rest of the code from surprise cancellations
-by spawning a separate task. The code that handles incoming HTTP
-requests, for example, spawns a separate task for each request,
-because Hyper will drop the request-handling Future if the HTTP
-connection is lost.  (FIXME: our HTTP handlers do not do that
-currently, but we should fix that. See [issue
-3478](https://github.com/neondatabase/neon/issues/3478)).
-
-
-#### How to cancel, then?
-
-If our code is not cancellation-safe, how do you cancel long-running
-tasks? Use CancellationTokens.
-
-TODO: More details on that. And we have an ongoing discussion on what
-to do if cancellations might come from multiple sources.
-
-#### Exceptions
-Some library functions are cancellation-safe, and are explicitly marked
-as such. For example, `utils::seqwait`.
-
-#### Rationale
-
-The alternative would be to make all async code cancellation-safe,
-unless otherwise marked. That way, you could use `tokio::select!` more
-liberally. The reasons we didn't choose that are explained in this
-section.
-
-Writing code in a cancellation-safe manner is tedious, as you need to
-scrutinize every `.await` and ensure that if the `.await` call never
-returns, the system is in a safe, consistent state. In some ways, you
-need to do that with `?` and early `returns`, too, but `.await`s are
-easier to miss. It is also easier to perform cleanup tasks when a
-function returns an `Err` than when an `.await` simply never
-returns. You can use `scopeguard` and Drop guards to perform cleanup
-tasks, but it is more tedious. An `.await` that never returns is more
-similar to a panic.
-
-Note that even if you only use building blocks that themselves are
-cancellation-safe, it doesn't mean that the code as whole is
-cancellation-safe. For example, consider the following code:
-
-```
-while let Some(i) = work_inbox.recv().await {
-	if let Err(_) = results_outbox.send(i).await {
-		println!("receiver dropped");
-		return;
-		}
-	}
-}
-```
-
-It reads messages from one channel, sends them to another channel. If
-this code is cancelled at the `results_outbox.send(i).await`, the
-message read from the receiver is lost. That may or may not be OK,
-depending on the context.
-
-Another reason to not require cancellation-safety is historical: we
-already had a lot of async code that was not scrutinized for
-cancellation-safety when this issue was raised. Scrutinizing all
-existing code is no fun.
+Be very careful when mixing sync and async code!

docs/rfcs/023-the-state-of-pageserver-tenant-relocation.md

@@ -1,232 +0,0 @@
-# The state of pageserver tenant relocation
-
-Created on 17.03.23
-
-## Motivation
-
-There were previous write ups on the subject. The design of tenant relocation was planned at the time when we had quite different landscape. I e there was no on-demand download/eviction. They were on the horizon but we still planned for cases when they were not available. Some other things have changed. Now safekeepers offload wal to s3 so we're not risking overflowing their disks. Having all of the above, it makes sense to recap and take a look at the options we have now, which adjustments we'd like to make to original process, etc.
-
-Related (in chronological order):
-
-- Tracking issue with initial discussion: [#886](https://github.com/neondatabase/neon/issues/886)
-- [015. Storage Messaging](015-storage-messaging.md)
-- [020. Pageserver S3 Coordination](020-pageserver-s3-coordination.md)
-
-## Summary
-
-The RFC consists of a walkthrough of prior art on tenant relocation and corresponding problems. It describes 3 approaches.
-
-1. Simplistic approach that uses ignore and is the fastest to implement. The main downside is a requirement of short downtime.
-2. More complicated approach that avoids even short downtime.
-3. Even more complicated approach that will allow multiple pageservers to operate concurrently on the same tenant possibly allowing for HA cluster topologies and horizontal scaling of reads (i e compute talks to multiple pageservers).
-
-The order in which solutions are described is a bit different. We start from 2, then move to possible compromises (aka simplistic approach) and then move to discussing directions for solving HA/Pageserver replica case with 3.
-
-## Components
-
-pageserver, control-plane, safekeepers (a bit)
-
-## Requirements
-
-Relocation procedure should move tenant from one pageserver to another without downtime introduced by storage side. For now restarting compute for applying new configuration is fine.
-
-- component restarts
-- component outage
-- pageserver loss
-
-## The original proposed implementation
-
-The starting point is this sequence:
-
-```mermaid
-sequenceDiagram
-    autonumber
-    participant CP as Control Plane
-    participant PS1 as Pageserver 1
-    participant PS2 as Pageserver 2
-    participant S3
-
-    CP->>PS2: Attach tenant X
-    PS2->>S3: Fetch timelines, indexes for them
-    PS2->>CP: Accepted
-    CP->>CP: Change pageserver id in project
-    CP->>PS1: Detach
-```
-
-Which problems do we have with naive approach?
-
-### Concurrent GC and Compaction
-
-The problem is that they can run on both, PS1 and PS2. Consider this example from [Pageserver S3 Coordination RFC](020-pageserver-s3-coordination.md)
-
-```mermaid
-sequenceDiagram
-    autonumber
-    participant PS1
-    participant S3
-    participant PS2
-
-    PS1->>S3: Uploads L1, L2 <br/> Index contains L1 L2
-    PS2->>S3: Attach called, sees L1, L2
-    PS1->>S3: Compaction comes <br/> Removes L1, adds L3
-    note over S3: Index now L2, L3
-    PS2->>S3: Uploads new layer L4 <br/> (added to previous view of the index)
-    note over S3: Index now L1, L2, L4
-```
-
-At this point it is not possible to restore the state from index, it contains L2 which
-is no longer available in s3 and doesnt contain L3 added by compaction by the
-first pageserver. So if any of the pageservers restart, initial sync will fail
-(or in on-demand world it will fail a bit later during page request from
-missing layer)
-
-The problem lies in shared index_part.json. Having intersecting layers from append only edits is expected to work, though this is an uncharted territory without tests.
-
-#### Options
-
-There are several options on how to restrict concurrent access to index file.
-
-First and the simplest one is external orchestration. Control plane which runs migration can use special api call on pageserver to stop background processes (gc, compaction), and even possibly all uploads.
-
-So the sequence becomes:
-
-```mermaid
-sequenceDiagram
-    autonumber
-    participant CP as Control Plane
-    participant PS1 as Pageserver 1
-    participant PS2 as Pageserver 2
-    participant S3
-
-    CP->>PS1: Pause background jobs, pause uploading new layers.
-    CP->>PS2: Attach tenant X.
-    PS2->>S3: Fetch timelines, index, start background operations
-    PS2->>CP: Accepted
-    CP->>CP: Monitor PS2 last record lsn, ensure OK lag
-    CP->>CP: Change pageserver id in project
-    CP->>PS1: Detach
-```
-
-The downside of this sequence is the potential rollback process. What if something goes wrong on new pageserver? Can we safely roll back to source pageserver?
-
-There are two questions:
-
-#### How can we detect that something went wrong?
-
-We can run usual availability check (consists of compute startup and an update of one row).
-Note that we cant run separate compute for that before touching compute that client runs actual workload on, because we cant have two simultaneous computes running in read-write mode on the same timeline (enforced by safekeepers consensus algorithm). So we can either run some readonly check first (basebackup) and then change pageserver id and run availability check. If it failed we can roll it back to the old one.
-
-#### What can go wrong? And how we can safely roll-back?
-
-In the sequence above during attach we start background processes/uploads. They change state in remote storage so it is possible that after rollback remote state will be different from one that was observed by source pageserver. So if target pageserver goes wild then source pageserver may fail to start with changed remote state.
-
-Proposed option would be to implement a barrier (read-only) mode when pageserver does not update remote state.
-
-So the sequence for happy path becomes this one:
-
-```mermaid
-sequenceDiagram
-    autonumber
-    participant CP as Control Plane
-    participant PS1 as Pageserver 1
-    participant PS2 as Pageserver 2
-    participant S3
-
-    CP->>PS1: Pause background jobs, pause uploading new layers.
-    CP->>PS2: Attach tenant X in remote readonly mode.
-    PS2->>S3: Fetch timelines, index
-    PS2->>CP: Accepted
-    CP->>CP: Monitor PS2 last record lsn, ensure OK lag
-    CP->>CP: Change pageserver id in project
-    CP->>CP: Run successful availability check
-    CP->>PS2: Start uploads, background tasks
-    CP->>PS1: Detach
-```
-
-With this sequence we restrict any changes to remote storage to one pageserver. So there is no concurrent access at all, not only for index_part.json, but for everything else too. This approach makes it possible to roll back after failure on new pageserver.
-
-The sequence with roll back process:
-
-```mermaid
-sequenceDiagram
-    autonumber
-    participant CP as Control Plane
-    participant PS1 as Pageserver 1
-    participant PS2 as Pageserver 2
-    participant S3
-
-    CP->>PS1: Pause background jobs, pause uploading new layers.
-    CP->>PS2: Attach tenant X in remote readonly mode.
-    PS2->>S3: Fetch timelines, index
-    PS2->>CP: Accepted
-    CP->>CP: Monitor PS2 last record lsn, ensure OK lag
-    CP->>CP: Change pageserver id in project
-    CP->>CP: Availability check Failed
-    CP->>CP: Change pageserver id back
-    CP->>PS1: Resume remote operations
-    CP->>PS2: Ignore (instead of detach for investigation purposes)
-```
-
-## Concurrent branch creation
-
-Another problem is a possibility of concurrent branch creation calls.
-
-I e during migration create_branch can be called on old pageserver and newly created branch wont be seen on new pageserver. Prior art includes prototyping an approach of trying to mirror such branches, but currently it lost its importance, because now attach is fast because we dont need to download all data, and additionally to the best of my knowledge of control plane internals (cc @ololobus to confirm) operations on one project are executed sequentially, so it is not possible to have such case. So branch create operation will be executed only when relocation is completed. As a safety measure we can forbid branch creation for tenants that are in readonly remote state.
-
-## Simplistic approach
-
-The difference of simplistic approach from one described above is that it calls ignore on source tenant first and then calls attach on target pageserver. Approach above does it in opposite order thus opening a possibility for race conditions we strive to avoid.
-
-The approach largely follows this guide: <https://github.com/neondatabase/cloud/wiki/Cloud:-Ad-hoc-tenant-relocation>
-
-The happy path sequence:
-
-```mermaid
-sequenceDiagram
-    autonumber
-    participant CP as Control Plane
-    participant PS1 as Pageserver 1
-    participant PS2 as Pageserver 2
-    participant SK as Safekeeper
-    participant S3
-
-    CP->>CP: Enable maintenance mode
-    CP->>PS1: Ignore
-    CP->>PS2: Attach
-    PS2->>CP: Accepted
-    loop Delete layers for each timeline
-        CP->>PS2: Get last record lsn
-        CP->>SK: Get commit lsn
-        CP->>CP: OK? Timed out?
-    end
-    CP->>CP: Change pageserver id in project
-    CP->>CP: Run successful availability check
-    CP->>CP: Disable maintenance mode
-    CP->>PS1: Detach ignored
-```
-
-The sequence contains exactly the same rollback problems as in previous approach described above. They can be resolved the same way.
-
-Most probably we'd like to move forward without this safety measure and implement it on top of this approach to make progress towards the downtime-less one.
-
-## Lease based approach
-
-In order to allow for concurrent operation on the same data on remote storage for multiple pageservers we need to go further than external orchestration.
-
-NOTE: [020. Pageserver S3 Coordination](020-pageserver-s3-coordination.md) discusses one more approach that relies on duplication of index_part.json for each pageserver operating on the timeline. This approach still requires external coordination which makes certain things easier but requires additional bookkeeping to account for multiple index_part.json files. Discussion/comparison with proposed lease based approach
-
-The problems are outlined in [020. Pageserver S3 Coordination](020-pageserver-s3-coordination.md) and suggested solution includes [Coordination based approach](020-pageserver-s3-coordination.md#coordination-based-approach). This way it will allow to do basic leader election for pageservers so they can decide which node will be responsible for running GC and compaction. The process is based on extensive communication via storage broker and consists of a lease that is taken by one of the pageservers that extends it to continue serving a leader role.
-
-There are two options for ingesting new data into pageserver in follower role. One option is to avoid WAL ingestion at all and rely on notifications from leader to discover new layers on s3. Main downside of this approach is that follower will always lag behind the primary node because it wont have the last layer until it is uploaded to remote storage. In case of a primary failure follower will be required to reingest last segment (up to 256Mb of WAL currently) which slows down recovery. Additionally if compute is connected to follower pageserver it will observe latest data with a delay. Queries from compute will likely experience bigger delays when recent lsn is required.
-
-The second option is to consume WAL stream on both pageservers. In this case the only problem is non deterministic layer generation. Additional bookkeeping will be required to deduplicate layers from primary with local ones. Some process needs to somehow merge them to remove duplicated data. Additionally we need to have good testing coverage to ensure that our implementation of `get_page@lsn` properly handles intersecting layers.
-
-There is another tradeoff. Approaches may be different in amount of traffic between system components. With first approach there can be increased traffic between follower and remote storage. But only in case follower has some activity that actually requests pages (!). With other approach traffic increase will be permanent and will be caused by two WAL streams instead of one.
-
-## Summary
-
-Proposed implementation strategy:
-
-Go with the simplest approach for now. Then work on tech debt, increase test coverage. Then gradually move forward to second approach by implementing safety measures first, finishing with switch of order between ignore and attach operation.
-
-And only then go to lease based approach to solve HA/Pageserver replica use cases.

libs/pageserver_api/src/models.rs

@@ -131,26 +131,11 @@ pub struct TimelineCreateRequest {
 }
 
 #[serde_as]
-#[derive(Serialize, Deserialize, Debug, Default)]
-#[serde(deny_unknown_fields)]
+#[derive(Serialize, Deserialize, Default)]
 pub struct TenantCreateRequest {
     #[serde(default)]
     #[serde_as(as = "Option<DisplayFromStr>")]
     pub new_tenant_id: Option<TenantId>,
-    #[serde(flatten)]
-    pub config: TenantConfig, // as we have a flattened field, we should reject all unknown fields in it
-}
-
-impl std::ops::Deref for TenantCreateRequest {
-    type Target = TenantConfig;
-
-    fn deref(&self) -> &Self::Target {
-        &self.config
-    }
-}
-
-#[derive(Serialize, Deserialize, Debug, Default)]
-pub struct TenantConfig {
     pub checkpoint_distance: Option<u64>,
     pub checkpoint_timeout: Option<String>,
     pub compaction_target_size: Option<u64>,
@@ -193,26 +178,37 @@ impl TenantCreateRequest {
 }
 
 #[serde_as]
-#[derive(Serialize, Deserialize, Debug)]
-#[serde(deny_unknown_fields)]
+#[derive(Serialize, Deserialize)]
 pub struct TenantConfigRequest {
     #[serde_as(as = "DisplayFromStr")]
     pub tenant_id: TenantId,
-    #[serde(flatten)]
-    pub config: TenantConfig, // as we have a flattened field, we should reject all unknown fields in it
-}
-
-impl std::ops::Deref for TenantConfigRequest {
-    type Target = TenantConfig;
-
-    fn deref(&self) -> &Self::Target {
-        &self.config
-    }
+    #[serde(default)]
+    pub checkpoint_distance: Option<u64>,
+    pub checkpoint_timeout: Option<String>,
+    pub compaction_target_size: Option<u64>,
+    pub compaction_period: Option<String>,
+    pub compaction_threshold: Option<usize>,
+    pub gc_horizon: Option<u64>,
+    pub gc_period: Option<String>,
+    pub image_creation_threshold: Option<usize>,
+    pub pitr_interval: Option<String>,
+    pub walreceiver_connect_timeout: Option<String>,
+    pub lagging_wal_timeout: Option<String>,
+    pub max_lsn_wal_lag: Option<NonZeroU64>,
+    pub trace_read_requests: Option<bool>,
+    // We defer the parsing of the eviction_policy field to the request handler.
+    // Otherwise we'd have to move the types for eviction policy into this package.
+    // We might do that once the eviction feature has stabilizied.
+    // For now, this field is not even documented in the openapi_spec.yml.
+    pub eviction_policy: Option<serde_json::Value>,
+    pub min_resident_size_override: Option<u64>,
+    pub evictions_low_residence_duration_metric_threshold: Option<String>,
 }
 
 impl TenantConfigRequest {
     pub fn new(tenant_id: TenantId) -> TenantConfigRequest {
-        let config = TenantConfig {
+        TenantConfigRequest {
+            tenant_id,
             checkpoint_distance: None,
             checkpoint_timeout: None,
             compaction_target_size: None,
@@ -229,8 +225,7 @@ impl TenantConfigRequest {
             eviction_policy: None,
             min_resident_size_override: None,
             evictions_low_residence_duration_metric_threshold: None,
-        };
-        TenantConfigRequest { tenant_id, config }
+        }
     }
 }
 
@@ -770,31 +765,4 @@ mod tests {
         assert!(format!("{:?}", &original_broken.state).contains("reason"));
         assert!(format!("{:?}", &original_broken.state).contains("backtrace info"));
     }
-
-    #[test]
-    fn test_reject_unknown_field() {
-        let id = TenantId::generate();
-        let create_request = json!({
-            "new_tenant_id": id.to_string(),
-            "unknown_field": "unknown_value".to_string(),
-        });
-        let err = serde_json::from_value::<TenantCreateRequest>(create_request).unwrap_err();
-        assert!(
-            err.to_string().contains("unknown field `unknown_field`"),
-            "expect unknown field `unknown_field` error, got: {}",
-            err
-        );
-
-        let id = TenantId::generate();
-        let config_request = json!({
-            "tenant_id": id.to_string(),
-            "unknown_field": "unknown_value".to_string(),
-        });
-        let err = serde_json::from_value::<TenantConfigRequest>(config_request).unwrap_err();
-        assert!(
-            err.to_string().contains("unknown field `unknown_field`"),
-            "expect unknown field `unknown_field` error, got: {}",
-            err
-        );
-    }
 }

libs/remote_storage/Cargo.toml

@@ -12,7 +12,6 @@ aws-smithy-http.workspace = true
 aws-types.workspace = true
 aws-config.workspace = true
 aws-sdk-s3.workspace = true
-aws-credential-types.workspace = true
 hyper = { workspace = true, features = ["stream"] }
 serde.workspace = true
 serde_json.workspace = true

libs/remote_storage/src/s3_bucket.rs

@@ -9,15 +9,14 @@ use std::sync::Arc;
 use anyhow::Context;
 use aws_config::{
     environment::credentials::EnvironmentVariableCredentialsProvider,
-    imds::credentials::ImdsCredentialsProvider, meta::credentials::CredentialsProviderChain,
+    imds::credentials::ImdsCredentialsProvider,
+    meta::credentials::{CredentialsProviderChain, LazyCachingCredentialsProvider},
 };
-use aws_credential_types::cache::CredentialsCache;
 use aws_sdk_s3::{
-    config::{Config, Region},
-    error::SdkError,
-    operation::get_object::GetObjectError,
-    primitives::ByteStream,
-    Client,
+    config::Config,
+    error::{GetObjectError, GetObjectErrorKind},
+    types::{ByteStream, SdkError},
+    Client, Endpoint, Region,
 };
 use aws_smithy_http::body::SdkBody;
 use hyper::Body;
@@ -126,23 +125,28 @@ impl S3Bucket {
 
         let credentials_provider = {
             // uses "AWS_ACCESS_KEY_ID", "AWS_SECRET_ACCESS_KEY"
-            CredentialsProviderChain::first_try(
-                "env",
-                EnvironmentVariableCredentialsProvider::new(),
-            )
+            let env_creds = EnvironmentVariableCredentialsProvider::new();
             // uses imds v2
-            .or_else("imds", ImdsCredentialsProvider::builder().build())
+            let imds = ImdsCredentialsProvider::builder().build();
+
+            // finally add caching.
+            // this might change in future, see https://github.com/awslabs/aws-sdk-rust/issues/629
+            LazyCachingCredentialsProvider::builder()
+                .load(CredentialsProviderChain::first_try("env", env_creds).or_else("imds", imds))
+                .build()
         };
 
         let mut config_builder = Config::builder()
             .region(Region::new(aws_config.bucket_region.clone()))
-            .credentials_cache(CredentialsCache::lazy())
             .credentials_provider(credentials_provider);
 
         if let Some(custom_endpoint) = aws_config.endpoint.clone() {
-            config_builder = config_builder
-                .endpoint_url(custom_endpoint)
-                .force_path_style(true);
+            let endpoint = Endpoint::immutable(
+                custom_endpoint
+                    .parse()
+                    .expect("Failed to parse S3 custom endpoint"),
+            );
+            config_builder.set_endpoint_resolver(Some(Arc::new(endpoint)));
         }
         let client = Client::from_conf(config_builder.build());
 
@@ -225,9 +229,14 @@ impl S3Bucket {
                     ))),
                 })
             }
-            Err(SdkError::ServiceError(e)) if matches!(e.err(), GetObjectError::NoSuchKey(_)) => {
-                Err(DownloadError::NotFound)
-            }
+            Err(SdkError::ServiceError {
+                err:
+                    GetObjectError {
+                        kind: GetObjectErrorKind::NoSuchKey(..),
+                        ..
+                    },
+                ..
+            }) => Err(DownloadError::NotFound),
             Err(e) => {
                 metrics::inc_get_object_fail();
                 Err(DownloadError::Other(anyhow::anyhow!(

libs/utils/src/seqwait.rs

@@ -144,8 +144,6 @@ where
     ///
     /// This call won't complete until someone has called `advance`
     /// with a number greater than or equal to the one we're waiting for.
-    ///
-    /// This function is async cancellation-safe.
     pub async fn wait_for(&self, num: V) -> Result<(), SeqWaitError> {
         match self.queue_for_wait(num) {
             Ok(None) => Ok(()),
@@ -161,8 +159,6 @@ where
     ///
     /// If that hasn't happened after the specified timeout duration,
     /// [`SeqWaitError::Timeout`] will be returned.
-    ///
-    /// This function is async cancellation-safe.
     pub async fn wait_for_timeout(
         &self,
         num: V,

pageserver/src/bin/pageserver_binutils.rs

@@ -110,18 +110,6 @@ fn handle_metadata(path: &Path, arg_matches: &clap::ArgMatches) -> Result<(), an
         );
         update_meta = true;
     }
-    if let Some(latest_gc_cuttoff) = arg_matches.get_one::<String>("latest_gc_cuttoff") {
-        meta = TimelineMetadata::new(
-            meta.disk_consistent_lsn(),
-            meta.prev_record_lsn(),
-            meta.ancestor_timeline(),
-            meta.ancestor_lsn(),
-            Lsn::from_str(latest_gc_cuttoff)?,
-            meta.initdb_lsn(),
-            meta.pg_version(),
-        );
-        update_meta = true;
-    }
 
     if update_meta {
         let metadata_bytes = meta.to_bytes()?;
@@ -159,11 +147,6 @@ fn cli() -> Command {
                     Arg::new("prev_record_lsn")
                         .long("prev_record_lsn")
                         .help("Replace previous record Lsn"),
-                )
-                .arg(
-                    Arg::new("latest_gc_cuttoff")
-                        .long("latest_gc_cuttoff")
-                        .help("Replace latest gc cuttoff"),
                 ),
         )
 }

pageserver/src/config.rs

@@ -797,8 +797,7 @@ impl PageServerConf {
             )?);
         }
         if let Some(max_lsn_wal_lag) = item.get("max_lsn_wal_lag") {
-            t_conf.max_lsn_wal_lag =
-                Some(deserialize_from_item("max_lsn_wal_lag", max_lsn_wal_lag)?);
+            t_conf.max_lsn_wal_lag = Some(parse_toml_from_str("max_lsn_wal_lag", max_lsn_wal_lag)?);
         }
         if let Some(trace_read_requests) = item.get("trace_read_requests") {
             t_conf.trace_read_requests =

pageserver/src/http/openapi_spec.yml

@@ -741,16 +741,13 @@ paths:
                 $ref: "#/components/schemas/Error"
     post:
       description: |
-        Create a tenant. Returns new tenant id on success.
-
+        Create a tenant. Returns new tenant id on success.\
         If no new tenant id is specified in parameters, it would be generated. It's an error to recreate the same tenant.
-
-        Invalid fields in the tenant config will cause the request to be rejected with status 400.
       requestBody:
         content:
           application/json:
             schema:
-              $ref: "#/components/schemas/TenantCreateRequest"
+              $ref: "#/components/schemas/TenantCreateInfo"
       responses:
         "201":
           description: New tenant created successfully
@@ -793,13 +790,11 @@ paths:
     put:
       description: |
         Update tenant's config.
-
-        Invalid fields in the tenant config will cause the request to be rejected with status 400.
       requestBody:
         content:
           application/json:
             schema:
-              $ref: "#/components/schemas/TenantConfigRequest"
+              $ref: "#/components/schemas/TenantConfigInfo"
       responses:
         "200":
           description: OK
@@ -851,7 +846,7 @@ paths:
           content:
             application/json:
               schema:
-                $ref: "#/components/schemas/TenantConfigResponse"
+                $ref: "#/components/schemas/TenantConfig"
         "400":
           description: Malformed get tenanant config request
           content:
@@ -914,27 +909,35 @@ components:
             See the tenant `/attach` endpoint for more information.
           type: string
           enum: [ "maybe", "attached" ]
-    TenantCreateRequest:
-      allOf:
-        - $ref: '#/components/schemas/TenantConfig'
-        - type: object
-          properties:
-            new_tenant_id:
-              type: string
-              format: hex
-    TenantConfigRequest:
-      allOf:
-        - $ref: '#/components/schemas/TenantConfig'
-        - type: object
-          required:
-            - tenant_id
-          properties:
-            tenant_id:
-              type: string
-              format: hex
-    TenantConfig:
+    TenantCreateInfo:
       type: object
       properties:
+        new_tenant_id:
+          type: string
+          format: hex
+        tenant_id:
+          type: string
+          format: hex
+        gc_period:
+          type: string
+        gc_horizon:
+          type: integer
+        pitr_interval:
+          type: string
+        checkpoint_distance:
+          type: integer
+        checkpoint_timeout:
+          type: string
+        compaction_period:
+          type: string
+        compaction_threshold:
+          type: string
+    TenantConfigInfo:
+      type: object
+      properties:
+        tenant_id:
+          type: string
+          format: hex
         gc_period:
           type: string
         gc_horizon:
@@ -961,13 +964,13 @@ components:
           type: integer
         trace_read_requests:
           type: boolean
-    TenantConfigResponse:
+    TenantConfig:
       type: object
       properties:
         tenant_specific_overrides:
-          $ref: "#/components/schemas/TenantConfig"
+          $ref: "#/components/schemas/TenantConfigInfo"
         effective_config:
-          $ref: "#/components/schemas/TenantConfig"
+          $ref: "#/components/schemas/TenantConfigInfo"
     TimelineInfo:
       type: object
       required:

pageserver/src/http/routes.rs

@@ -19,7 +19,6 @@ use super::models::{
 };
 use crate::context::{DownloadBehavior, RequestContext};
 use crate::disk_usage_eviction_task;
-use crate::metrics::{StorageTimeOperation, STORAGE_TIME_GLOBAL};
 use crate::pgdatadir_mapping::LsnForTimestamp;
 use crate::task_mgr::TaskKind;
 use crate::tenant::config::TenantConfOpt;
@@ -398,17 +397,9 @@ async fn tenant_attach_handler(request: Request<Body>) -> Result<Response<Body>,
     let state = get_state(&request);
 
     if let Some(remote_storage) = &state.remote_storage {
-        mgr::attach_tenant(
-            state.conf,
-            tenant_id,
-            // XXX: Attach should provide the config, especially during tenant migration.
-            //      See https://github.com/neondatabase/neon/issues/1555
-            TenantConfOpt::default(),
-            remote_storage.clone(),
-            &ctx,
-        )
-        .instrument(info_span!("tenant_attach", tenant = %tenant_id))
-        .await?;
+        mgr::attach_tenant(state.conf, tenant_id, remote_storage.clone(), &ctx)
+            .instrument(info_span!("tenant_attach", tenant = %tenant_id))
+            .await?;
     } else {
         return Err(ApiError::BadRequest(anyhow!(
             "attach_tenant is not possible because pageserver was configured without remote storage"
@@ -717,17 +708,11 @@ pub fn html_response(status: StatusCode, data: String) -> Result<Response<Body>,
 async fn tenant_create_handler(mut request: Request<Body>) -> Result<Response<Body>, ApiError> {
     check_permission(&request, None)?;
 
-    let _timer = STORAGE_TIME_GLOBAL
-        .get_metric_with_label_values(&[StorageTimeOperation::CreateTenant.into()])
-        .expect("bug")
-        .start_timer();
-
     let ctx = RequestContext::new(TaskKind::MgmtRequest, DownloadBehavior::Warn);
 
     let request_data: TenantCreateRequest = json_request(&mut request).await?;
 
-    let tenant_conf =
-        TenantConfOpt::try_from(&request_data.config).map_err(ApiError::BadRequest)?;
+    let tenant_conf = TenantConfOpt::try_from(&request_data).map_err(ApiError::BadRequest)?;
 
     let target_tenant_id = request_data
         .new_tenant_id
@@ -758,7 +743,6 @@ async fn tenant_create_handler(mut request: Request<Body>) -> Result<Response<Bo
         res.context("created tenant failed to become active")
             .map_err(ApiError::InternalServerError)?;
     }
-
     json_response(
         StatusCode::CREATED,
         TenantCreateResponse(new_tenant.tenant_id()),
@@ -796,8 +780,7 @@ async fn update_tenant_config_handler(
     let tenant_id = request_data.tenant_id;
     check_permission(&request, Some(tenant_id))?;
 
-    let tenant_conf =
-        TenantConfOpt::try_from(&request_data.config).map_err(ApiError::BadRequest)?;
+    let tenant_conf = TenantConfOpt::try_from(&request_data).map_err(ApiError::BadRequest)?;
 
     let state = get_state(&request);
     mgr::set_new_tenant_config(state.conf, tenant_conf, tenant_id)

pageserver/src/keyspace.rs

@@ -62,6 +62,46 @@ impl KeySpace {
         KeyPartitioning { parts }
     }
 
+    ///
+    /// Calculate logical size of delta layers: total size of all blocks covered by it's key range
+    ///
+    pub fn get_logical_size(&self, range: &Range<Key>) -> u64 {
+        let mut start_key = range.start;
+        let n_ranges = self.ranges.len();
+        let start_index = match self.ranges.binary_search_by_key(&start_key, |r| r.start) {
+            Ok(index) => index, // keyspace range starts with start_key
+            Err(index) => {
+                if index != 0 && self.ranges[index - 1].end > start_key {
+                    index - 1 // previous keyspace range overlaps with specified
+                } else if index == n_ranges {
+                    return 0; // no intersection with specified range
+                } else {
+                    start_key = self.ranges[index].start;
+                    index
+                }
+            }
+        };
+        let mut size = 0u64;
+        for i in start_index..n_ranges {
+            if self.ranges[i].start >= range.end {
+                break;
+            }
+            let end_key = if self.ranges[i].end < range.end {
+                self.ranges[i].end
+            } else {
+                range.end
+            };
+            let n_blocks = key_range_size(&(start_key..end_key));
+            if n_blocks != u32::MAX {
+                size += n_blocks as u64 * BLCKSZ as u64;
+            }
+            if i + 1 < n_ranges {
+                start_key = self.ranges[i + 1].start;
+            }
+        }
+        size
+    }
+
     ///
     /// Check if key space contains overlapping range
     ///

pageserver/src/metrics.rs

@@ -8,7 +8,6 @@ use metrics::{
 use once_cell::sync::Lazy;
 use pageserver_api::models::TenantState;
 use strum::VariantNames;
-use strum_macros::{EnumVariantNames, IntoStaticStr};
 use utils::id::{TenantId, TimelineId};
 
 /// Prometheus histogram buckets (in seconds) for operations in the critical
@@ -25,33 +24,16 @@ const CRITICAL_OP_BUCKETS: &[f64] = &[
 ];
 
 // Metrics collected on operations on the storage repository.
-#[derive(Debug, EnumVariantNames, IntoStaticStr)]
-#[strum(serialize_all = "kebab_case")]
-pub enum StorageTimeOperation {
-    #[strum(serialize = "layer flush")]
-    LayerFlush,
-
-    #[strum(serialize = "compact")]
-    Compact,
-
-    #[strum(serialize = "create images")]
-    CreateImages,
-
-    #[strum(serialize = "logical size")]
-    LogicalSize,
-
-    #[strum(serialize = "imitate logical size")]
-    ImitateLogicalSize,
-
-    #[strum(serialize = "load layer map")]
-    LoadLayerMap,
-
-    #[strum(serialize = "gc")]
-    Gc,
-
-    #[strum(serialize = "create tenant")]
-    CreateTenant,
-}
+const STORAGE_TIME_OPERATIONS: &[&str] = &[
+    "layer flush",
+    "compact",
+    "create images",
+    "init logical size",
+    "logical size",
+    "imitate logical size",
+    "load layer map",
+    "gc",
+];
 
 pub static STORAGE_TIME_SUM_PER_TIMELINE: Lazy<CounterVec> = Lazy::new(|| {
     register_counter_vec!(
@@ -690,9 +672,7 @@ pub struct StorageTimeMetrics {
 }
 
 impl StorageTimeMetrics {
-    pub fn new(operation: StorageTimeOperation, tenant_id: &str, timeline_id: &str) -> Self {
-        let operation: &'static str = operation.into();
-
+    pub fn new(operation: &str, tenant_id: &str, timeline_id: &str) -> Self {
         let timeline_sum = STORAGE_TIME_SUM_PER_TIMELINE
             .get_metric_with_label_values(&[operation, tenant_id, timeline_id])
             .unwrap();
@@ -756,23 +736,16 @@ impl TimelineMetrics {
         let materialized_page_cache_hit_counter = MATERIALIZED_PAGE_CACHE_HIT
             .get_metric_with_label_values(&[&tenant_id, &timeline_id])
             .unwrap();
-        let flush_time_histo =
-            StorageTimeMetrics::new(StorageTimeOperation::LayerFlush, &tenant_id, &timeline_id);
-        let compact_time_histo =
-            StorageTimeMetrics::new(StorageTimeOperation::Compact, &tenant_id, &timeline_id);
+        let flush_time_histo = StorageTimeMetrics::new("layer flush", &tenant_id, &timeline_id);
+        let compact_time_histo = StorageTimeMetrics::new("compact", &tenant_id, &timeline_id);
         let create_images_time_histo =
-            StorageTimeMetrics::new(StorageTimeOperation::CreateImages, &tenant_id, &timeline_id);
-        let logical_size_histo =
-            StorageTimeMetrics::new(StorageTimeOperation::LogicalSize, &tenant_id, &timeline_id);
-        let imitate_logical_size_histo = StorageTimeMetrics::new(
-            StorageTimeOperation::ImitateLogicalSize,
-            &tenant_id,
-            &timeline_id,
-        );
+            StorageTimeMetrics::new("create images", &tenant_id, &timeline_id);
+        let logical_size_histo = StorageTimeMetrics::new("logical size", &tenant_id, &timeline_id);
+        let imitate_logical_size_histo =
+            StorageTimeMetrics::new("imitate logical size", &tenant_id, &timeline_id);
         let load_layer_map_histo =
-            StorageTimeMetrics::new(StorageTimeOperation::LoadLayerMap, &tenant_id, &timeline_id);
-        let garbage_collect_histo =
-            StorageTimeMetrics::new(StorageTimeOperation::Gc, &tenant_id, &timeline_id);
+            StorageTimeMetrics::new("load layer map", &tenant_id, &timeline_id);
+        let garbage_collect_histo = StorageTimeMetrics::new("gc", &tenant_id, &timeline_id);
         let last_record_gauge = LAST_RECORD_LSN
             .get_metric_with_label_values(&[&tenant_id, &timeline_id])
             .unwrap();
@@ -840,7 +813,7 @@ impl Drop for TimelineMetrics {
             .write()
             .unwrap()
             .remove(tenant_id, timeline_id);
-        for op in StorageTimeOperation::VARIANTS {
+        for op in STORAGE_TIME_OPERATIONS {
             let _ =
                 STORAGE_TIME_SUM_PER_TIMELINE.remove_label_values(&[op, tenant_id, timeline_id]);
             let _ =

pageserver/src/tenant.rs

@@ -602,9 +602,12 @@ impl Tenant {
         remote_storage: GenericRemoteStorage,
         ctx: &RequestContext,
     ) -> anyhow::Result<Arc<Tenant>> {
-        // TODO dedup with spawn_load
-        let tenant_conf =
-            Self::load_tenant_config(conf, tenant_id).context("load tenant config")?;
+        // XXX: Attach should provide the config, especially during tenant migration.
+        //      See https://github.com/neondatabase/neon/issues/1555
+        let tenant_conf = TenantConfOpt::default();
+
+        Self::attach_idempotent_create_marker_file(conf, tenant_id)
+            .context("create attach marker file")?;
 
         let wal_redo_manager = Arc::new(PostgresRedoManager::new(conf, tenant_id));
         let tenant = Arc::new(Tenant::new(
@@ -641,6 +644,45 @@ impl Tenant {
         Ok(tenant)
     }
 
+    fn attach_idempotent_create_marker_file(
+        conf: &'static PageServerConf,
+        tenant_id: TenantId,
+    ) -> anyhow::Result<()> {
+        // Create directory with marker file to indicate attaching state.
+        // The load_local_tenants() function in tenant::mgr relies on the marker file
+        // to determine whether a tenant has finished attaching.
+        let tenant_dir = conf.tenant_path(&tenant_id);
+        let marker_file = conf.tenant_attaching_mark_file_path(&tenant_id);
+        debug_assert_eq!(marker_file.parent().unwrap(), tenant_dir);
+        // TODO: should use tokio::fs here, but
+        // 1. caller is not async, for good reason (it holds tenants map lock)
+        // 2. we'd need to think about cancel safety. Turns out dropping a tokio::fs future
+        //    doesn't wait for the activity in the fs thread pool.
+        crashsafe::create_dir_all(&tenant_dir).context("create tenant directory")?;
+        match fs::OpenOptions::new()
+            .write(true)
+            .create_new(true)
+            .open(&marker_file)
+        {
+            Ok(_) => {}
+            Err(e) if e.kind() == std::io::ErrorKind::AlreadyExists => {
+                // Either this is a retry of attach or there is a concurrent task also doing attach for this tenant.
+                // We cannot distinguish this here.
+                // The caller is responsible for ensuring there's no concurrent attach for a tenant.
+                {} // fsync again, we don't know if that already happened
+            }
+            err => {
+                err.context("create tenant attaching marker file")?;
+                unreachable!("we covered the Ok() case above");
+            }
+        }
+        crashsafe::fsync_file_and_parent(&marker_file)
+            .context("fsync tenant attaching marker file and parent")?;
+        debug_assert!(tenant_dir.is_dir());
+        debug_assert!(marker_file.is_file());
+        Ok(())
+    }
+
     ///
     /// Background task that downloads all data for a tenant and brings it to Active state.
     ///
@@ -792,8 +834,6 @@ impl Tenant {
         remote_client: RemoteTimelineClient,
         ctx: &RequestContext,
     ) -> anyhow::Result<()> {
-        debug_assert_current_span_has_tenant_id();
-
         info!("downloading index file for timeline {}", timeline_id);
         tokio::fs::create_dir_all(self.conf.timeline_path(&timeline_id, &self.tenant_id))
             .await
@@ -1058,8 +1098,6 @@ impl Tenant {
         local_metadata: TimelineMetadata,
         ctx: &RequestContext,
     ) -> anyhow::Result<()> {
-        debug_assert_current_span_has_tenant_id();
-
         let remote_client = self.remote_storage.as_ref().map(|remote_storage| {
             RemoteTimelineClient::new(
                 remote_storage.clone(),
@@ -1589,8 +1627,6 @@ impl Tenant {
 
     /// Changes tenant status to active, unless shutdown was already requested.
     fn activate(&self, ctx: &RequestContext) -> anyhow::Result<()> {
-        debug_assert_current_span_has_tenant_id();
-
         let mut result = Ok(());
         self.state.send_modify(|current_state| {
             match &*current_state {
@@ -2082,7 +2118,6 @@ impl Tenant {
             // enough to just fsync it always.
 
             crashsafe::fsync(target_config_parent)?;
-            // XXX we're not fsyncing the parent dir, need to do that in case `creating_tenant`
             Ok(())
         };
 
@@ -2160,7 +2195,7 @@ impl Tenant {
                 // made.
                 break;
             }
-            let result = timeline.gc().await?;
+            let result = timeline.gc(ctx).await?;
             totals += result;
         }
 
@@ -2726,23 +2761,15 @@ fn remove_timeline_and_uninit_mark(timeline_dir: &Path, uninit_mark: &Path) -> a
     Ok(())
 }
 
-pub(crate) enum CreateTenantFilesMode {
-    Create,
-    Attach,
-}
-
 pub(crate) fn create_tenant_files(
     conf: &'static PageServerConf,
     tenant_conf: TenantConfOpt,
     tenant_id: TenantId,
-    mode: CreateTenantFilesMode,
 ) -> anyhow::Result<PathBuf> {
     let target_tenant_directory = conf.tenant_path(&tenant_id);
     anyhow::ensure!(
-        !target_tenant_directory
-            .try_exists()
-            .context("check existence of tenant directory")?,
-        "tenant directory already exists",
+        !target_tenant_directory.exists(),
+        "cannot create new tenant repo: '{tenant_id}' directory already exists",
     );
 
     let temporary_tenant_dir =
@@ -2764,7 +2791,6 @@ pub(crate) fn create_tenant_files(
         conf,
         tenant_conf,
         tenant_id,
-        mode,
         &temporary_tenant_dir,
         &target_tenant_directory,
     );
@@ -2789,28 +2815,9 @@ fn try_create_target_tenant_dir(
     conf: &'static PageServerConf,
     tenant_conf: TenantConfOpt,
     tenant_id: TenantId,
-    mode: CreateTenantFilesMode,
     temporary_tenant_dir: &Path,
     target_tenant_directory: &Path,
 ) -> Result<(), anyhow::Error> {
-    match mode {
-        CreateTenantFilesMode::Create => {} // needs no attach marker, writing tenant conf + atomic rename of dir is good enough
-        CreateTenantFilesMode::Attach => {
-            let attach_marker_path = temporary_tenant_dir.join(TENANT_ATTACHING_MARKER_FILENAME);
-            let file = std::fs::OpenOptions::new()
-                .create_new(true)
-                .write(true)
-                .open(&attach_marker_path)
-                .with_context(|| {
-                    format!("could not create attach marker file {attach_marker_path:?}")
-                })?;
-            file.sync_all().with_context(|| {
-                format!("could not sync attach marker file: {attach_marker_path:?}")
-            })?;
-            // fsync of the directory in which the file resides comes later in this function
-        }
-    }
-
     let temporary_tenant_timelines_dir = rebase_directory(
         &conf.timelines_path(&tenant_id),
         target_tenant_directory,
@@ -2837,11 +2844,6 @@ fn try_create_target_tenant_dir(
         anyhow::bail!("failpoint tenant-creation-before-tmp-rename");
     });
 
-    // Make sure the current tenant directory entries are durable before renaming.
-    // Without this, a crash may reorder any of the directory entry creations above.
-    crashsafe::fsync(temporary_tenant_dir)
-        .with_context(|| format!("sync temporary tenant directory {temporary_tenant_dir:?}"))?;
-
     fs::rename(temporary_tenant_dir, target_tenant_directory).with_context(|| {
         format!(
             "move tenant {} temporary directory {} into the permanent one {}",
@@ -3752,7 +3754,7 @@ mod tests {
                 .await?;
             tline.freeze_and_flush().await?;
             tline.compact(&ctx).await?;
-            tline.gc().await?;
+            tline.gc(&ctx).await?;
         }
 
         Ok(())
@@ -3824,7 +3826,7 @@ mod tests {
                 .await?;
             tline.freeze_and_flush().await?;
             tline.compact(&ctx).await?;
-            tline.gc().await?;
+            tline.gc(&ctx).await?;
         }
 
         Ok(())
@@ -3908,7 +3910,7 @@ mod tests {
                 .await?;
             tline.freeze_and_flush().await?;
             tline.compact(&ctx).await?;
-            tline.gc().await?;
+            tline.gc(&ctx).await?;
         }
 
         Ok(())
@@ -3977,28 +3979,3 @@ mod tests {
         Ok(())
     }
 }
-
-#[cfg(not(debug_assertions))]
-#[inline]
-pub(crate) fn debug_assert_current_span_has_tenant_id() {}
-
-#[cfg(debug_assertions)]
-pub static TENANT_ID_EXTRACTOR: once_cell::sync::Lazy<
-    utils::tracing_span_assert::MultiNameExtractor<2>,
-> = once_cell::sync::Lazy::new(|| {
-    utils::tracing_span_assert::MultiNameExtractor::new("TenantId", ["tenant_id", "tenant"])
-});
-
-#[cfg(debug_assertions)]
-#[inline]
-pub(crate) fn debug_assert_current_span_has_tenant_id() {
-    use utils::tracing_span_assert;
-
-    match tracing_span_assert::check_fields_present([&*TENANT_ID_EXTRACTOR]) {
-        Ok(()) => (),
-        Err(missing) => panic!(
-            "missing extractors: {:?}",
-            missing.into_iter().map(|e| e.name()).collect::<Vec<_>>()
-        ),
-    }
-}

pageserver/src/tenant/config.rs

@@ -9,7 +9,7 @@
 //! may lead to a data loss.
 //!
 use anyhow::Context;
-use pageserver_api::models;
+use pageserver_api::models::{TenantConfigRequest, TenantCreateRequest};
 use serde::{Deserialize, Serialize};
 use std::num::NonZeroU64;
 use std::time::Duration;
@@ -292,77 +292,93 @@ fn bad_duration<'a>(field_name: &'static str, value: &'a str) -> impl 'a + Fn()
     move || format!("Cannot parse `{field_name}` duration {value:?}")
 }
 
-impl TryFrom<&'_ models::TenantConfig> for TenantConfOpt {
-    type Error = anyhow::Error;
-
-    fn try_from(request_data: &'_ models::TenantConfig) -> Result<Self, Self::Error> {
+impl TenantConfOpt {
+    #[allow(clippy::too_many_arguments)]
+    fn from_request(
+        checkpoint_distance: Option<u64>,
+        checkpoint_timeout: &Option<String>,
+        compaction_target_size: Option<u64>,
+        compaction_period: &Option<String>,
+        compaction_threshold: Option<usize>,
+        gc_horizon: Option<u64>,
+        gc_period: &Option<String>,
+        image_creation_threshold: Option<usize>,
+        pitr_interval: &Option<String>,
+        walreceiver_connect_timeout: &Option<String>,
+        lagging_wal_timeout: &Option<String>,
+        max_lsn_wal_lag: Option<NonZeroU64>,
+        trace_read_requests: Option<bool>,
+        eviction_policy: &Option<serde_json::Value>,
+        min_resident_size_override: Option<u64>,
+        evictions_low_residence_duration_metric_threshold: &Option<String>,
+    ) -> Result<Self, anyhow::Error> {
         let mut tenant_conf = TenantConfOpt::default();
 
-        if let Some(gc_period) = &request_data.gc_period {
+        if let Some(gc_period) = &gc_period {
             tenant_conf.gc_period = Some(
                 humantime::parse_duration(gc_period)
                     .with_context(bad_duration("gc_period", gc_period))?,
             );
         }
-        tenant_conf.gc_horizon = request_data.gc_horizon;
-        tenant_conf.image_creation_threshold = request_data.image_creation_threshold;
+        tenant_conf.gc_horizon = gc_horizon;
+        tenant_conf.image_creation_threshold = image_creation_threshold;
 
-        if let Some(pitr_interval) = &request_data.pitr_interval {
+        if let Some(pitr_interval) = &pitr_interval {
             tenant_conf.pitr_interval = Some(
                 humantime::parse_duration(pitr_interval)
                     .with_context(bad_duration("pitr_interval", pitr_interval))?,
             );
         }
 
-        if let Some(walreceiver_connect_timeout) = &request_data.walreceiver_connect_timeout {
+        if let Some(walreceiver_connect_timeout) = &walreceiver_connect_timeout {
             tenant_conf.walreceiver_connect_timeout = Some(
                 humantime::parse_duration(walreceiver_connect_timeout).with_context(
                     bad_duration("walreceiver_connect_timeout", walreceiver_connect_timeout),
                 )?,
             );
         }
-        if let Some(lagging_wal_timeout) = &request_data.lagging_wal_timeout {
+        if let Some(lagging_wal_timeout) = &lagging_wal_timeout {
             tenant_conf.lagging_wal_timeout = Some(
                 humantime::parse_duration(lagging_wal_timeout)
                     .with_context(bad_duration("lagging_wal_timeout", lagging_wal_timeout))?,
             );
         }
-        if let Some(max_lsn_wal_lag) = request_data.max_lsn_wal_lag {
+        if let Some(max_lsn_wal_lag) = max_lsn_wal_lag {
             tenant_conf.max_lsn_wal_lag = Some(max_lsn_wal_lag);
         }
-        if let Some(trace_read_requests) = request_data.trace_read_requests {
+        if let Some(trace_read_requests) = trace_read_requests {
             tenant_conf.trace_read_requests = Some(trace_read_requests);
         }
 
-        tenant_conf.checkpoint_distance = request_data.checkpoint_distance;
-        if let Some(checkpoint_timeout) = &request_data.checkpoint_timeout {
+        tenant_conf.checkpoint_distance = checkpoint_distance;
+        if let Some(checkpoint_timeout) = &checkpoint_timeout {
             tenant_conf.checkpoint_timeout = Some(
                 humantime::parse_duration(checkpoint_timeout)
                     .with_context(bad_duration("checkpoint_timeout", checkpoint_timeout))?,
             );
         }
 
-        tenant_conf.compaction_target_size = request_data.compaction_target_size;
-        tenant_conf.compaction_threshold = request_data.compaction_threshold;
+        tenant_conf.compaction_target_size = compaction_target_size;
+        tenant_conf.compaction_threshold = compaction_threshold;
 
-        if let Some(compaction_period) = &request_data.compaction_period {
+        if let Some(compaction_period) = &compaction_period {
             tenant_conf.compaction_period = Some(
                 humantime::parse_duration(compaction_period)
                     .with_context(bad_duration("compaction_period", compaction_period))?,
             );
         }
 
-        if let Some(eviction_policy) = &request_data.eviction_policy {
+        if let Some(eviction_policy) = &eviction_policy {
             tenant_conf.eviction_policy = Some(
                 serde::Deserialize::deserialize(eviction_policy)
                     .context("parse field `eviction_policy`")?,
             );
         }
 
-        tenant_conf.min_resident_size_override = request_data.min_resident_size_override;
+        tenant_conf.min_resident_size_override = min_resident_size_override;
 
         if let Some(evictions_low_residence_duration_metric_threshold) =
-            &request_data.evictions_low_residence_duration_metric_threshold
+            &evictions_low_residence_duration_metric_threshold
         {
             tenant_conf.evictions_low_residence_duration_metric_threshold = Some(
                 humantime::parse_duration(evictions_low_residence_duration_metric_threshold)
@@ -377,6 +393,56 @@ impl TryFrom<&'_ models::TenantConfig> for TenantConfOpt {
     }
 }
 
+impl TryFrom<&'_ TenantCreateRequest> for TenantConfOpt {
+    type Error = anyhow::Error;
+
+    fn try_from(request_data: &TenantCreateRequest) -> Result<Self, Self::Error> {
+        Self::from_request(
+            request_data.checkpoint_distance,
+            &request_data.checkpoint_timeout,
+            request_data.compaction_target_size,
+            &request_data.compaction_period,
+            request_data.compaction_threshold,
+            request_data.gc_horizon,
+            &request_data.gc_period,
+            request_data.image_creation_threshold,
+            &request_data.pitr_interval,
+            &request_data.walreceiver_connect_timeout,
+            &request_data.lagging_wal_timeout,
+            request_data.max_lsn_wal_lag,
+            request_data.trace_read_requests,
+            &request_data.eviction_policy,
+            request_data.min_resident_size_override,
+            &request_data.evictions_low_residence_duration_metric_threshold,
+        )
+    }
+}
+
+impl TryFrom<&'_ TenantConfigRequest> for TenantConfOpt {
+    type Error = anyhow::Error;
+
+    fn try_from(request_data: &TenantConfigRequest) -> Result<Self, Self::Error> {
+        Self::from_request(
+            request_data.checkpoint_distance,
+            &request_data.checkpoint_timeout,
+            request_data.compaction_target_size,
+            &request_data.compaction_period,
+            request_data.compaction_threshold,
+            request_data.gc_horizon,
+            &request_data.gc_period,
+            request_data.image_creation_threshold,
+            &request_data.pitr_interval,
+            &request_data.walreceiver_connect_timeout,
+            &request_data.lagging_wal_timeout,
+            request_data.max_lsn_wal_lag,
+            request_data.trace_read_requests,
+            &request_data.eviction_policy,
+            request_data.min_resident_size_override,
+            &request_data.evictions_low_residence_duration_metric_threshold,
+        )
+    }
+}
+
 #[cfg(test)]
 mod tests {
     use super::*;

pageserver/src/tenant/mgr.rs

@@ -19,7 +19,7 @@ use crate::config::PageServerConf;
 use crate::context::{DownloadBehavior, RequestContext};
 use crate::task_mgr::{self, TaskKind};
 use crate::tenant::config::TenantConfOpt;
-use crate::tenant::{create_tenant_files, CreateTenantFilesMode, Tenant, TenantState};
+use crate::tenant::{Tenant, TenantState};
 use crate::IGNORED_TENANT_FILE_NAME;
 
 use utils::fs_ext::PathExt;
@@ -278,24 +278,19 @@ pub async fn create_tenant(
     remote_storage: Option<GenericRemoteStorage>,
     ctx: &RequestContext,
 ) -> Result<Arc<Tenant>, TenantMapInsertError> {
-    tenant_map_insert(tenant_id, || {
+    tenant_map_insert(tenant_id, |vacant_entry| {
         // We're holding the tenants lock in write mode while doing local IO.
         // If this section ever becomes contentious, introduce a new `TenantState::Creating`
         // and do the work in that state.
-        let tenant_directory = super::create_tenant_files(conf, tenant_conf, tenant_id, CreateTenantFilesMode::Create)?;
-        // TODO: tenant directory remains on disk if we bail out from here on.
-        //       See https://github.com/neondatabase/neon/issues/4233
-
+        let tenant_directory = super::create_tenant_files(conf, tenant_conf, tenant_id)?;
         let created_tenant =
             schedule_local_tenant_processing(conf, &tenant_directory, remote_storage, ctx)?;
-        // TODO: tenant object & its background loops remain, untracked in tenant map, if we fail here.
-        //      See https://github.com/neondatabase/neon/issues/4233
-
         let crated_tenant_id = created_tenant.tenant_id();
         anyhow::ensure!(
                 tenant_id == crated_tenant_id,
                 "loaded created tenant has unexpected tenant id (expect {tenant_id} != actual {crated_tenant_id})",
             );
+        vacant_entry.insert(Arc::clone(&created_tenant));
         Ok(created_tenant)
     }).await
 }
@@ -407,7 +402,7 @@ pub async fn load_tenant(
     remote_storage: Option<GenericRemoteStorage>,
     ctx: &RequestContext,
 ) -> Result<(), TenantMapInsertError> {
-    tenant_map_insert(tenant_id, || {
+    tenant_map_insert(tenant_id, |vacant_entry| {
         let tenant_path = conf.tenant_path(&tenant_id);
         let tenant_ignore_mark = conf.tenant_ignore_mark_file_path(tenant_id);
         if tenant_ignore_mark.exists() {
@@ -420,9 +415,9 @@ pub async fn load_tenant(
                 format!("Failed to schedule tenant processing in path {tenant_path:?}")
             })?;
 
-        Ok(new_tenant)
-    }).await?;
-    Ok(())
+        vacant_entry.insert(new_tenant);
+        Ok(())
+    }).await
 }
 
 pub async fn ignore_tenant(
@@ -471,35 +466,22 @@ pub async fn list_tenants() -> Result<Vec<(TenantId, TenantState)>, TenantMapLis
 pub async fn attach_tenant(
     conf: &'static PageServerConf,
     tenant_id: TenantId,
-    tenant_conf: TenantConfOpt,
     remote_storage: GenericRemoteStorage,
     ctx: &RequestContext,
 ) -> Result<(), TenantMapInsertError> {
-    tenant_map_insert(tenant_id, || {
-        let tenant_dir = create_tenant_files(conf, tenant_conf, tenant_id, CreateTenantFilesMode::Attach)?;
-        // TODO: tenant directory remains on disk if we bail out from here on.
-        //       See https://github.com/neondatabase/neon/issues/4233
-
-        // Without the attach marker, schedule_local_tenant_processing will treat the attached tenant as fully attached
-        let marker_file_exists = conf
-            .tenant_attaching_mark_file_path(&tenant_id)
-            .try_exists()
-            .context("check for attach marker file existence")?;
-        anyhow::ensure!(marker_file_exists, "create_tenant_files should have created the attach marker file");
-
-        let attached_tenant = schedule_local_tenant_processing(conf, &tenant_dir, Some(remote_storage), ctx)?;
-        // TODO: tenant object & its background loops remain, untracked in tenant map, if we fail here.
-        //      See https://github.com/neondatabase/neon/issues/4233
-
-        let attached_tenant_id = attached_tenant.tenant_id();
+    tenant_map_insert(tenant_id, |vacant_entry| {
+        let tenant_path = conf.tenant_path(&tenant_id);
         anyhow::ensure!(
-            tenant_id == attached_tenant_id,
-            "loaded created tenant has unexpected tenant id (expect {tenant_id} != actual {attached_tenant_id})",
+            !tenant_path.exists(),
+            "Cannot attach tenant {tenant_id}, local tenant directory already exists"
         );
-        Ok(attached_tenant)
+
+        let tenant =
+            Tenant::spawn_attach(conf, tenant_id, remote_storage, ctx).context("spawn_attach")?;
+        vacant_entry.insert(tenant);
+        Ok(())
     })
-    .await?;
-    Ok(())
+    .await
 }
 
 #[derive(Debug, thiserror::Error)]
@@ -520,12 +502,12 @@ pub enum TenantMapInsertError {
 ///
 /// NB: the closure should return quickly because the current implementation of tenants map
 /// serializes access through an `RwLock`.
-async fn tenant_map_insert<F>(
+async fn tenant_map_insert<F, V>(
     tenant_id: TenantId,
     insert_fn: F,
-) -> Result<Arc<Tenant>, TenantMapInsertError>
+) -> Result<V, TenantMapInsertError>
 where
-    F: FnOnce() -> anyhow::Result<Arc<Tenant>>,
+    F: FnOnce(hash_map::VacantEntry<TenantId, Arc<Tenant>>) -> anyhow::Result<V>,
 {
     let mut guard = TENANTS.write().await;
     let m = match &mut *guard {
@@ -538,11 +520,8 @@ where
             tenant_id,
             e.get().current_state(),
         )),
-        hash_map::Entry::Vacant(v) => match insert_fn() {
-            Ok(tenant) => {
-                v.insert(tenant.clone());
-                Ok(tenant)
-            }
+        hash_map::Entry::Vacant(v) => match insert_fn(v) {
+            Ok(v) => Ok(v),
             Err(e) => Err(TenantMapInsertError::Closure(e)),
         },
     }

pageserver/src/tenant/timeline.rs

@@ -955,7 +955,7 @@ impl Timeline {
 
     pub async fn wait_to_become_active(
         &self,
-        _ctx: &RequestContext, // Prepare for use by cancellation
+        _ctx: &RequestContext, /* Prepare for use by cancellation */
     ) -> Result<(), TimelineState> {
         let mut receiver = self.state.subscribe();
         loop {
@@ -3667,7 +3667,7 @@ impl Timeline {
     /// within a layer file. We can only remove the whole file if it's fully
     /// obsolete.
     ///
-    pub(super) async fn gc(&self) -> anyhow::Result<GcResult> {
+    pub(super) async fn gc(&self, ctx: &RequestContext) -> anyhow::Result<GcResult> {
         let timer = self.metrics.garbage_collect_histo.start_timer();
 
         fail_point!("before-timeline-gc");
@@ -3697,6 +3697,7 @@ impl Timeline {
                 pitr_cutoff,
                 retain_lsns,
                 new_gc_cutoff,
+                ctx,
             )
             .instrument(
                 info_span!("gc_timeline", timeline = %self.timeline_id, cutoff = %new_gc_cutoff),
@@ -3716,6 +3717,7 @@ impl Timeline {
         pitr_cutoff: Lsn,
         retain_lsns: Vec<Lsn>,
         new_gc_cutoff: Lsn,
+        ctx: &RequestContext,
     ) -> anyhow::Result<GcResult> {
         let now = SystemTime::now();
         let mut result: GcResult = GcResult::default();
@@ -3762,6 +3764,15 @@ impl Timeline {
 
         let mut layers_to_remove = Vec::new();
         let mut wanted_image_layers = KeySpaceRandomAccum::default();
+        // Do not collect keyspace for Unit tests
+        let gc_keyspace = if ctx.task_kind() == TaskKind::GarbageCollector {
+            Some(
+                self.collect_keyspace(self.get_last_record_lsn(), ctx)
+                    .await?,
+            )
+        } else {
+            None
+        };
 
         // Scan all layers in the timeline (remote or on-disk).
         //
@@ -3852,7 +3863,13 @@ impl Timeline {
                 // But image layers are in any case less sparse than delta layers. Also we need some
                 // protection from replacing recent image layers with new one after each GC iteration.
                 if l.is_incremental() && !LayerMap::is_l0(&*l) {
-                    wanted_image_layers.add_range(l.get_key_range());
+                    if let Some(keyspace) = &gc_keyspace {
+                        let layer_logical_size = keyspace.get_logical_size(&l.get_key_range());
+                        let layer_age = new_gc_cutoff.0 - l.get_lsn_range().start.0;
+                        if layer_logical_size <= layer_age {
+                            wanted_image_layers.add_range(l.get_key_range());
+                        }
+                    }
                 }
                 result.layers_not_updated += 1;
                 continue 'outer;
@@ -4471,6 +4488,12 @@ pub(crate) fn debug_assert_current_span_has_tenant_and_timeline_id() {}
 pub(crate) fn debug_assert_current_span_has_tenant_and_timeline_id() {
     use utils::tracing_span_assert;
 
+    pub static TENANT_ID_EXTRACTOR: once_cell::sync::Lazy<
+        tracing_span_assert::MultiNameExtractor<2>,
+    > = once_cell::sync::Lazy::new(|| {
+        tracing_span_assert::MultiNameExtractor::new("TenantId", ["tenant_id", "tenant"])
+    });
+
     pub static TIMELINE_ID_EXTRACTOR: once_cell::sync::Lazy<
         tracing_span_assert::MultiNameExtractor<2>,
     > = once_cell::sync::Lazy::new(|| {
@@ -4478,7 +4501,7 @@ pub(crate) fn debug_assert_current_span_has_tenant_and_timeline_id() {
     });
 
     match tracing_span_assert::check_fields_present([
-        &*super::TENANT_ID_EXTRACTOR,
+        &*TENANT_ID_EXTRACTOR,
         &*TIMELINE_ID_EXTRACTOR,
     ]) {
         Ok(()) => (),

pageserver/src/tenant/timeline/walreceiver/connection_manager.rs

@@ -28,8 +28,8 @@ use storage_broker::proto::SubscribeSafekeeperInfoRequest;
 use storage_broker::proto::TenantTimelineId as ProtoTenantTimelineId;
 use storage_broker::BrokerClientChannel;
 use storage_broker::Streaming;
-use tokio::select;
 use tokio::sync::RwLock;
+use tokio::{select, sync::watch};
 use tracing::*;
 
 use crate::{exponential_backoff, DEFAULT_BASE_BACKOFF_SECONDS, DEFAULT_MAX_BACKOFF_SECONDS};
@@ -50,13 +50,13 @@ pub(super) async fn connection_manager_loop_step(
     ctx: &RequestContext,
     manager_status: &RwLock<Option<ConnectionManagerStatus>>,
 ) -> ControlFlow<(), ()> {
-    match connection_manager_state
+    let mut timeline_state_updates = connection_manager_state
         .timeline
-        .wait_to_become_active(ctx)
-        .await
-    {
-        Ok(()) => {}
-        Err(_) => {
+        .subscribe_for_state_updates();
+
+    match wait_for_active_timeline(&mut timeline_state_updates).await {
+        ControlFlow::Continue(()) => {}
+        ControlFlow::Break(()) => {
             info!("Timeline dropped state updates sender before becoming active, stopping wal connection manager loop");
             return ControlFlow::Break(());
         }
@@ -72,10 +72,6 @@ pub(super) async fn connection_manager_loop_step(
         timeline_id: connection_manager_state.timeline.timeline_id,
     };
 
-    let mut timeline_state_updates = connection_manager_state
-        .timeline
-        .subscribe_for_state_updates();
-
     // Subscribe to the broker updates. Stream shares underlying TCP connection
     // with other streams on this client (other connection managers). When
     // object goes out of scope, stream finishes in drop() automatically.
@@ -199,6 +195,34 @@ pub(super) async fn connection_manager_loop_step(
     }
 }
 
+async fn wait_for_active_timeline(
+    timeline_state_updates: &mut watch::Receiver<TimelineState>,
+) -> ControlFlow<(), ()> {
+    let current_state = *timeline_state_updates.borrow();
+    if current_state == TimelineState::Active {
+        return ControlFlow::Continue(());
+    }
+
+    loop {
+        match timeline_state_updates.changed().await {
+            Ok(()) => {
+                let new_state = *timeline_state_updates.borrow();
+                match new_state {
+                    TimelineState::Active => {
+                        debug!("Timeline state changed to active, continuing the walreceiver connection manager");
+                        return ControlFlow::Continue(());
+                    }
+                    state => {
+                        debug!("Not running the walreceiver connection manager, timeline is not active: {state:?}");
+                        continue;
+                    }
+                }
+            }
+            Err(_sender_dropped_error) => return ControlFlow::Break(()),
+        }
+    }
+}
+
 /// Endlessly try to subscribe for broker updates for a given timeline.
 async fn subscribe_for_timeline_updates(
     broker_client: &mut BrokerClientChannel,

pageserver/src/walrecord.rs

@@ -379,6 +379,17 @@ impl XlXactParsedRecord {
                 });
             }
         }
+        if xinfo & pg_constants::XACT_XINFO_HAS_INVALS != 0 {
+            let nmsgs = buf.get_i32_le();
+            for _i in 0..nmsgs {
+                let sizeof_shared_invalidation_message = 0;
+                buf.advance(sizeof_shared_invalidation_message);
+            }
+        }
+        if xinfo & pg_constants::XACT_XINFO_HAS_TWOPHASE != 0 {
+            xid = buf.get_u32_le();
+            trace!("XLOG_XACT_COMMIT-XACT_XINFO_HAS_TWOPHASE");
+        }
 
         if xinfo & postgres_ffi::v15::bindings::XACT_XINFO_HAS_DROPPED_STATS != 0 {
             let nitems = buf.get_i32_le();
@@ -386,23 +397,7 @@ impl XlXactParsedRecord {
                 "XLOG_XACT_COMMIT-XACT_XINFO_HAS_DROPPED_STAT nitems {}",
                 nitems
             );
-            let sizeof_xl_xact_stats_item = 12;
-            buf.advance((nitems * sizeof_xl_xact_stats_item).try_into().unwrap());
-        }
-
-        if xinfo & pg_constants::XACT_XINFO_HAS_INVALS != 0 {
-            let nmsgs = buf.get_i32_le();
-            let sizeof_shared_invalidation_message = 16;
-            buf.advance(
-                (nmsgs * sizeof_shared_invalidation_message)
-                    .try_into()
-                    .unwrap(),
-            );
-        }
-
-        if xinfo & pg_constants::XACT_XINFO_HAS_TWOPHASE != 0 {
-            xid = buf.get_u32_le();
-            debug!("XLOG_XACT_COMMIT-XACT_XINFO_HAS_TWOPHASE xid {}", xid);
+            //FIXME: do we need to handle dropped stats here?
         }
 
         XlXactParsedRecord {

pgxn/neon/libpagestore.c

@@ -192,9 +192,8 @@ retry:
 		{
 			if (!PQconsumeInput(pageserver_conn))
 			{
-				char	   *msg = pchomp(PQerrorMessage(pageserver_conn));
-				neon_log(LOG, "could not get response from pageserver: %s", msg);
-				pfree(msg);
+				neon_log(LOG, "could not get response from pageserver: %s",
+						 PQerrorMessage(pageserver_conn));
 				return -1;
 			}
 		}
@@ -344,7 +343,7 @@ pageserver_receive(void)
 			resp = NULL;
 		}
 		else if (rc == -2)
-			neon_log(ERROR, "could not read COPY data: %s", pchomp(PQerrorMessage(pageserver_conn)));
+			neon_log(ERROR, "could not read COPY data: %s", PQerrorMessage(pageserver_conn));
 		else
 			neon_log(ERROR, "unexpected PQgetCopyData return value: %d", rc);
 	}
@@ -368,7 +367,7 @@ pageserver_flush(void)
 	}
 	else if (PQflush(pageserver_conn))
 	{
-		char	   *msg = pchomp(PQerrorMessage(pageserver_conn));
+		char	   *msg = PQerrorMessage(pageserver_conn);
 
 		pageserver_disconnect();
 		neon_log(ERROR, "failed to flush page requests: %s", msg);

poetry.lock

@@ -2092,21 +2092,21 @@ files = [
 
 [[package]]
 name = "requests"
-version = "2.31.0"
+version = "2.28.1"
 description = "Python HTTP for Humans."
 category = "main"
 optional = false
-python-versions = ">=3.7"
+python-versions = ">=3.7, <4"
 files = [
-    {file = "requests-2.31.0-py3-none-any.whl", hash = "sha256:58cd2187c01e70e6e26505bca751777aa9f2ee0b7f4300988b709f44e013003f"},
-    {file = "requests-2.31.0.tar.gz", hash = "sha256:942c5a758f98d790eaed1a29cb6eefc7ffb0d1cf7af05c3d2791656dbd6ad1e1"},
+    {file = "requests-2.28.1-py3-none-any.whl", hash = "sha256:8fefa2a1a1365bf5520aac41836fbee479da67864514bdb821f31ce07ce65349"},
+    {file = "requests-2.28.1.tar.gz", hash = "sha256:7c5599b102feddaa661c826c56ab4fee28bfd17f5abca1ebbe3e7f19d7c97983"},
 ]
 
 [package.dependencies]
 certifi = ">=2017.4.17"
-charset-normalizer = ">=2,<4"
+charset-normalizer = ">=2,<3"
 idna = ">=2.5,<4"
-urllib3 = ">=1.21.1,<3"
+urllib3 = ">=1.21.1,<1.27"
 
 [package.extras]
 socks = ["PySocks (>=1.5.6,!=1.5.7)"]
@@ -2611,4 +2611,4 @@ testing = ["func-timeout", "jaraco.itertools", "pytest (>=6)", "pytest-black (>=
 [metadata]
 lock-version = "2.0"
 python-versions = "^3.9"
-content-hash = "a0bd73376a3e9479f2379265ccec8dd6ac9df2e525909d12b77d918d590fba55"
+content-hash = "b689ffd6eae32b966f1744b5ac3343fe0dd26b31ee1f50e13daf5045ee0623e1"

proxy/Cargo.toml

@@ -35,9 +35,7 @@ postgres_backend.workspace = true
 pq_proto.workspace = true
 prometheus.workspace = true
 rand.workspace = true
-ref-cast.workspace = true
 regex.workspace = true
-redis.workspace = true
 reqwest = { workspace = true, features = ["json"] }
 reqwest-middleware.workspace = true
 reqwest-tracing.workspace = true
@@ -52,9 +50,9 @@ socket2.workspace = true
 sync_wrapper.workspace = true
 thiserror.workspace = true
 tls-listener.workspace = true
-tokio = { workspace = true, features = ["signal"] }
 tokio-postgres.workspace = true
 tokio-rustls.workspace = true
+tokio = { workspace = true, features = ["signal"] }
 tracing-opentelemetry.workspace = true
 tracing-subscriber.workspace = true
 tracing-utils.workspace = true

proxy/README.md

@@ -1,6 +1,6 @@
 # Proxy
 
-Proxy binary accepts `--auth-backend` CLI option, which determines auth scheme and cluster routing method. Following routing backends are currently implemented:
+Proxy binary accepts `--auth-backend` CLI option, which determines auth scheme and cluster routing method. Following backends are currently implemented:
 
 * console
   new SCRAM-based console API; uses SNI info to select the destination project (endpoint soon)
@@ -9,90 +9,6 @@ Proxy binary accepts `--auth-backend` CLI option, which determines auth scheme a
 * link
   sends login link for all usernames
 
-Also proxy can expose following services to the external world:
-
-* postgres protocol over TCP -- usual postgres endpoint compatible with usual
-  postgres drivers
-* postgres protocol over WebSockets -- same protocol tunneled over websockets
-  for environments where TCP connection is not available. We have our own
-  implementation of a client that uses node-postgres and tunnels traffic through
-  websockets: https://github.com/neondatabase/serverless
-* SQL over HTTP -- service that accepts POST requests with SQL text over HTTP
-  and responds with JSON-serialised results.
-
-
-## SQL over HTTP
-
-Contrary to the usual postgres proto over TCP and WebSockets using plain
-one-shot HTTP request achieves smaller amortized latencies in edge setups due to
-fewer round trips and an enhanced open connection reuse by the v8 engine. Also
-such endpoint could be used directly without any driver.
-
-To play with it locally one may start proxy over a local postgres installation
-(see end of this page on how to generate certs with openssl):
-
-```
-./target/debug/proxy -c server.crt -k server.key --auth-backend=postgres --auth-endpoint=postgres://stas@127.0.0.1:5432/stas --wss 0.0.0.0:4444
-```
-
-If both postgres and proxy are running you may send a SQL query:
-```json
-curl -k -X POST 'https://proxy.localtest.me:4444/sql' \
-  -H 'Neon-Connection-String: postgres://stas:pass@proxy.localtest.me:4444/postgres' \
-  -H 'Content-Type: application/json' \
-  --data '{
-    "query":"SELECT $1::int[] as arr, $2::jsonb as obj, 42 as num",
-    "params":[ "{{1,2},{\"3\",4}}", {"key":"val", "ikey":4242}]
-  }' | jq
-
-{
-  "command": "SELECT",
-  "fields": [
-    { "dataTypeID": 1007, "name": "arr" },
-    { "dataTypeID": 3802, "name": "obj" },
-    { "dataTypeID": 23, "name": "num" }
-  ],
-  "rowCount": 1,
-  "rows": [
-    {
-      "arr": [[1,2],[3,4]],
-      "num": 42,
-      "obj": {
-        "ikey": 4242,
-        "key": "val"
-      }
-    }
-  ]
-}
-```
-
-
-With the current approach we made the following design decisions:
-
-1. SQL injection protection: We employed the extended query protocol, modifying
-   the rust-postgres driver to send queries in one roundtrip using a text
-   protocol rather than binary, bypassing potential issues like those identified
-   in sfackler/rust-postgres#1030.
-
-2. Postgres type compatibility: As not all postgres types have binary
-   representations (e.g., acl's in pg_class), we adjusted rust-postgres to
-   respond with text protocol, simplifying serialization and fixing queries with
-   text-only types in response.
-
-3. Data type conversion: Considering JSON supports fewer data types than
-   Postgres, we perform conversions where possible, passing all other types as
-   strings. Key conversions include:
-   - postgres int2, int4, float4, float8 -> json number (NaN and Inf remain
-     text)
-   - postgres bool, null, text -> json bool, null, string
-   - postgres array -> json array
-   - postgres json and jsonb -> json object
-
-4. Alignment with node-postgres: To facilitate integration with js libraries,
-   we've matched the response structure of node-postgres, returning command tags
-   and column oids. Command tag capturing was added to the rust-postgres
-   functionality as part of this change.
-
 ## Using SNI-based routing on localhost
 
 Now proxy determines project name from the subdomain, request to the `round-rice-566201.somedomain.tld` will be routed to the project named `round-rice-566201`. Unfortunately, `/etc/hosts` does not support domain wildcards, so I usually use `*.localtest.me` which resolves to `127.0.0.1`. Now we can create self-signed certificate and play with proxy:

proxy/src/auth.rs

@@ -7,7 +7,6 @@ mod credentials;
 pub use credentials::ClientCredentials;
 
 mod password_hack;
-pub use password_hack::parse_endpoint_param;
 use password_hack::PasswordHackPayload;
 
 mod flow;
@@ -45,10 +44,10 @@ pub enum AuthErrorImpl {
     #[error(
         "Endpoint ID is not specified. \
         Either please upgrade the postgres client library (libpq) for SNI support \
-        or pass the endpoint ID (first part of the domain name) as a parameter: '?options=endpoint%3D<endpoint-id>'. \
+        or pass the endpoint ID (first part of the domain name) as a parameter: '?options=project%3D<endpoint-id>'. \
         See more at https://neon.tech/sni"
     )]
-    MissingEndpointName,
+    MissingProjectName,
 
     #[error("password authentication failed for user '{0}'")]
     AuthFailed(Box<str>),
@@ -89,7 +88,7 @@ impl UserFacingError for AuthError {
             AuthFailed(_) => self.to_string(),
             BadAuthMethod(_) => self.to_string(),
             MalformedPassword(_) => self.to_string(),
-            MissingEndpointName => self.to_string(),
+            MissingProjectName => self.to_string(),
             Io(_) => "Internal error".to_string(),
         }
     }

proxy/src/auth/backend.rs

@@ -6,7 +6,6 @@ pub use link::LinkAuthError;
 
 use crate::{
     auth::{self, ClientCredentials},
-    compute::ComputeNode,
     console::{
         self,
         provider::{CachedNodeInfo, ConsoleReqExtra},
@@ -115,7 +114,7 @@ async fn auth_quirks(
     creds: &mut ClientCredentials<'_>,
     client: &mut stream::PqStream<impl AsyncRead + AsyncWrite + Unpin>,
     allow_cleartext: bool,
-) -> auth::Result<AuthSuccess<ComputeNode>> {
+) -> auth::Result<AuthSuccess<CachedNodeInfo>> {
     // If there's no project so far, that entails that client doesn't
     // support SNI or other means of passing the endpoint (project) name.
     // We now expect to see a very specific payload in the place of password.
@@ -140,16 +139,6 @@ async fn auth_quirks(
 }
 
 impl BackendType<'_, ClientCredentials<'_>> {
-    /// Get compute endpoint name from the credentials.
-    pub fn get_endpoint(&self) -> Option<String> {
-        use BackendType::*;
-
-        match self {
-            Console(_, creds) => creds.project.clone(),
-            Postgres(_, creds) => creds.project.clone(),
-            Link(_) => Some("link".to_owned()),
-        }
-    }
     /// Authenticate the client via the requested backend, possibly using credentials.
     #[tracing::instrument(fields(allow_cleartext = allow_cleartext), skip_all)]
     pub async fn authenticate(
@@ -157,7 +146,7 @@ impl BackendType<'_, ClientCredentials<'_>> {
         extra: &ConsoleReqExtra<'_>,
         client: &mut stream::PqStream<impl AsyncRead + AsyncWrite + Unpin>,
         allow_cleartext: bool,
-    ) -> auth::Result<AuthSuccess<ComputeNode>> {
+    ) -> auth::Result<AuthSuccess<CachedNodeInfo>> {
         use BackendType::*;
 
         let res = match self {
@@ -185,7 +174,9 @@ impl BackendType<'_, ClientCredentials<'_>> {
             Link(url) => {
                 info!("performing link authentication");
 
-                link::authenticate(url, client).await?
+                link::authenticate(url, client)
+                    .await?
+                    .map(CachedNodeInfo::new_uncached)
             }
         };
 

proxy/src/auth/backend/classic.rs

@@ -1,8 +1,8 @@
 use super::AuthSuccess;
 use crate::{
     auth::{self, AuthFlow, ClientCredentials},
-    compute::{self, ComputeNode, Password},
-    console::{self, AuthInfo, CachedAuthInfo, ConsoleReqExtra},
+    compute,
+    console::{self, AuthInfo, CachedNodeInfo, ConsoleReqExtra},
     sasl, scram,
     stream::PqStream,
 };
@@ -14,26 +14,25 @@ pub(super) async fn authenticate(
     extra: &ConsoleReqExtra<'_>,
     creds: &ClientCredentials<'_>,
     client: &mut PqStream<impl AsyncRead + AsyncWrite + Unpin>,
-) -> auth::Result<AuthSuccess<ComputeNode>> {
+) -> auth::Result<AuthSuccess<CachedNodeInfo>> {
     info!("fetching user's authentication info");
     let info = api.get_auth_info(extra, creds).await?.unwrap_or_else(|| {
         // If we don't have an authentication secret, we mock one to
         // prevent malicious probing (possible due to missing protocol steps).
         // This mocked secret will never lead to successful authentication.
         info!("authentication info not found, mocking it");
-        let info = scram::ServerSecret::mock(creds.user, rand::random());
-        CachedAuthInfo::new_uncached(AuthInfo::Scram(info))
+        AuthInfo::Scram(scram::ServerSecret::mock(creds.user, rand::random()))
     });
 
     let flow = AuthFlow::new(client);
-    let keys = match &*info {
+    let scram_keys = match info {
         AuthInfo::Md5(_) => {
             info!("auth endpoint chooses MD5");
             return Err(auth::AuthError::bad_auth_method("MD5"));
         }
         AuthInfo::Scram(secret) => {
             info!("auth endpoint chooses SCRAM");
-            let scram = auth::Scram(secret);
+            let scram = auth::Scram(&secret);
             let client_key = match flow.begin(scram).await?.authenticate().await? {
                 sasl::Outcome::Success(key) => key,
                 sasl::Outcome::Failure(reason) => {
@@ -42,20 +41,21 @@ pub(super) async fn authenticate(
                 }
             };
 
-            compute::ScramKeys {
+            Some(compute::ScramKeys {
                 client_key: client_key.as_bytes(),
                 server_key: secret.server_key.as_bytes(),
-            }
+            })
         }
     };
 
-    let info = api.wake_compute(extra, creds).await?;
+    let mut node = api.wake_compute(extra, creds).await?;
+    if let Some(keys) = scram_keys {
+        use tokio_postgres::config::AuthKeys;
+        node.config.auth_keys(AuthKeys::ScramSha256(keys));
+    }
 
     Ok(AuthSuccess {
         reported_auth_ok: false,
-        value: ComputeNode::Static {
-            password: Password::ScramKeys(keys),
-            info,
-        },
+        value: node,
     })
 }

proxy/src/auth/backend/hacks.rs

@@ -1,8 +1,10 @@
 use super::AuthSuccess;
 use crate::{
     auth::{self, AuthFlow, ClientCredentials},
-    compute::{ComputeNode, Password},
-    console::{self, provider::ConsoleReqExtra},
+    console::{
+        self,
+        provider::{CachedNodeInfo, ConsoleReqExtra},
+    },
     stream,
 };
 use tokio::io::{AsyncRead, AsyncWrite};
@@ -17,7 +19,7 @@ pub async fn cleartext_hack(
     extra: &ConsoleReqExtra<'_>,
     creds: &mut ClientCredentials<'_>,
     client: &mut stream::PqStream<impl AsyncRead + AsyncWrite + Unpin>,
-) -> auth::Result<AuthSuccess<ComputeNode>> {
+) -> auth::Result<AuthSuccess<CachedNodeInfo>> {
     warn!("cleartext auth flow override is enabled, proceeding");
     let password = AuthFlow::new(client)
         .begin(auth::CleartextPassword)
@@ -25,15 +27,13 @@ pub async fn cleartext_hack(
         .authenticate()
         .await?;
 
-    let info = api.wake_compute(extra, creds).await?;
+    let mut node = api.wake_compute(extra, creds).await?;
+    node.config.password(password);
 
     // Report tentative success; compute node will check the password anyway.
     Ok(AuthSuccess {
         reported_auth_ok: false,
-        value: ComputeNode::Static {
-            password: Password::ClearText(password),
-            info,
-        },
+        value: node,
     })
 }
 
@@ -44,7 +44,7 @@ pub async fn password_hack(
     extra: &ConsoleReqExtra<'_>,
     creds: &mut ClientCredentials<'_>,
     client: &mut stream::PqStream<impl AsyncRead + AsyncWrite + Unpin>,
-) -> auth::Result<AuthSuccess<ComputeNode>> {
+) -> auth::Result<AuthSuccess<CachedNodeInfo>> {
     warn!("project not specified, resorting to the password hack auth flow");
     let payload = AuthFlow::new(client)
         .begin(auth::PasswordHack)
@@ -52,17 +52,15 @@ pub async fn password_hack(
         .authenticate()
         .await?;
 
-    info!(project = &payload.endpoint, "received missing parameter");
-    creds.project = Some(payload.endpoint);
+    info!(project = &payload.project, "received missing parameter");
+    creds.project = Some(payload.project);
 
-    let info = api.wake_compute(extra, creds).await?;
+    let mut node = api.wake_compute(extra, creds).await?;
+    node.config.password(payload.password);
 
     // Report tentative success; compute node will check the password anyway.
     Ok(AuthSuccess {
         reported_auth_ok: false,
-        value: ComputeNode::Static {
-            password: Password::ClearText(payload.password),
-            info,
-        },
+        value: node,
     })
 }

proxy/src/auth/backend/link.rs

@@ -1,10 +1,15 @@
 use super::AuthSuccess;
 use crate::{
-    auth, compute::ComputeNode, console, error::UserFacingError, stream::PqStream, waiters,
+    auth, compute,
+    console::{self, provider::NodeInfo},
+    error::UserFacingError,
+    stream::PqStream,
+    waiters,
 };
 use pq_proto::BeMessage as Be;
 use thiserror::Error;
 use tokio::io::{AsyncRead, AsyncWrite};
+use tokio_postgres::config::SslMode;
 use tracing::{info, info_span};
 
 #[derive(Debug, Error)]
@@ -52,12 +57,12 @@ pub fn new_psql_session_id() -> String {
 pub(super) async fn authenticate(
     link_uri: &reqwest::Url,
     client: &mut PqStream<impl AsyncRead + AsyncWrite + Unpin>,
-) -> auth::Result<AuthSuccess<ComputeNode>> {
+) -> auth::Result<AuthSuccess<NodeInfo>> {
     let psql_session_id = new_psql_session_id();
-    let span = info_span!("link", psql_session_id);
+    let span = info_span!("link", psql_session_id = &psql_session_id);
     let greeting = hello_message(link_uri, &psql_session_id);
 
-    let info = console::mgmt::with_waiter(psql_session_id, |waiter| async {
+    let db_info = console::mgmt::with_waiter(psql_session_id, |waiter| async {
         // Give user a URL to spawn a new database.
         info!(parent: &span, "sending the auth URL to the user");
         client
@@ -74,8 +79,35 @@ pub(super) async fn authenticate(
 
     client.write_message_noflush(&Be::NoticeResponse("Connecting to database."))?;
 
+    // This config should be self-contained, because we won't
+    // take username or dbname from client's startup message.
+    let mut config = compute::ConnCfg::new();
+    config
+        .host(&db_info.host)
+        .port(db_info.port)
+        .dbname(&db_info.dbname)
+        .user(&db_info.user);
+
+    // Backwards compatibility. pg_sni_proxy uses "--" in domain names
+    // while direct connections do not. Once we migrate to pg_sni_proxy
+    // everywhere, we can remove this.
+    if db_info.host.contains("--") {
+        // we need TLS connection with SNI info to properly route it
+        config.ssl_mode(SslMode::Require);
+    } else {
+        config.ssl_mode(SslMode::Disable);
+    }
+
+    if let Some(password) = db_info.password {
+        config.password(password.as_ref());
+    }
+
     Ok(AuthSuccess {
         reported_auth_ok: true,
-        value: ComputeNode::Link(info),
+        value: NodeInfo {
+            config,
+            aux: db_info.aux.into(),
+            allow_self_signed_compute: false, // caller may override
+        },
     })
 }

proxy/src/auth/credentials.rs

@@ -1,7 +1,6 @@
 //! User credentials used in authentication.
 
-use crate::{auth::password_hack::parse_endpoint_param, error::UserFacingError};
-use itertools::Itertools;
+use crate::error::UserFacingError;
 use pq_proto::StartupMessageParams;
 use std::collections::HashSet;
 use thiserror::Error;
@@ -62,15 +61,7 @@ impl<'a> ClientCredentials<'a> {
         // Project name might be passed via PG's command-line options.
         let project_option = params
             .options_raw()
-            .and_then(|options| {
-                // We support both `project` (deprecated) and `endpoint` options for backward compatibility.
-                // However, if both are present, we don't exactly know which one to use.
-                // Therefore we require that only one of them is present.
-                options
-                    .filter_map(parse_endpoint_param)
-                    .at_most_one()
-                    .ok()?
-            })
+            .and_then(|mut options| options.find_map(|opt| opt.strip_prefix("project=")))
             .map(|name| name.to_string());
 
         let project_from_domain = if let Some(sni_str) = sni {
@@ -186,51 +177,6 @@ mod tests {
         Ok(())
     }
 
-    #[test]
-    fn parse_endpoint_from_options() -> anyhow::Result<()> {
-        let options = StartupMessageParams::new([
-            ("user", "john_doe"),
-            ("options", "-ckey=1 endpoint=bar -c geqo=off"),
-        ]);
-
-        let creds = ClientCredentials::parse(&options, None, None)?;
-        assert_eq!(creds.user, "john_doe");
-        assert_eq!(creds.project.as_deref(), Some("bar"));
-
-        Ok(())
-    }
-
-    #[test]
-    fn parse_three_endpoints_from_options() -> anyhow::Result<()> {
-        let options = StartupMessageParams::new([
-            ("user", "john_doe"),
-            (
-                "options",
-                "-ckey=1 endpoint=one endpoint=two endpoint=three -c geqo=off",
-            ),
-        ]);
-
-        let creds = ClientCredentials::parse(&options, None, None)?;
-        assert_eq!(creds.user, "john_doe");
-        assert!(creds.project.is_none());
-
-        Ok(())
-    }
-
-    #[test]
-    fn parse_when_endpoint_and_project_are_in_options() -> anyhow::Result<()> {
-        let options = StartupMessageParams::new([
-            ("user", "john_doe"),
-            ("options", "-ckey=1 endpoint=bar project=foo -c geqo=off"),
-        ]);
-
-        let creds = ClientCredentials::parse(&options, None, None)?;
-        assert_eq!(creds.user, "john_doe");
-        assert!(creds.project.is_none());
-
-        Ok(())
-    }
-
     #[test]
     fn parse_projects_identical() -> anyhow::Result<()> {
         let options = StartupMessageParams::new([("user", "john_doe"), ("options", "project=baz")]);

proxy/src/auth/flow.rs

@@ -91,7 +91,7 @@ impl<S: AsyncRead + AsyncWrite + Unpin> AuthFlow<'_, S, PasswordHack> {
             // the user neither enabled SNI nor resorted to any other method
             // for passing the project name we rely on. We should show them
             // the most helpful error message and point to the documentation.
-            .ok_or(AuthErrorImpl::MissingEndpointName)?;
+            .ok_or(AuthErrorImpl::MissingProjectName)?;
 
         Ok(payload)
     }

proxy/src/auth/password_hack.rs

@@ -6,55 +6,27 @@
 use bstr::ByteSlice;
 
 pub struct PasswordHackPayload {
-    pub endpoint: String,
+    pub project: String,
     pub password: Vec<u8>,
 }
 
 impl PasswordHackPayload {
     pub fn parse(bytes: &[u8]) -> Option<Self> {
         // The format is `project=<utf-8>;<password-bytes>`.
-        let mut iter = bytes.splitn_str(2, ";");
-        let endpoint = iter.next()?.to_str().ok()?;
-        let endpoint = parse_endpoint_param(endpoint)?.to_owned();
+        let mut iter = bytes.strip_prefix(b"project=")?.splitn_str(2, ";");
+        let project = iter.next()?.to_str().ok()?.to_owned();
         let password = iter.next()?.to_owned();
 
-        Some(Self { endpoint, password })
+        Some(Self { project, password })
     }
 }
 
-pub fn parse_endpoint_param(bytes: &str) -> Option<&str> {
-    bytes
-        .strip_prefix("project=")
-        .or_else(|| bytes.strip_prefix("endpoint="))
-}
-
 #[cfg(test)]
 mod tests {
     use super::*;
 
     #[test]
-    fn parse_endpoint_param_fn() {
-        let input = "";
-        assert!(parse_endpoint_param(input).is_none());
-
-        let input = "project=";
-        assert_eq!(parse_endpoint_param(input), Some(""));
-
-        let input = "project=foobar";
-        assert_eq!(parse_endpoint_param(input), Some("foobar"));
-
-        let input = "endpoint=";
-        assert_eq!(parse_endpoint_param(input), Some(""));
-
-        let input = "endpoint=foobar";
-        assert_eq!(parse_endpoint_param(input), Some("foobar"));
-
-        let input = "other_option=foobar";
-        assert!(parse_endpoint_param(input).is_none());
-    }
-
-    #[test]
-    fn parse_password_hack_payload_project() {
+    fn parse_password_hack_payload() {
         let bytes = b"";
         assert!(PasswordHackPayload::parse(bytes).is_none());
 
@@ -62,33 +34,13 @@ mod tests {
         assert!(PasswordHackPayload::parse(bytes).is_none());
 
         let bytes = b"project=;";
-        let payload: PasswordHackPayload =
-            PasswordHackPayload::parse(bytes).expect("parsing failed");
-        assert_eq!(payload.endpoint, "");
+        let payload = PasswordHackPayload::parse(bytes).expect("parsing failed");
+        assert_eq!(payload.project, "");
         assert_eq!(payload.password, b"");
 
         let bytes = b"project=foobar;pass;word";
         let payload = PasswordHackPayload::parse(bytes).expect("parsing failed");
-        assert_eq!(payload.endpoint, "foobar");
-        assert_eq!(payload.password, b"pass;word");
-    }
-
-    #[test]
-    fn parse_password_hack_payload_endpoint() {
-        let bytes = b"";
-        assert!(PasswordHackPayload::parse(bytes).is_none());
-
-        let bytes = b"endpoint=";
-        assert!(PasswordHackPayload::parse(bytes).is_none());
-
-        let bytes = b"endpoint=;";
-        let payload = PasswordHackPayload::parse(bytes).expect("parsing failed");
-        assert_eq!(payload.endpoint, "");
-        assert_eq!(payload.password, b"");
-
-        let bytes = b"endpoint=foobar;pass;word";
-        let payload = PasswordHackPayload::parse(bytes).expect("parsing failed");
-        assert_eq!(payload.endpoint, "foobar");
+        assert_eq!(payload.project, "foobar");
         assert_eq!(payload.password, b"pass;word");
     }
 }

proxy/src/bin/proxy.rs

@@ -1,15 +1,16 @@
-use anyhow::{bail, Context};
+use proxy::auth;
+use proxy::console;
+use proxy::http;
+use proxy::metrics;
+
+use anyhow::bail;
 use clap::{self, Arg};
-use futures::future::try_join_all;
-use proxy::{
-    auth,
-    config::{self, MetricCollectionConfig, ProxyConfig, TlsConfig},
-    console, http, metrics,
-};
-use std::{borrow::Cow, net::SocketAddr, sync::atomic::Ordering};
+use proxy::config::{self, ProxyConfig};
+use std::{borrow::Cow, net::SocketAddr};
 use tokio::net::TcpListener;
 use tokio_util::sync::CancellationToken;
-use tracing::{info, warn};
+use tracing::info;
+use tracing::warn;
 use utils::{project_git_version, sentry_init::init_sentry};
 
 project_git_version!(GIT_VERSION);
@@ -24,7 +25,8 @@ async fn main() -> anyhow::Result<()> {
     ::metrics::set_build_info_metric(GIT_VERSION);
 
     let args = cli().get_matches();
-    let config: &ProxyConfig = Box::leak(Box::new(build_config(&args)?));
+    let config = build_config(&args)?;
+
     info!("Authentication backend: {}", config.auth_backend);
 
     // Check that we can bind to address before further initialization
@@ -69,29 +71,20 @@ async fn main() -> anyhow::Result<()> {
         tasks.push(tokio::spawn(metrics::task_main(metrics_config)));
     }
 
-    if let auth::BackendType::Console(api, _) = &config.auth_backend {
-        if let Some(url) = args.get_one::<String>("redis-notifications") {
-            info!("Starting redis notifications listener ({url})");
-            tasks.push(tokio::spawn(console::notifications::task_main(
-                url.to_owned(),
-                api.caches,
-            )));
-        }
-    }
-
-    let tasks = try_join_all(tasks.into_iter().map(proxy::flatten_err));
-    let client_tasks = try_join_all(client_tasks.into_iter().map(proxy::flatten_err));
+    let tasks = futures::future::try_join_all(tasks.into_iter().map(proxy::flatten_err));
+    let client_tasks =
+        futures::future::try_join_all(client_tasks.into_iter().map(proxy::flatten_err));
     tokio::select! {
         // We are only expecting an error from these forever tasks
         res = tasks => { res?; },
         res = client_tasks => { res?; },
     }
-
     Ok(())
 }
 
-fn build_tls_config(args: &clap::ArgMatches) -> anyhow::Result<Option<TlsConfig>> {
-    let config = match (
+/// 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"),
     ) {
@@ -108,22 +101,16 @@ fn build_tls_config(args: &clap::ArgMatches) -> anyhow::Result<Option<TlsConfig>
         .get_one::<String>("allow-self-signed-compute")
         .unwrap()
         .parse()?;
-
     if allow_self_signed_compute {
         warn!("allowing self-signed compute certificates");
-        proxy::compute::ALLOW_SELF_SIGNED_COMPUTE.store(true, Ordering::Relaxed);
     }
 
-    Ok(config)
-}
-
-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) {
+    let metric_collection = match (
+        args.get_one::<String>("metric-collection-endpoint"),
+        args.get_one::<String>("metric-collection-interval"),
+    ) {
         (Some(endpoint), Some(interval)) => Some(config::MetricCollectionConfig {
-            endpoint: endpoint.parse().context("bad metrics endpoint")?,
+            endpoint: endpoint.parse()?,
             interval: humantime::parse_duration(interval)?,
         }),
         (None, None) => None,
@@ -133,40 +120,21 @@ fn build_metrics_config(args: &clap::ArgMatches) -> anyhow::Result<Option<Metric
         ),
     };
 
-    Ok(config)
-}
-
-fn make_caches(args: &clap::ArgMatches) -> anyhow::Result<console::caches::ApiCaches> {
-    let config::CacheOptions { size, ttl } = args
-        .get_one::<String>("get-auth-info-cache")
-        .unwrap()
-        .parse()?;
-
-    info!("Using AuthInfoCache (get_auth_info) with size={size} ttl={ttl:?}");
-    let auth_info = console::caches::AuthInfoCache::new("auth_info_cache", size, ttl);
-
-    let config::CacheOptions { size, ttl } = args
-        .get_one::<String>("wake-compute-cache")
-        .unwrap()
-        .parse()?;
-
-    info!("Using NodeInfoCache (wake_compute) with size={size} ttl={ttl:?}");
-    let node_info = console::caches::NodeInfoCache::new("node_info_cache", size, ttl);
-
-    let caches = console::caches::ApiCaches {
-        auth_info,
-        node_info,
-    };
-
-    Ok(caches)
-}
-
-fn build_auth_config(args: &clap::ArgMatches) -> anyhow::Result<auth::BackendType<'static, ()>> {
-    let config = match args.get_one::<String>("auth-backend").unwrap().as_str() {
+    let auth_backend = match args.get_one::<String>("auth-backend").unwrap().as_str() {
         "console" => {
+            let config::CacheOptions { size, ttl } = args
+                .get_one::<String>("wake-compute-cache")
+                .unwrap()
+                .parse()?;
+
+            info!("Using NodeInfoCache (wake_compute) with size={size} ttl={ttl:?}");
+            let caches = Box::leak(Box::new(console::caches::ApiCaches {
+                node_info: console::caches::NodeInfoCache::new("node_info_cache", size, ttl),
+            }));
+
             let url = args.get_one::<String>("auth-endpoint").unwrap().parse()?;
             let endpoint = http::Endpoint::new(url, http::new_client());
-            let caches = Box::leak(Box::new(make_caches(args)?));
+
             let api = console::provider::neon::Api::new(endpoint, caches);
             auth::BackendType::Console(Cow::Owned(api), ())
         }
@@ -182,16 +150,12 @@ fn build_auth_config(args: &clap::ArgMatches) -> anyhow::Result<auth::BackendTyp
         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<ProxyConfig> {
-    let config = ProxyConfig {
-        tls_config: build_tls_config(args)?,
-        auth_backend: build_auth_config(args)?,
-        metric_collection: build_metrics_config(args)?,
-    };
+    let config = Box::leak(Box::new(ProxyConfig {
+        tls_config,
+        auth_backend,
+        metric_collection,
+        allow_self_signed_compute,
+    }));
 
     Ok(config)
 }
@@ -275,22 +239,11 @@ fn cli() -> clap::Command {
                 .long("metric-collection-interval")
                 .help("how often metrics should be sent to a collection endpoint"),
         )
-        .arg(
-            Arg::new("redis-notifications")
-                .long("redis-notifications")
-                .help("for receiving notifications from console (e.g. redis://127.0.0.1:6379)"),
-        )
-        .arg(
-            Arg::new("get-auth-info-cache")
-                .long("get-auth-info-cache")
-                .help("cache for `get_auth_info` api method (use `size=0` to disable)")
-                .default_value(config::CacheOptions::DEFAULT_AUTH_INFO),
-        )
         .arg(
             Arg::new("wake-compute-cache")
                 .long("wake-compute-cache")
                 .help("cache for `wake_compute` api method (use `size=0` to disable)")
-                .default_value(config::CacheOptions::DEFAULT_NODE_INFO),
+                .default_value(config::CacheOptions::DEFAULT_OPTIONS_NODE_INFO),
         )
         .arg(
             Arg::new("allow-self-signed-compute")

proxy/src/cache.rs

@@ -1,117 +1,304 @@
-use std::{any::Any, sync::Arc, time::Instant};
+use std::{
+    borrow::Borrow,
+    hash::Hash,
+    ops::{Deref, DerefMut},
+    time::{Duration, Instant},
+};
+use tracing::debug;
 
-/// A variant of LRU where every entry has a TTL.
-pub mod timed_lru;
-pub use timed_lru::TimedLru;
+// This seems to make more sense than `lru` or `cached`:
+//
+// * `near/nearcore` ditched `cached` in favor of `lru`
+//   (https://github.com/near/nearcore/issues?q=is%3Aissue+lru+is%3Aclosed).
+//
+// * `lru` methods use an obscure `KeyRef` type in their contraints (which is deliberately excluded from docs).
+//   This severely hinders its usage both in terms of creating wrappers and supported key types.
+//
+// On the other hand, `hashlink` has good download stats and appears to be maintained.
+use hashlink::{linked_hash_map::RawEntryMut, LruCache};
 
-/// Useful type aliases.
-pub mod types {
-    pub type Cached<T> = super::Cached<'static, T>;
-}
+/// A generic trait which exposes types of cache's key and value,
+/// as well as the notion of cache entry invalidation.
+/// This is useful for [`timed_lru::Cached`].
+pub trait Cache {
+    /// Entry's key.
+    type Key;
 
-/// Lookup information for cache entry invalidation.
-#[derive(Clone)]
-pub struct LookupInfo<K> {
-    /// Cache entry creation time.
-    /// We use this during invalidation lookups to prevent eviction of a newer
-    /// entry sharing the same key (it might've been inserted by a different
-    /// task after we got the entry we're trying to invalidate now).
-    created_at: Instant,
+    /// Entry's value.
+    type Value;
 
-    /// Search by this key.
-    key: K,
-}
+    /// Used for entry invalidation.
+    type LookupInfo<Key>;
 
-/// This type incapsulates everything needed for cache entry invalidation.
-/// For convenience, we completely erase the types of a cache ref and a key.
-/// This lets us store multiple tokens in a homogeneous collection (e.g. Vec).
-#[derive(Clone)]
-pub struct InvalidationToken<'a> {
-    // TODO: allow more than one type of references (e.g. Arc) if it's ever needed.
-    cache: &'a (dyn Cache + Sync + Send),
-    info: LookupInfo<Arc<dyn Any + Sync + Send>>,
-}
-
-impl InvalidationToken<'_> {
-    /// Invalidate a corresponding cache entry.
-    pub fn invalidate(&self) {
-        let info = LookupInfo {
-            created_at: self.info.created_at,
-            key: self.info.key.as_ref(),
-        };
-        self.cache.invalidate_entry(info);
-    }
-}
-
-/// A combination of a cache entry and its invalidation token.
-/// Makes it easier to see how those two are connected.
-#[derive(Clone)]
-pub struct Cached<'a, T> {
-    pub token: Option<InvalidationToken<'a>>,
-    pub value: Arc<T>,
-}
-
-impl<T> Cached<'_, T> {
-    /// Place any entry into this wrapper; invalidation will be a no-op.
-    pub fn new_uncached(value: T) -> Self {
-        Self {
-            token: None,
-            value: value.into(),
-        }
-    }
-
-    /// Invalidate a corresponding cache entry.
-    pub fn invalidate(&self) {
-        if let Some(token) = &self.token {
-            token.invalidate();
-        }
-    }
-}
-
-impl<T> std::ops::Deref for Cached<'_, T> {
-    type Target = T;
-
-    fn deref(&self) -> &Self::Target {
-        &self.value
-    }
-}
-
-/// This trait captures the notion of cache entry invalidation.
-/// It doesn't have any associated types because we use dyn-based type erasure.
-trait Cache {
     /// Invalidate an entry using a lookup info.
     /// We don't have an empty default impl because it's error-prone.
-    fn invalidate_entry(&self, info: LookupInfo<&(dyn Any + Send + Sync)>);
+    fn invalidate(&self, _: &Self::LookupInfo<Self::Key>);
 }
 
-#[cfg(test)]
-mod tests {
+impl<C: Cache> Cache for &C {
+    type Key = C::Key;
+    type Value = C::Value;
+    type LookupInfo<Key> = C::LookupInfo<Key>;
+
+    fn invalidate(&self, info: &Self::LookupInfo<Self::Key>) {
+        C::invalidate(self, info)
+    }
+}
+
+pub use timed_lru::TimedLru;
+pub mod timed_lru {
     use super::*;
 
-    #[test]
-    fn trivial_properties_of_cached() {
-        let cached = Cached::new_uncached(0);
-        assert_eq!(*cached, 0);
-        cached.invalidate();
+    /// An implementation of timed LRU cache with fixed capacity.
+    /// Key properties:
+    ///
+    /// * Whenever a new entry is inserted, the least recently accessed one is evicted.
+    ///   The cache also keeps track of entry's insertion time (`created_at`) and TTL (`expires_at`).
+    ///
+    /// * When the entry is about to be retrieved, we check its expiration timestamp.
+    ///   If the entry has expired, we remove it from the cache; Otherwise we bump the
+    ///   expiration timestamp (e.g. +5mins) and change its place in LRU list to prolong
+    ///   its existence.
+    ///
+    /// * There's an API for immediate invalidation (removal) of a cache entry;
+    ///   It's useful in case we know for sure that the entry is no longer correct.
+    ///   See [`timed_lru::LookupInfo`] & [`timed_lru::Cached`] for more information.
+    ///
+    /// * Expired entries are kept in the cache, until they are evicted by the LRU policy,
+    ///   or by a successful lookup (i.e. the entry hasn't expired yet).
+    ///   There is no background job to reap the expired records.
+    ///
+    /// * It's possible for an entry that has not yet expired entry to be evicted
+    ///   before expired items. That's a bit wasteful, but probably fine in practice.
+    pub struct TimedLru<K, V> {
+        /// Cache's name for tracing.
+        name: &'static str,
+
+        /// The underlying cache implementation.
+        cache: parking_lot::Mutex<LruCache<K, Entry<V>>>,
+
+        /// Default time-to-live of a single entry.
+        ttl: Duration,
     }
 
-    #[test]
-    fn invalidation_token_type_erasure() {
-        let lifetime = std::time::Duration::from_secs(10);
-        let foo = TimedLru::<u32, u32>::new("foo", 128, lifetime);
-        let bar = TimedLru::<String, usize>::new("bar", 128, lifetime);
+    impl<K: Hash + Eq, V> Cache for TimedLru<K, V> {
+        type Key = K;
+        type Value = V;
+        type LookupInfo<Key> = LookupInfo<Key>;
 
-        let (_, x) = foo.insert(100.into(), 0.into());
-        let (_, y) = bar.insert(String::new().into(), 404.into());
+        fn invalidate(&self, info: &Self::LookupInfo<K>) {
+            self.invalidate_raw(info)
+        }
+    }
 
-        // Invalidation tokens should be cloneable and homogeneous (same type).
-        let tokens = [x.token.clone().unwrap(), y.token.clone().unwrap()];
-        for token in tokens {
-            token.invalidate();
+    struct Entry<T> {
+        created_at: Instant,
+        expires_at: Instant,
+        value: T,
+    }
+
+    impl<K: Hash + Eq, V> TimedLru<K, V> {
+        /// Construct a new LRU cache with timed entries.
+        pub fn new(name: &'static str, capacity: usize, ttl: Duration) -> Self {
+            Self {
+                name,
+                cache: LruCache::new(capacity).into(),
+                ttl,
+            }
         }
 
-        // Values are still there.
-        assert_eq!(*x, 0);
-        assert_eq!(*y, 404);
+        /// Drop an entry from the cache if it's outdated.
+        #[tracing::instrument(level = "debug", fields(cache = self.name), skip_all)]
+        fn invalidate_raw(&self, info: &LookupInfo<K>) {
+            let now = Instant::now();
+
+            // Do costly things before taking the lock.
+            let mut cache = self.cache.lock();
+            let raw_entry = match cache.raw_entry_mut().from_key(&info.key) {
+                RawEntryMut::Vacant(_) => return,
+                RawEntryMut::Occupied(x) => x,
+            };
+
+            // Remove the entry if it was created prior to lookup timestamp.
+            let entry = raw_entry.get();
+            let (created_at, expires_at) = (entry.created_at, entry.expires_at);
+            let should_remove = created_at <= info.created_at || expires_at <= now;
+
+            if should_remove {
+                raw_entry.remove();
+            }
+
+            drop(cache); // drop lock before logging
+            debug!(
+                created_at = format_args!("{created_at:?}"),
+                expires_at = format_args!("{expires_at:?}"),
+                entry_removed = should_remove,
+                "processed a cache entry invalidation event"
+            );
+        }
+
+        /// Try retrieving an entry by its key, then execute `extract` if it exists.
+        #[tracing::instrument(level = "debug", fields(cache = self.name), skip_all)]
+        fn get_raw<Q, R>(&self, key: &Q, extract: impl FnOnce(&K, &Entry<V>) -> R) -> Option<R>
+        where
+            K: Borrow<Q>,
+            Q: Hash + Eq + ?Sized,
+        {
+            let now = Instant::now();
+            let deadline = now.checked_add(self.ttl).expect("time overflow");
+
+            // Do costly things before taking the lock.
+            let mut cache = self.cache.lock();
+            let mut raw_entry = match cache.raw_entry_mut().from_key(key) {
+                RawEntryMut::Vacant(_) => return None,
+                RawEntryMut::Occupied(x) => x,
+            };
+
+            // Immeditely drop the entry if it has expired.
+            let entry = raw_entry.get();
+            if entry.expires_at <= now {
+                raw_entry.remove();
+                return None;
+            }
+
+            let value = extract(raw_entry.key(), entry);
+            let (created_at, expires_at) = (entry.created_at, entry.expires_at);
+
+            // Update the deadline and the entry's position in the LRU list.
+            raw_entry.get_mut().expires_at = deadline;
+            raw_entry.to_back();
+
+            drop(cache); // drop lock before logging
+            debug!(
+                created_at = format_args!("{created_at:?}"),
+                old_expires_at = format_args!("{expires_at:?}"),
+                new_expires_at = format_args!("{deadline:?}"),
+                "accessed a cache entry"
+            );
+
+            Some(value)
+        }
+
+        /// Insert an entry to the cache. If an entry with the same key already
+        /// existed, return the previous value and its creation timestamp.
+        #[tracing::instrument(level = "debug", fields(cache = self.name), skip_all)]
+        fn insert_raw(&self, key: K, value: V) -> (Instant, Option<V>) {
+            let created_at = Instant::now();
+            let expires_at = created_at.checked_add(self.ttl).expect("time overflow");
+
+            let entry = Entry {
+                created_at,
+                expires_at,
+                value,
+            };
+
+            // Do costly things before taking the lock.
+            let old = self
+                .cache
+                .lock()
+                .insert(key, entry)
+                .map(|entry| entry.value);
+
+            debug!(
+                created_at = format_args!("{created_at:?}"),
+                expires_at = format_args!("{expires_at:?}"),
+                replaced = old.is_some(),
+                "created a cache entry"
+            );
+
+            (created_at, old)
+        }
+    }
+
+    impl<K: Hash + Eq + Clone, V: Clone> TimedLru<K, V> {
+        pub fn insert(&self, key: K, value: V) -> (Option<V>, Cached<&Self>) {
+            let (created_at, old) = self.insert_raw(key.clone(), value.clone());
+
+            let cached = Cached {
+                token: Some((self, LookupInfo { created_at, key })),
+                value,
+            };
+
+            (old, cached)
+        }
+    }
+
+    impl<K: Hash + Eq, V: Clone> TimedLru<K, V> {
+        /// Retrieve a cached entry in convenient wrapper.
+        pub fn get<Q>(&self, key: &Q) -> Option<timed_lru::Cached<&Self>>
+        where
+            K: Borrow<Q> + Clone,
+            Q: Hash + Eq + ?Sized,
+        {
+            self.get_raw(key, |key, entry| {
+                let info = LookupInfo {
+                    created_at: entry.created_at,
+                    key: key.clone(),
+                };
+
+                Cached {
+                    token: Some((self, info)),
+                    value: entry.value.clone(),
+                }
+            })
+        }
+    }
+
+    /// Lookup information for key invalidation.
+    pub struct LookupInfo<K> {
+        /// Time of creation of a cache [`Entry`].
+        /// We use this during invalidation lookups to prevent eviction of a newer
+        /// entry sharing the same key (it might've been inserted by a different
+        /// task after we got the entry we're trying to invalidate now).
+        created_at: Instant,
+
+        /// Search by this key.
+        key: K,
+    }
+
+    /// Wrapper for convenient entry invalidation.
+    pub struct Cached<C: Cache> {
+        /// Cache + lookup info.
+        token: Option<(C, C::LookupInfo<C::Key>)>,
+
+        /// The value itself.
+        pub value: C::Value,
+    }
+
+    impl<C: Cache> Cached<C> {
+        /// Place any entry into this wrapper; invalidation will be a no-op.
+        /// Unfortunately, rust doesn't let us implement [`From`] or [`Into`].
+        pub fn new_uncached(value: impl Into<C::Value>) -> Self {
+            Self {
+                token: None,
+                value: value.into(),
+            }
+        }
+
+        /// Drop this entry from a cache if it's still there.
+        pub fn invalidate(&self) {
+            if let Some((cache, info)) = &self.token {
+                cache.invalidate(info);
+            }
+        }
+
+        /// Tell if this entry is actually cached.
+        pub fn cached(&self) -> bool {
+            self.token.is_some()
+        }
+    }
+
+    impl<C: Cache> Deref for Cached<C> {
+        type Target = C::Value;
+
+        fn deref(&self) -> &Self::Target {
+            &self.value
+        }
+    }
+
+    impl<C: Cache> DerefMut for Cached<C> {
+        fn deref_mut(&mut self) -> &mut Self::Target {
+            &mut self.value
+        }
     }
 }

proxy/src/cache/timed_lru.rs

@@ -1,293 +0,0 @@
-#[cfg(test)]
-mod tests;
-
-use super::{Cache, Cached, InvalidationToken, LookupInfo};
-use ref_cast::RefCast;
-use std::{
-    any::Any,
-    borrow::Borrow,
-    hash::Hash,
-    sync::Arc,
-    time::{Duration, Instant},
-};
-use tracing::debug;
-
-// This seems to make more sense than `lru` or `cached`:
-//
-// * `near/nearcore` ditched `cached` in favor of `lru`
-//   (https://github.com/near/nearcore/issues?q=is%3Aissue+lru+is%3Aclosed).
-//
-// * `lru` methods use an obscure `KeyRef` type in their contraints (which is deliberately excluded from docs).
-//   This severely hinders its usage both in terms of creating wrappers and supported key types.
-//
-// On the other hand, `hashlink` has good download stats and appears to be maintained.
-use hashlink::{linked_hash_map::RawEntryMut, LruCache};
-
-/// An implementation of timed LRU cache with fixed capacity.
-/// Key properties:
-///
-/// * Whenever a new entry is inserted, the least recently accessed one is evicted.
-///   The cache also keeps track of entry's insertion time (`created_at`) and TTL (`expires_at`).
-///
-/// * When the entry is about to be retrieved, we check its expiration timestamp.
-///   If the entry has expired, we remove it from the cache; Otherwise we bump the
-///   expiration timestamp (e.g. +5mins) and change its place in LRU list to prolong
-///   its existence.
-///
-/// * There's an API for immediate invalidation (removal) of a cache entry;
-///   It's useful in case we know for sure that the entry is no longer correct.
-///   See [`timed_lru::LookupInfo`] & [`timed_lru::Cached`] for more information.
-///
-/// * Expired entries are kept in the cache, until they are evicted by the LRU policy,
-///   or by a successful lookup (i.e. the entry hasn't expired yet).
-///   There is no background job to reap the expired records.
-///
-/// * It's possible for an entry that has not yet expired entry to be evicted
-///   before expired items. That's a bit wasteful, but probably fine in practice.
-pub struct TimedLru<K, V> {
-    /// Cache's name for tracing.
-    name: &'static str,
-
-    /// The underlying cache implementation.
-    cache: parking_lot::Mutex<LruCache<Key<K>, Entry<V>>>,
-
-    /// Default time-to-live of a single entry.
-    ttl: Duration,
-}
-
-#[derive(RefCast, Hash, PartialEq, Eq)]
-#[repr(transparent)]
-struct Query<Q: ?Sized>(Q);
-
-#[derive(Hash, PartialEq, Eq)]
-#[repr(transparent)]
-struct Key<T: ?Sized>(Arc<T>);
-
-/// It's impossible to implement this without [`Key`] & [`Query`]:
-/// * We can't implement std traits for [`Arc`].
-/// * Even if we could, it'd conflict with `impl<T> Borrow<T> for T`.
-impl<Q, T> Borrow<Query<Q>> for Key<T>
-where
-    Q: ?Sized,
-    T: Borrow<Q>,
-{
-    #[inline(always)]
-    fn borrow(&self) -> &Query<Q> {
-        RefCast::ref_cast(self.0.as_ref().borrow())
-    }
-}
-
-struct Entry<T> {
-    created_at: Instant,
-    expires_at: Instant,
-    value: Arc<T>,
-}
-
-impl<K: Hash + Eq, V> TimedLru<K, V> {
-    /// Construct a new LRU cache with timed entries.
-    pub fn new(name: &'static str, capacity: usize, ttl: Duration) -> Self {
-        Self {
-            name,
-            cache: LruCache::new(capacity).into(),
-            ttl,
-        }
-    }
-
-    /// Get the number of entries in the cache.
-    /// Note that this method will not try to evict stale entries.
-    pub fn size(&self) -> usize {
-        self.cache.lock().len()
-    }
-
-    /// Try retrieving an entry by its key, then execute `extract` if it exists.
-    #[tracing::instrument(level = "debug", fields(cache = self.name), skip_all)]
-    fn get_raw<Q, F, R>(&self, key: &Q, extract: F) -> Option<R>
-    where
-        K: Borrow<Q>,
-        Q: Hash + Eq + ?Sized,
-        F: FnOnce(&Arc<K>, &Entry<V>) -> R,
-    {
-        let key: &Query<Q> = RefCast::ref_cast(key);
-        let now = Instant::now();
-        let deadline = now.checked_add(self.ttl).expect("time overflow");
-
-        // Do costly things before taking the lock.
-        let mut cache = self.cache.lock();
-        let mut raw_entry = match cache.raw_entry_mut().from_key(key) {
-            RawEntryMut::Vacant(_) => return None,
-            RawEntryMut::Occupied(x) => x,
-        };
-
-        // Immeditely drop the entry if it has expired.
-        let entry = raw_entry.get();
-        if entry.expires_at <= now {
-            raw_entry.remove();
-            return None;
-        }
-
-        let value = extract(&raw_entry.key().0, entry);
-        let (created_at, expires_at) = (entry.created_at, entry.expires_at);
-
-        // Update the deadline and the entry's position in the LRU list.
-        raw_entry.get_mut().expires_at = deadline;
-        raw_entry.to_back();
-
-        drop(cache); // drop lock before logging
-        debug!(
-            ?created_at,
-            old_expires_at = ?expires_at,
-            new_expires_at = ?deadline,
-            "accessed a cache entry"
-        );
-
-        Some(value)
-    }
-
-    /// Insert an entry to the cache. If an entry with the same key already
-    /// existed, return the previous value and its creation timestamp.
-    #[tracing::instrument(level = "debug", fields(cache = self.name), skip_all)]
-    fn insert_raw(&self, key: Arc<K>, value: Arc<V>) -> (Option<Arc<V>>, Instant) {
-        let created_at = Instant::now();
-        let expires_at = created_at.checked_add(self.ttl).expect("time overflow");
-
-        let entry = Entry {
-            created_at,
-            expires_at,
-            value,
-        };
-
-        // Do costly things before taking the lock.
-        let old = self
-            .cache
-            .lock()
-            .insert(Key(key), entry)
-            .map(|entry| entry.value);
-
-        debug!(
-            ?created_at,
-            ?expires_at,
-            replaced = old.is_some(),
-            "created a cache entry"
-        );
-
-        (old, created_at)
-    }
-
-    #[tracing::instrument(level = "debug", fields(cache = self.name), skip_all)]
-    fn remove_raw<Q>(&self, key: &Q)
-    where
-        K: Borrow<Q>,
-        Q: Hash + Eq + ?Sized,
-    {
-        let key: &Query<Q> = RefCast::ref_cast(key);
-        let mut cache = self.cache.lock();
-        cache.remove(key);
-
-        debug!("removed a cache entry");
-    }
-}
-
-/// Convenient wrappers for raw methods.
-impl<K: Hash + Eq + Sync + Send + 'static, V> TimedLru<K, V>
-where
-    Self: Sync + Send,
-{
-    pub fn get<Q>(&self, key: &Q) -> Option<Cached<V>>
-    where
-        K: Borrow<Q>,
-        Q: Hash + Eq + ?Sized,
-    {
-        self.get_raw(key, |key, entry| {
-            let info = LookupInfo {
-                created_at: entry.created_at,
-                key: Arc::clone(key),
-            };
-
-            Cached {
-                token: Some(Self::invalidation_token(self, info)),
-                value: entry.value.clone(),
-            }
-        })
-    }
-
-    pub fn insert(&self, key: Arc<K>, value: Arc<V>) -> (Option<Arc<V>>, Cached<V>) {
-        let (old, created_at) = self.insert_raw(key.clone(), value.clone());
-
-        let info = LookupInfo { created_at, key };
-        let cached = Cached {
-            token: Some(Self::invalidation_token(self, info)),
-            value,
-        };
-
-        (old, cached)
-    }
-
-    pub fn remove<Q>(&self, key: &Q)
-    where
-        K: Borrow<Q>,
-        Q: Hash + Eq + ?Sized,
-    {
-        self.remove_raw(key)
-    }
-}
-
-/// Implementation details of the entry invalidation machinery.
-impl<K: Hash + Eq + Sync + Send + 'static, V> TimedLru<K, V>
-where
-    Self: Sync + Send,
-{
-    /// This is a proper (safe) way to create an invalidation token for [`TimedLru`].
-    fn invalidation_token(&self, info: LookupInfo<Arc<K>>) -> InvalidationToken<'_> {
-        InvalidationToken {
-            cache: self,
-            info: LookupInfo {
-                created_at: info.created_at,
-                key: info.key,
-            },
-        }
-    }
-}
-
-/// This implementation depends on [`Self::invalidation_token`].
-impl<K: Hash + Eq + 'static, V> Cache for TimedLru<K, V> {
-    fn invalidate_entry(&self, info: LookupInfo<&(dyn Any + Sync + Send)>) {
-        let info = LookupInfo {
-            created_at: info.created_at,
-            // NOTE: it's important to downcast to the correct type!
-            key: info.key.downcast_ref::<K>().expect("bad key type"),
-        };
-        self.invalidate_raw(info)
-    }
-}
-
-impl<K: Hash + Eq, V> TimedLru<K, V> {
-    #[tracing::instrument(level = "debug", fields(cache = self.name), skip_all)]
-    fn invalidate_raw(&self, info: LookupInfo<&K>) {
-        let key: &Query<K> = RefCast::ref_cast(info.key);
-        let now = Instant::now();
-
-        // Do costly things before taking the lock.
-        let mut cache = self.cache.lock();
-        let raw_entry = match cache.raw_entry_mut().from_key(key) {
-            RawEntryMut::Vacant(_) => return,
-            RawEntryMut::Occupied(x) => x,
-        };
-
-        // Remove the entry if it was created prior to lookup timestamp.
-        let entry = raw_entry.get();
-        let (created_at, expires_at) = (entry.created_at, entry.expires_at);
-        let should_remove = created_at <= info.created_at || expires_at <= now;
-
-        if should_remove {
-            raw_entry.remove();
-        }
-
-        drop(cache); // drop lock before logging
-        debug!(
-            ?created_at,
-            ?expires_at,
-            entry_removed = should_remove,
-            "processed a cache entry invalidation event"
-        );
-    }
-}

proxy/src/cache/timed_lru/tests.rs

@@ -1,104 +0,0 @@
-use super::*;
-
-/// Check that we can define the cache for certain types.
-#[test]
-fn definition() {
-    // Check for trivial yet possible types.
-    let cache = TimedLru::<(), ()>::new("test", 128, Duration::from_secs(0));
-    let _ = cache.insert(Default::default(), Default::default());
-    let _ = cache.get(&());
-
-    // Now for something less trivial.
-    let cache = TimedLru::<String, String>::new("test", 128, Duration::from_secs(0));
-    let _ = cache.insert(Default::default(), Default::default());
-    let _ = cache.get(&String::default());
-    let _ = cache.get("str should work");
-
-    // It should also work for non-cloneable values.
-    struct NoClone;
-    let cache = TimedLru::<Box<str>, NoClone>::new("test", 128, Duration::from_secs(0));
-    let _ = cache.insert(Default::default(), NoClone.into());
-    let _ = cache.get(&Box::<str>::from("boxed str"));
-    let _ = cache.get("str should work");
-}
-
-#[test]
-fn insert() {
-    const CAPACITY: usize = 2;
-    let cache = TimedLru::<String, u32>::new("test", CAPACITY, Duration::from_secs(10));
-    assert_eq!(cache.size(), 0);
-
-    let key = Arc::new(String::from("key"));
-
-    let (old, cached) = cache.insert(key.clone(), 42.into());
-    assert_eq!(old, None);
-    assert_eq!(*cached, 42);
-    assert_eq!(cache.size(), 1);
-
-    let (old, cached) = cache.insert(key, 1.into());
-    assert_eq!(old.as_deref(), Some(&42));
-    assert_eq!(*cached, 1);
-    assert_eq!(cache.size(), 1);
-
-    let (old, cached) = cache.insert(Arc::new("N1".to_owned()), 10.into());
-    assert_eq!(old, None);
-    assert_eq!(*cached, 10);
-    assert_eq!(cache.size(), 2);
-
-    let (old, cached) = cache.insert(Arc::new("N2".to_owned()), 20.into());
-    assert_eq!(old, None);
-    assert_eq!(*cached, 20);
-    assert_eq!(cache.size(), 2);
-}
-
-#[test]
-fn get_none() {
-    let cache = TimedLru::<String, u32>::new("test", 2, Duration::from_secs(10));
-    let cached = cache.get("missing");
-    assert!(matches!(cached, None));
-}
-
-#[test]
-fn invalidation_simple() {
-    let cache = TimedLru::<String, u32>::new("test", 2, Duration::from_secs(10));
-    let (_, cached) = cache.insert(String::from("key").into(), 100.into());
-    assert_eq!(cache.size(), 1);
-
-    cached.invalidate();
-
-    assert_eq!(cache.size(), 0);
-    assert!(matches!(cache.get("key"), None));
-}
-
-#[test]
-fn invalidation_preserve_newer() {
-    let cache = TimedLru::<String, u32>::new("test", 2, Duration::from_secs(10));
-    let key = Arc::new(String::from("key"));
-
-    let (_, cached) = cache.insert(key.clone(), 100.into());
-    assert_eq!(cache.size(), 1);
-    let _ = cache.insert(key.clone(), 200.into());
-    assert_eq!(cache.size(), 1);
-    cached.invalidate();
-    assert_eq!(cache.size(), 1);
-
-    let cached = cache.get(key.as_ref());
-    assert_eq!(cached.as_deref(), Some(&200));
-}
-
-#[test]
-fn auto_expiry() {
-    let lifetime = Duration::from_millis(300);
-    let cache = TimedLru::<String, u32>::new("test", 2, lifetime);
-
-    let key = Arc::new(String::from("key"));
-    let _ = cache.insert(key.clone(), 42.into());
-
-    let cached = cache.get(key.as_ref());
-    assert_eq!(cached.as_deref(), Some(&42));
-
-    std::thread::sleep(lifetime);
-
-    let cached = cache.get(key.as_ref());
-    assert_eq!(cached.as_deref(), None);
-}

proxy/src/compute.rs

@@ -1,28 +1,13 @@
-use crate::{
-    auth::parse_endpoint_param,
-    cancellation::CancelClosure,
-    console::messages::{DatabaseInfo, MetricsAuxInfo},
-    console::CachedNodeInfo,
-    error::UserFacingError,
-};
+use crate::{cancellation::CancelClosure, error::UserFacingError};
 use futures::{FutureExt, TryFutureExt};
 use itertools::Itertools;
 use pq_proto::StartupMessageParams;
-use std::{
-    io,
-    net::SocketAddr,
-    sync::atomic::{AtomicBool, Ordering},
-    time::Duration,
-};
+use std::{io, net::SocketAddr, time::Duration};
 use thiserror::Error;
 use tokio::net::TcpStream;
 use tokio_postgres::tls::MakeTlsConnect;
 use tracing::{error, info, warn};
 
-/// Should we allow self-signed certificates in TLS connections?
-/// Most definitely, this shouldn't be allowed in production.
-pub static ALLOW_SELF_SIGNED_COMPUTE: AtomicBool = AtomicBool::new(false);
-
 const COULD_NOT_CONNECT: &str = "Couldn't connect to compute node";
 
 #[derive(Debug, Error)]
@@ -57,88 +42,6 @@ impl UserFacingError for ConnectionError {
 /// A pair of `ClientKey` & `ServerKey` for `SCRAM-SHA-256`.
 pub type ScramKeys = tokio_postgres::config::ScramKeys<32>;
 
-#[derive(Clone)]
-pub enum Password {
-    /// A regular cleartext password.
-    ClearText(Vec<u8>),
-    /// A pair of `ClientKey` & `ServerKey` for `SCRAM-SHA-256`.
-    ScramKeys(ScramKeys),
-}
-
-pub enum ComputeNode {
-    /// Route via link auth.
-    Link(DatabaseInfo),
-    /// Regular compute node.
-    Static {
-        password: Password,
-        info: CachedNodeInfo,
-    },
-}
-
-impl ComputeNode {
-    /// Get metrics auxiliary info.
-    pub fn metrics_aux_info(&self) -> &MetricsAuxInfo {
-        match self {
-            Self::Link(info) => &info.aux,
-            Self::Static { info, .. } => &info.aux,
-        }
-    }
-
-    /// Invalidate compute node info if it's cached.
-    pub fn invalidate(&self) -> bool {
-        if let Self::Static { info, .. } = self {
-            warn!("invalidating compute node info cache entry");
-            info.invalidate();
-            return true;
-        }
-
-        false
-    }
-
-    /// Turn compute node info into a postgres connection config.
-    pub fn to_conn_config(&self) -> ConnCfg {
-        let mut config = ConnCfg::new();
-
-        let (host, port) = match self {
-            Self::Link(info) => {
-                // NB: use pre-supplied dbname, user and password for link auth.
-                // See `ConnCfg::set_startup_params` below.
-                config.0.dbname(&info.dbname).user(&info.user);
-                if let Some(password) = &info.password {
-                    config.0.password(password.as_bytes());
-                }
-
-                (&info.host, info.port)
-            }
-            Self::Static { info, password } => {
-                // NB: setup auth keys (for SCRAM) or plaintext password.
-                match password {
-                    Password::ClearText(text) => config.0.password(text),
-                    Password::ScramKeys(keys) => {
-                        use tokio_postgres::config::AuthKeys;
-                        config.0.auth_keys(AuthKeys::ScramSha256(keys.to_owned()))
-                    }
-                };
-
-                (&info.address.host, info.address.port)
-            }
-        };
-
-        // Backwards compatibility. pg_sni_proxy uses "--" in domain names
-        // while direct connections do not. Once we migrate to pg_sni_proxy
-        // everywhere, we can remove this.
-        config.0.ssl_mode(if host.contains("--") {
-            // We need TLS connection with SNI info to properly route it.
-            tokio_postgres::config::SslMode::Require
-        } else {
-            tokio_postgres::config::SslMode::Disable
-        });
-
-        config.0.host(host).port(port);
-        config
-    }
-}
-
 /// A config for establishing a connection to compute node.
 /// Eventually, `tokio_postgres` will be replaced with something better.
 /// Newtype allows us to implement methods on top of it.
@@ -148,32 +51,43 @@ pub struct ConnCfg(Box<tokio_postgres::Config>);
 
 /// Creation and initialization routines.
 impl ConnCfg {
-    fn new() -> Self {
+    pub fn new() -> Self {
         Self(Default::default())
     }
 
+    /// Reuse password or auth keys from the other config.
+    pub fn reuse_password(&mut self, other: &Self) {
+        if let Some(password) = other.get_password() {
+            self.password(password);
+        }
+
+        if let Some(keys) = other.get_auth_keys() {
+            self.auth_keys(keys);
+        }
+    }
+
     /// Apply startup message params to the connection config.
     pub fn set_startup_params(&mut self, params: &StartupMessageParams) {
         // Only set `user` if it's not present in the config.
         // Link auth flow takes username from the console's response.
-        if let (None, Some(user)) = (self.0.get_user(), params.get("user")) {
-            self.0.user(user);
+        if let (None, Some(user)) = (self.get_user(), params.get("user")) {
+            self.user(user);
         }
 
         // Only set `dbname` if it's not present in the config.
         // Link auth flow takes dbname from the console's response.
-        if let (None, Some(dbname)) = (self.0.get_dbname(), params.get("database")) {
-            self.0.dbname(dbname);
+        if let (None, Some(dbname)) = (self.get_dbname(), params.get("database")) {
+            self.dbname(dbname);
         }
 
         // Don't add `options` if they were only used for specifying a project.
         // Connection pools don't support `options`, because they affect backend startup.
         if let Some(options) = filtered_options(params) {
-            self.0.options(&options);
+            self.options(&options);
         }
 
         if let Some(app_name) = params.get("application_name") {
-            self.0.application_name(app_name);
+            self.application_name(app_name);
         }
 
         // TODO: This is especially ugly...
@@ -181,10 +95,10 @@ impl ConnCfg {
             use tokio_postgres::config::ReplicationMode;
             match replication {
                 "true" | "on" | "yes" | "1" => {
-                    self.0.replication_mode(ReplicationMode::Physical);
+                    self.replication_mode(ReplicationMode::Physical);
                 }
                 "database" => {
-                    self.0.replication_mode(ReplicationMode::Logical);
+                    self.replication_mode(ReplicationMode::Logical);
                 }
                 _other => {}
             }
@@ -199,6 +113,27 @@ impl ConnCfg {
     }
 }
 
+impl std::ops::Deref for ConnCfg {
+    type Target = tokio_postgres::Config;
+
+    fn deref(&self) -> &Self::Target {
+        &self.0
+    }
+}
+
+/// For now, let's make it easier to setup the config.
+impl std::ops::DerefMut for ConnCfg {
+    fn deref_mut(&mut self) -> &mut Self::Target {
+        &mut self.0
+    }
+}
+
+impl Default for ConnCfg {
+    fn default() -> Self {
+        Self::new()
+    }
+}
+
 impl ConnCfg {
     /// Establish a raw TCP connection to the compute node.
     async fn connect_raw(&self) -> io::Result<(SocketAddr, TcpStream, &str)> {
@@ -285,13 +220,16 @@ pub struct PostgresConnection {
 }
 
 impl ConnCfg {
-    async fn do_connect(&self) -> Result<PostgresConnection, ConnectionError> {
+    async fn do_connect(
+        &self,
+        allow_self_signed_compute: bool,
+    ) -> Result<PostgresConnection, ConnectionError> {
         let (socket_addr, stream, host) = self.connect_raw().await?;
 
         let tls_connector = native_tls::TlsConnector::builder()
-            .danger_accept_invalid_certs(ALLOW_SELF_SIGNED_COMPUTE.load(Ordering::Relaxed))
-            .build()?;
-
+            .danger_accept_invalid_certs(allow_self_signed_compute)
+            .build()
+            .unwrap();
         let mut mk_tls = postgres_native_tls::MakeTlsConnector::new(tls_connector);
         let tls = MakeTlsConnect::<tokio::net::TcpStream>::make_tls_connect(&mut mk_tls, host)?;
 
@@ -323,8 +261,11 @@ impl ConnCfg {
     }
 
     /// Connect to a corresponding compute node.
-    pub async fn connect(&self) -> Result<PostgresConnection, ConnectionError> {
-        self.do_connect()
+    pub async fn connect(
+        &self,
+        allow_self_signed_compute: bool,
+    ) -> Result<PostgresConnection, ConnectionError> {
+        self.do_connect(allow_self_signed_compute)
             .inspect_err(|err| {
                 // Immediately log the error we have at our disposal.
                 error!("couldn't connect to compute node: {err}");
@@ -338,7 +279,7 @@ fn filtered_options(params: &StartupMessageParams) -> Option<String> {
     #[allow(unstable_name_collisions)]
     let options: String = params
         .options_raw()?
-        .filter(|opt| parse_endpoint_param(opt).is_none())
+        .filter(|opt| !opt.starts_with("project="))
         .intersperse(" ") // TODO: use impl from std once it's stabilized
         .collect();
 

proxy/src/config.rs

@@ -12,6 +12,7 @@ pub struct ProxyConfig {
     pub tls_config: Option<TlsConfig>,
     pub auth_backend: auth::BackendType<'static, ()>,
     pub metric_collection: Option<MetricCollectionConfig>,
+    pub allow_self_signed_compute: bool,
 }
 
 #[derive(Debug)]
@@ -99,10 +100,9 @@ impl CertResolver {
         is_default: bool,
     ) -> anyhow::Result<()> {
         let priv_key = {
-            let key_bytes = std::fs::read(key_path)
-                .context(format!("Failed to read TLS keys at '{key_path}'"))?;
+            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 parse TLS keys at '{key_path}'"))?;
+                .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()
@@ -210,11 +210,8 @@ pub struct CacheOptions {
 }
 
 impl CacheOptions {
-    /// Default options for [`crate::auth::caches::AuthInfoCache`].
-    pub const DEFAULT_AUTH_INFO: &str = "size=4000,ttl=5s";
-
     /// Default options for [`crate::auth::caches::NodeInfoCache`].
-    pub const DEFAULT_NODE_INFO: &str = "size=4000,ttl=5m";
+    pub const DEFAULT_OPTIONS_NODE_INFO: &str = "size=4000,ttl=5m";
 
     /// Parse cache options passed via cmdline.
     /// Example: [`Self::DEFAULT_OPTIONS_NODE_INFO`].

proxy/src/console.rs

@@ -6,17 +6,12 @@ pub mod messages;
 
 /// Wrappers for console APIs and their mocks.
 pub mod provider;
-pub use provider::{errors, Api, ConsoleReqExtra};
-pub use provider::{AuthInfo, NodeInfo};
-pub use provider::{CachedAuthInfo, CachedNodeInfo};
+pub use provider::{errors, Api, AuthInfo, CachedNodeInfo, ConsoleReqExtra, NodeInfo};
 
 /// Various cache-related types.
 pub mod caches {
-    pub use super::provider::{ApiCaches, AuthInfoCache, AuthInfoCacheKey, NodeInfoCache};
+    pub use super::provider::{ApiCaches, NodeInfoCache};
 }
 
 /// Console's management API.
 pub mod mgmt;
-
-/// Console's notification bus.
-pub mod notifications;

proxy/src/console/messages.rs

@@ -22,37 +22,11 @@ impl fmt::Debug for GetRoleSecret {
     }
 }
 
-/// Represents compute node's host & port pair.
-#[derive(Debug)]
-pub struct PgEndpoint {
-    pub host: Box<str>,
-    pub port: u16,
-}
-
-impl<'de> Deserialize<'de> for PgEndpoint {
-    fn deserialize<D>(deserializer: D) -> Result<Self, D::Error>
-    where
-        D: serde::Deserializer<'de>,
-    {
-        use serde::de::Error;
-
-        let raw = String::deserialize(deserializer)?;
-        let (host, port) = raw
-            .rsplit_once(':')
-            .ok_or_else(|| Error::custom(format!("bad compute address: {raw}")))?;
-
-        Ok(PgEndpoint {
-            host: host.into(),
-            port: port.parse().map_err(Error::custom)?,
-        })
-    }
-}
-
 /// Response which holds compute node's `host:port` pair.
 /// Returned by the `/proxy_wake_compute` API method.
 #[derive(Debug, Deserialize)]
 pub struct WakeCompute {
-    pub address: PgEndpoint,
+    pub address: Box<str>,
     pub aux: MetricsAuxInfo,
 }
 
@@ -213,24 +187,4 @@ mod tests {
 
         Ok(())
     }
-
-    #[test]
-    fn parse_wake_compute() -> anyhow::Result<()> {
-        let _: WakeCompute = serde_json::from_value(json!({
-            "address": "127.0.0.1:5432",
-            "aux": dummy_aux(),
-        }))?;
-
-        let _: WakeCompute = serde_json::from_value(json!({
-            "address": "[::1]:5432",
-            "aux": dummy_aux(),
-        }))?;
-
-        serde_json::from_value::<WakeCompute>(json!({
-            "address": "localhost:5432",
-            "aux": dummy_aux(),
-        }))?;
-
-        Ok(())
-    }
 }

proxy/src/console/notifications.rs

@@ -1,70 +0,0 @@
-use crate::console::caches::{ApiCaches, AuthInfoCacheKey};
-use futures::StreamExt;
-use serde::Deserialize;
-
-const CHANNEL_NAME: &str = "proxy_notifications";
-
-#[derive(Debug, Deserialize)]
-#[serde(tag = "type")]
-enum Notification<'a> {
-    #[serde(rename = "password_changed")]
-    PasswordChanged { project: &'a str, role: &'a str },
-}
-
-#[tracing::instrument(skip(caches))]
-fn handle_message(msg: redis::Msg, caches: &ApiCaches) -> anyhow::Result<()> {
-    let payload: String = msg.get_payload()?;
-
-    use Notification::*;
-    match serde_json::from_str(&payload) {
-        Ok(PasswordChanged { project, role }) => {
-            let key = AuthInfoCacheKey {
-                project: project.into(),
-                role: role.into(),
-            };
-
-            tracing::info!(key = ?key, "invalidating auth info");
-            caches.auth_info.remove(&key);
-        }
-        Err(e) => tracing::error!("broken message: {e}"),
-    }
-
-    Ok(())
-}
-
-/// Handle console's invalidation messages.
-#[tracing::instrument(name = "console_notifications", skip_all)]
-pub async fn task_main(url: String, caches: &ApiCaches) -> anyhow::Result<()> {
-    let client = redis::Client::open(url.as_ref())?;
-    let mut conn = client.get_async_connection().await?.into_pubsub();
-
-    tracing::info!("subscribing to a channel `{CHANNEL_NAME}`");
-    conn.subscribe(CHANNEL_NAME).await?;
-    let mut stream = conn.on_message();
-
-    while let Some(msg) = stream.next().await {
-        handle_message(msg, caches)?;
-    }
-
-    Ok(())
-}
-
-#[cfg(test)]
-mod tests {
-    use super::*;
-    use serde_json::json;
-
-    #[test]
-    fn parse_notification() -> anyhow::Result<()> {
-        let text = json!({
-            "type": "password_changed",
-            "project": "very-nice",
-            "role": "borat",
-        })
-        .to_string();
-
-        let _: Notification = serde_json::from_str(&text)?;
-
-        Ok(())
-    }
-}

proxy/src/console/provider.rs

@@ -1,13 +1,14 @@
 pub mod mock;
 pub mod neon;
 
-use super::messages::{MetricsAuxInfo, PgEndpoint};
+use super::messages::MetricsAuxInfo;
 use crate::{
     auth::ClientCredentials,
-    cache::{types::Cached, TimedLru},
-    scram,
+    cache::{timed_lru, TimedLru},
+    compute, scram,
 };
 use async_trait::async_trait;
+use std::sync::Arc;
 
 pub mod errors {
     use crate::{
@@ -111,9 +112,11 @@ pub mod errors {
             }
         }
     }
-
     #[derive(Debug, Error)]
     pub enum WakeComputeError {
+        #[error("Console responded with a malformed compute address: {0}")]
+        BadComputeAddress(Box<str>),
+
         #[error(transparent)]
         ApiError(ApiError),
     }
@@ -129,7 +132,10 @@ pub mod errors {
         fn to_string_client(&self) -> String {
             use WakeComputeError::*;
             match self {
-                // API might return a meaningful error.
+                // We shouldn't show user the address even if it's broken.
+                // Besides, user is unlikely to care about this detail.
+                BadComputeAddress(_) => REQUEST_FAILED.to_owned(),
+                // However, API might return a meaningful error.
                 ApiError(e) => e.to_string_client(),
             }
         }
@@ -154,26 +160,23 @@ pub enum AuthInfo {
 }
 
 /// Info for establishing a connection to a compute node.
-/// This struct is cached, so we shouldn't store any user creds here.
+/// This is what we get after auth succeeded, but not before!
+#[derive(Clone)]
 pub struct NodeInfo {
-    /// Address of the compute node.
-    pub address: PgEndpoint,
+    /// Compute node connection params.
+    /// It's sad that we have to clone this, but this will improve
+    /// once we migrate to a bespoke connection logic.
+    pub config: compute::ConnCfg,
 
     /// Labels for proxy's metrics.
-    pub aux: MetricsAuxInfo,
+    pub aux: Arc<MetricsAuxInfo>,
+
+    /// Whether we should accept self-signed certificates (for testing)
+    pub allow_self_signed_compute: bool,
 }
 
-pub type NodeInfoCache = TimedLru<Box<str>, NodeInfo>;
-pub type CachedNodeInfo = Cached<NodeInfo>;
-
-#[derive(Debug, Hash, PartialEq, Eq)]
-pub struct AuthInfoCacheKey {
-    pub project: Box<str>,
-    pub role: Box<str>,
-}
-
-pub type AuthInfoCache = TimedLru<AuthInfoCacheKey, AuthInfo>;
-pub type CachedAuthInfo = Cached<AuthInfo>;
+pub type NodeInfoCache = TimedLru<Arc<str>, NodeInfo>;
+pub type CachedNodeInfo = timed_lru::Cached<&'static NodeInfoCache>;
 
 /// This will allocate per each call, but the http requests alone
 /// already require a few allocations, so it should be fine.
@@ -184,7 +187,7 @@ pub trait Api {
         &self,
         extra: &ConsoleReqExtra<'_>,
         creds: &ClientCredentials<'_>,
-    ) -> Result<Option<CachedAuthInfo>, errors::GetAuthInfoError>;
+    ) -> Result<Option<AuthInfo>, errors::GetAuthInfoError>;
 
     /// Wake up the compute node and return the corresponding connection info.
     async fn wake_compute(
@@ -196,8 +199,6 @@ pub trait Api {
 
 /// Various caches for [`console`].
 pub struct ApiCaches {
-    /// Cache for the `get_auth_info` method.
-    pub auth_info: AuthInfoCache,
-    /// Cache for the `wake_compute` method.
+    /// Cache for the `wake_compute` API method.
     pub node_info: NodeInfoCache,
 }

proxy/src/console/provider/mock.rs

@@ -2,14 +2,13 @@
 
 use super::{
     errors::{ApiError, GetAuthInfoError, WakeComputeError},
-    ConsoleReqExtra, PgEndpoint,
+    AuthInfo, CachedNodeInfo, ConsoleReqExtra, NodeInfo,
 };
-use super::{AuthInfo, NodeInfo};
-use super::{CachedAuthInfo, CachedNodeInfo};
-use crate::{auth::ClientCredentials, error::io_error, scram, url::ApiUrl};
+use crate::{auth::ClientCredentials, compute, error::io_error, scram, url::ApiUrl};
 use async_trait::async_trait;
 use futures::TryFutureExt;
 use thiserror::Error;
+use tokio_postgres::config::SslMode;
 use tracing::{error, info, info_span, warn, Instrument};
 
 #[derive(Debug, Error)]
@@ -85,15 +84,19 @@ impl Api {
     }
 
     async fn do_wake_compute(&self) -> Result<NodeInfo, WakeComputeError> {
-        let host = self.endpoint.host_str().unwrap_or("localhost").into();
-        let port = self.endpoint.port().unwrap_or(5432);
+        let mut config = compute::ConnCfg::new();
+        config
+            .host(self.endpoint.host_str().unwrap_or("localhost"))
+            .port(self.endpoint.port().unwrap_or(5432))
+            .ssl_mode(SslMode::Disable);
 
-        let info = NodeInfo {
-            address: PgEndpoint { host, port },
+        let node = NodeInfo {
+            config,
             aux: Default::default(),
+            allow_self_signed_compute: false,
         };
 
-        Ok(info)
+        Ok(node)
     }
 }
 
@@ -104,9 +107,8 @@ impl super::Api for Api {
         &self,
         _extra: &ConsoleReqExtra<'_>,
         creds: &ClientCredentials<'_>,
-    ) -> Result<Option<CachedAuthInfo>, GetAuthInfoError> {
-        let res = self.do_get_auth_info(creds).await?;
-        Ok(res.map(CachedAuthInfo::new_uncached))
+    ) -> Result<Option<AuthInfo>, GetAuthInfoError> {
+        self.do_get_auth_info(creds).await
     }
 
     #[tracing::instrument(skip_all)]
@@ -115,8 +117,9 @@ impl super::Api for Api {
         _extra: &ConsoleReqExtra<'_>,
         _creds: &ClientCredentials<'_>,
     ) -> Result<CachedNodeInfo, WakeComputeError> {
-        let res = self.do_wake_compute().await?;
-        Ok(CachedNodeInfo::new_uncached(res))
+        self.do_wake_compute()
+            .map_ok(CachedNodeInfo::new_uncached)
+            .await
     }
 }
 

proxy/src/console/provider/neon.rs

@@ -3,19 +3,18 @@
 use super::{
     super::messages::{ConsoleError, GetRoleSecret, WakeCompute},
     errors::{ApiError, GetAuthInfoError, WakeComputeError},
-    ApiCaches, ConsoleReqExtra,
+    ApiCaches, AuthInfo, CachedNodeInfo, ConsoleReqExtra, NodeInfo,
 };
-use super::{AuthInfo, AuthInfoCacheKey, CachedAuthInfo};
-use super::{CachedNodeInfo, NodeInfo};
-use crate::{auth::ClientCredentials, http, scram};
+use crate::{auth::ClientCredentials, compute, http, scram};
 use async_trait::async_trait;
 use futures::TryFutureExt;
+use tokio_postgres::config::SslMode;
 use tracing::{error, info, info_span, warn, Instrument};
 
 #[derive(Clone)]
 pub struct Api {
-    pub endpoint: http::Endpoint,
-    pub caches: &'static ApiCaches,
+    endpoint: http::Endpoint,
+    caches: &'static ApiCaches,
 }
 
 impl Api {
@@ -92,9 +91,22 @@ impl Api {
             let response = self.endpoint.execute(request).await?;
             let body = parse_body::<WakeCompute>(response).await?;
 
+            // Unfortunately, ownership won't let us use `Option::ok_or` here.
+            let (host, port) = match parse_host_port(&body.address) {
+                None => return Err(WakeComputeError::BadComputeAddress(body.address)),
+                Some(x) => x,
+            };
+
+            // Don't set anything but host and port! This config will be cached.
+            // We'll set username and such later using the startup message.
+            // TODO: add more type safety (in progress).
+            let mut config = compute::ConnCfg::new();
+            config.host(host).port(port).ssl_mode(SslMode::Disable); // TLS is not configured on compute nodes.
+
             let node = NodeInfo {
-                address: body.address,
-                aux: body.aux,
+                config,
+                aux: body.aux.into(),
+                allow_self_signed_compute: false,
             };
 
             Ok(node)
@@ -112,28 +124,8 @@ impl super::Api for Api {
         &self,
         extra: &ConsoleReqExtra<'_>,
         creds: &ClientCredentials<'_>,
-    ) -> Result<Option<CachedAuthInfo>, GetAuthInfoError> {
-        let key = AuthInfoCacheKey {
-            project: creds.project().expect("impossible").into(),
-            role: creds.user.into(),
-        };
-
-        // Check if we already have a cached auth info for this project + user combo.
-        // Beware! We shouldn't flush this for unsuccessful auth attempts, otherwise
-        // the cache makes no sense whatsoever in the presence of unfaithful clients.
-        // Instead, we snoop an invalidation queue to keep the cache up-to-date.
-        if let Some(cached) = self.caches.auth_info.get(&key) {
-            info!(key = ?key, "found cached auth info");
-            return Ok(Some(cached));
-        }
-
-        let info = self.do_get_auth_info(extra, creds).await?;
-
-        Ok(info.map(|info| {
-            info!(key = ?key, "creating a cache entry for auth info");
-            let (_, cached) = self.caches.auth_info.insert(key.into(), info.into());
-            cached
-        }))
+    ) -> Result<Option<AuthInfo>, GetAuthInfoError> {
+        self.do_get_auth_info(extra, creds).await
     }
 
     #[tracing::instrument(skip_all)]
@@ -142,21 +134,20 @@ impl super::Api for Api {
         extra: &ConsoleReqExtra<'_>,
         creds: &ClientCredentials<'_>,
     ) -> Result<CachedNodeInfo, WakeComputeError> {
-        let key: Box<str> = creds.project().expect("impossible").into();
+        let key = creds.project().expect("impossible");
 
         // Every time we do a wakeup http request, the compute node will stay up
         // for some time (highly depends on the console's scale-to-zero policy);
         // The connection info remains the same during that period of time,
         // which means that we might cache it to reduce the load and latency.
-        if let Some(cached) = self.caches.node_info.get(&key) {
-            info!(key = ?key, "found cached compute node info");
+        if let Some(cached) = self.caches.node_info.get(key) {
+            info!(key = key, "found cached compute node info");
             return Ok(cached);
         }
 
-        let info = self.do_wake_compute(extra, creds).await?;
-
-        info!(key = ?key, "creating a cache entry for compute node info");
-        let (_, cached) = self.caches.node_info.insert(key.into(), info.into());
+        let node = self.do_wake_compute(extra, creds).await?;
+        let (_, cached) = self.caches.node_info.insert(key.into(), node);
+        info!(key = key, "created a cache entry for compute node info");
 
         Ok(cached)
     }
@@ -186,3 +177,20 @@ async fn parse_body<T: for<'a> serde::Deserialize<'a>>(
     error!("console responded with an error ({status}): {text}");
     Err(ApiError::Console { status, text })
 }
+
+fn parse_host_port(input: &str) -> Option<(&str, u16)> {
+    let (host, port) = input.split_once(':')?;
+    Some((host, port.parse().ok()?))
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    #[test]
+    fn test_parse_host_port() {
+        let (host, port) = parse_host_port("127.0.0.1:5432").expect("failed to parse");
+        assert_eq!(host, "127.0.0.1");
+        assert_eq!(port, 5432);
+    }
+}

proxy/src/http.rs

@@ -3,7 +3,6 @@
 //! directly relying on deps like `reqwest` (think loose coupling).
 
 pub mod server;
-pub mod sql_over_http;
 pub mod websocket;
 
 pub use reqwest::{Request, Response, StatusCode};

proxy/src/http/sql_over_http.rs

@@ -1,598 +0,0 @@
-use futures::pin_mut;
-use futures::StreamExt;
-use hyper::body::HttpBody;
-use hyper::{Body, HeaderMap, Request};
-use pq_proto::StartupMessageParams;
-use serde_json::json;
-use serde_json::Map;
-use serde_json::Value;
-use tokio_postgres::types::Kind;
-use tokio_postgres::types::Type;
-use tokio_postgres::Row;
-use url::Url;
-
-use crate::{auth, config::ProxyConfig, console};
-
-#[derive(serde::Deserialize)]
-struct QueryData {
-    query: String,
-    params: Vec<serde_json::Value>,
-}
-
-const APP_NAME: &str = "sql_over_http";
-const MAX_RESPONSE_SIZE: usize = 1024 * 1024; // 1 MB
-const MAX_REQUEST_SIZE: u64 = 1024 * 1024; // 1 MB
-
-//
-// Convert json non-string types to strings, so that they can be passed to Postgres
-// as parameters.
-//
-fn json_to_pg_text(json: Vec<Value>) -> Result<Vec<String>, serde_json::Error> {
-    json.iter()
-        .map(|value| {
-            match value {
-                Value::Null => serde_json::to_string(value),
-                Value::Bool(_) => serde_json::to_string(value),
-                Value::Number(_) => serde_json::to_string(value),
-                Value::Object(_) => serde_json::to_string(value),
-
-                // no need to escape
-                Value::String(s) => Ok(s.to_string()),
-
-                // special care for arrays
-                Value::Array(_) => json_array_to_pg_array(value),
-            }
-        })
-        .collect()
-}
-
-//
-// Serialize a JSON array to a Postgres array. Contrary to the strings in the params
-// in the array we need to escape the strings. Postgres is okay with arrays of form
-// '{1,"2",3}'::int[], so we don't check that array holds values of the same type, leaving
-// it for Postgres to check.
-//
-// Example of the same escaping in node-postgres: packages/pg/lib/utils.js
-//
-fn json_array_to_pg_array(value: &Value) -> Result<String, serde_json::Error> {
-    match value {
-        // same
-        Value::Null => serde_json::to_string(value),
-        Value::Bool(_) => serde_json::to_string(value),
-        Value::Number(_) => serde_json::to_string(value),
-        Value::Object(_) => serde_json::to_string(value),
-
-        // now needs to be escaped, as it is part of the array
-        Value::String(_) => serde_json::to_string(value),
-
-        // recurse into array
-        Value::Array(arr) => {
-            let vals = arr
-                .iter()
-                .map(json_array_to_pg_array)
-                .collect::<Result<Vec<_>, _>>()?
-                .join(",");
-            Ok(format!("{{{}}}", vals))
-        }
-    }
-}
-
-fn get_conn_info(
-    headers: &HeaderMap,
-    sni_hostname: Option<String>,
-) -> Result<(String, String, String, String), anyhow::Error> {
-    let connection_string = headers
-        .get("Neon-Connection-String")
-        .ok_or(anyhow::anyhow!("missing connection string"))?
-        .to_str()?;
-
-    let connection_url = Url::parse(connection_string)?;
-
-    let protocol = connection_url.scheme();
-    if protocol != "postgres" && protocol != "postgresql" {
-        return Err(anyhow::anyhow!(
-            "connection string must start with postgres: or postgresql:"
-        ));
-    }
-
-    let mut url_path = connection_url
-        .path_segments()
-        .ok_or(anyhow::anyhow!("missing database name"))?;
-
-    let dbname = url_path
-        .next()
-        .ok_or(anyhow::anyhow!("invalid database name"))?;
-
-    let username = connection_url.username();
-    if username.is_empty() {
-        return Err(anyhow::anyhow!("missing username"));
-    }
-
-    let password = connection_url
-        .password()
-        .ok_or(anyhow::anyhow!("no password"))?;
-
-    // TLS certificate selector now based on SNI hostname, so if we are running here
-    // we are sure that SNI hostname is set to one of the configured domain names.
-    let sni_hostname = sni_hostname.ok_or(anyhow::anyhow!("no SNI hostname set"))?;
-
-    let hostname = connection_url
-        .host_str()
-        .ok_or(anyhow::anyhow!("no host"))?;
-
-    let host_header = headers
-        .get("host")
-        .and_then(|h| h.to_str().ok())
-        .and_then(|h| h.split(':').next());
-
-    if hostname != sni_hostname {
-        return Err(anyhow::anyhow!("mismatched SNI hostname and hostname"));
-    } else if let Some(h) = host_header {
-        if h != hostname {
-            return Err(anyhow::anyhow!("mismatched host header and hostname"));
-        }
-    }
-
-    Ok((
-        username.to_owned(),
-        dbname.to_owned(),
-        hostname.to_owned(),
-        password.to_owned(),
-    ))
-}
-
-// TODO: return different http error codes
-pub async fn handle(
-    config: &'static ProxyConfig,
-    request: Request<Body>,
-    sni_hostname: Option<String>,
-) -> anyhow::Result<Value> {
-    //
-    // Determine the destination and connection params
-    //
-    let headers = request.headers();
-    let (username, dbname, hostname, password) = get_conn_info(headers, sni_hostname)?;
-    let credential_params = StartupMessageParams::new([
-        ("user", &username),
-        ("database", &dbname),
-        ("application_name", APP_NAME),
-    ]);
-
-    //
-    // Wake up the destination if needed. Code here is a bit involved because
-    // we reuse the code from the usual proxy and we need to prepare few structures
-    // that this code expects.
-    //
-    let tls = config.tls_config.as_ref();
-    let common_names = tls.and_then(|tls| tls.common_names.clone());
-    let creds = config
-        .auth_backend
-        .as_ref()
-        .map(|_| auth::ClientCredentials::parse(&credential_params, Some(&hostname), common_names))
-        .transpose()?;
-    let extra = console::ConsoleReqExtra {
-        session_id: uuid::Uuid::new_v4(),
-        application_name: Some(APP_NAME),
-    };
-    let node = creds.wake_compute(&extra).await?.expect("msg");
-    let pg_endpoint = &node.value.address;
-    let port = pg_endpoint.port;
-    let host = &pg_endpoint.host;
-
-    let request_content_length = match request.body().size_hint().upper() {
-        Some(v) => v,
-        None => MAX_REQUEST_SIZE + 1,
-    };
-
-    if request_content_length > MAX_REQUEST_SIZE {
-        return Err(anyhow::anyhow!(
-            "request is too large (max {MAX_REQUEST_SIZE} bytes)"
-        ));
-    }
-
-    //
-    // Read the query and query params from the request body
-    //
-    let body = hyper::body::to_bytes(request.into_body()).await?;
-    let QueryData { query, params } = serde_json::from_slice(&body)?;
-    let query_params = json_to_pg_text(params)?;
-
-    //
-    // Connenct to the destination
-    //
-    let (client, connection) = tokio_postgres::Config::new()
-        .host(host)
-        .port(port)
-        .user(&username)
-        .password(&password)
-        .dbname(&dbname)
-        .max_backend_message_size(MAX_RESPONSE_SIZE)
-        .connect(tokio_postgres::NoTls)
-        .await?;
-
-    tokio::spawn(async move {
-        if let Err(e) = connection.await {
-            eprintln!("connection error: {}", e);
-        }
-    });
-
-    //
-    // Now execute the query and return the result
-    //
-    let row_stream = client.query_raw_txt(query, query_params).await?;
-
-    // Manually drain the stream into a vector to leave row_stream hanging
-    // around to get a command tag. Also check that the response is not too
-    // big.
-    pin_mut!(row_stream);
-    let mut rows: Vec<tokio_postgres::Row> = Vec::new();
-    let mut curret_size = 0;
-    while let Some(row) = row_stream.next().await {
-        let row = row?;
-        curret_size += row.body_len();
-        rows.push(row);
-        if curret_size > MAX_RESPONSE_SIZE {
-            return Err(anyhow::anyhow!("response too large"));
-        }
-    }
-
-    // grab the command tag and number of rows affected
-    let command_tag = row_stream.command_tag().unwrap_or_default();
-    let mut command_tag_split = command_tag.split(' ');
-    let command_tag_name = command_tag_split.next().unwrap_or_default();
-    let command_tag_count = if command_tag_name == "INSERT" {
-        // INSERT returns OID first and then number of rows
-        command_tag_split.nth(1)
-    } else {
-        // other commands return number of rows (if any)
-        command_tag_split.next()
-    }
-    .and_then(|s| s.parse::<i64>().ok());
-
-    let fields = if !rows.is_empty() {
-        rows[0]
-            .columns()
-            .iter()
-            .map(|c| {
-                json!({
-                    "name": Value::String(c.name().to_owned()),
-                    "dataTypeID": Value::Number(c.type_().oid().into()),
-                })
-            })
-            .collect::<Vec<_>>()
-    } else {
-        Vec::new()
-    };
-
-    // convert rows to JSON
-    let rows = rows
-        .iter()
-        .map(pg_text_row_to_json)
-        .collect::<Result<Vec<_>, _>>()?;
-
-    // resulting JSON format is based on the format of node-postgres result
-    Ok(json!({
-        "command": command_tag_name,
-        "rowCount": command_tag_count,
-        "rows": rows,
-        "fields": fields,
-    }))
-}
-
-//
-// Convert postgres row with text-encoded values to JSON object
-//
-pub fn pg_text_row_to_json(row: &Row) -> Result<Value, anyhow::Error> {
-    let res = row
-        .columns()
-        .iter()
-        .enumerate()
-        .map(|(i, column)| {
-            let name = column.name();
-            let pg_value = row.as_text(i)?;
-            let json_value = pg_text_to_json(pg_value, column.type_())?;
-            Ok((name.to_string(), json_value))
-        })
-        .collect::<Result<Map<String, Value>, anyhow::Error>>()?;
-
-    Ok(Value::Object(res))
-}
-
-//
-// Convert postgres text-encoded value to JSON value
-//
-pub fn pg_text_to_json(pg_value: Option<&str>, pg_type: &Type) -> Result<Value, anyhow::Error> {
-    if let Some(val) = pg_value {
-        if val == "NULL" {
-            return Ok(Value::Null);
-        }
-
-        if let Kind::Array(elem_type) = pg_type.kind() {
-            return pg_array_parse(val, elem_type);
-        }
-
-        match *pg_type {
-            Type::BOOL => Ok(Value::Bool(val == "t")),
-            Type::INT2 | Type::INT4 => {
-                let val = val.parse::<i32>()?;
-                Ok(Value::Number(serde_json::Number::from(val)))
-            }
-            Type::FLOAT4 | Type::FLOAT8 => {
-                let fval = val.parse::<f64>()?;
-                let num = serde_json::Number::from_f64(fval);
-                if let Some(num) = num {
-                    Ok(Value::Number(num))
-                } else {
-                    // Pass Nan, Inf, -Inf as strings
-                    // JS JSON.stringify() does converts them to null, but we
-                    // want to preserve them, so we pass them as strings
-                    Ok(Value::String(val.to_string()))
-                }
-            }
-            Type::JSON | Type::JSONB => Ok(serde_json::from_str(val)?),
-            _ => Ok(Value::String(val.to_string())),
-        }
-    } else {
-        Ok(Value::Null)
-    }
-}
-
-//
-// Parse postgres array into JSON array.
-//
-// This is a bit involved because we need to handle nested arrays and quoted
-// values. Unlike postgres we don't check that all nested arrays have the same
-// dimensions, we just return them as is.
-//
-fn pg_array_parse(pg_array: &str, elem_type: &Type) -> Result<Value, anyhow::Error> {
-    _pg_array_parse(pg_array, elem_type, false).map(|(v, _)| v)
-}
-
-fn _pg_array_parse(
-    pg_array: &str,
-    elem_type: &Type,
-    nested: bool,
-) -> Result<(Value, usize), anyhow::Error> {
-    let mut pg_array_chr = pg_array.char_indices();
-    let mut level = 0;
-    let mut quote = false;
-    let mut entries: Vec<Value> = Vec::new();
-    let mut entry = String::new();
-
-    // skip bounds decoration
-    if let Some('[') = pg_array.chars().next() {
-        for (_, c) in pg_array_chr.by_ref() {
-            if c == '=' {
-                break;
-            }
-        }
-    }
-
-    while let Some((mut i, mut c)) = pg_array_chr.next() {
-        let mut escaped = false;
-
-        if c == '\\' {
-            escaped = true;
-            (i, c) = pg_array_chr.next().unwrap();
-        }
-
-        match c {
-            '{' if !quote => {
-                level += 1;
-                if level > 1 {
-                    let (res, off) = _pg_array_parse(&pg_array[i..], elem_type, true)?;
-                    entries.push(res);
-                    for _ in 0..off - 1 {
-                        pg_array_chr.next();
-                    }
-                }
-            }
-            '}' => {
-                level -= 1;
-                if level == 0 {
-                    if !entry.is_empty() {
-                        entries.push(pg_text_to_json(Some(&entry), elem_type)?);
-                    }
-                    if nested {
-                        return Ok((Value::Array(entries), i));
-                    }
-                }
-            }
-            '"' if !escaped => {
-                if quote {
-                    // push even if empty
-                    entries.push(pg_text_to_json(Some(&entry), elem_type)?);
-                    entry = String::new();
-                }
-                quote = !quote;
-            }
-            ',' if !quote => {
-                if !entry.is_empty() {
-                    entries.push(pg_text_to_json(Some(&entry), elem_type)?);
-                    entry = String::new();
-                }
-            }
-            _ => {
-                entry.push(c);
-            }
-        }
-    }
-
-    if level != 0 {
-        return Err(anyhow::anyhow!("unbalanced array"));
-    }
-
-    Ok((Value::Array(entries), 0))
-}
-
-#[cfg(test)]
-mod tests {
-    use super::*;
-    use serde_json::json;
-
-    #[test]
-    fn test_atomic_types_to_pg_params() {
-        let json = vec![Value::Bool(true), Value::Bool(false)];
-        let pg_params = json_to_pg_text(json).unwrap();
-        assert_eq!(pg_params, vec!["true", "false"]);
-
-        let json = vec![Value::Number(serde_json::Number::from(42))];
-        let pg_params = json_to_pg_text(json).unwrap();
-        assert_eq!(pg_params, vec!["42"]);
-
-        let json = vec![Value::String("foo\"".to_string())];
-        let pg_params = json_to_pg_text(json).unwrap();
-        assert_eq!(pg_params, vec!["foo\""]);
-
-        let json = vec![Value::Null];
-        let pg_params = json_to_pg_text(json).unwrap();
-        assert_eq!(pg_params, vec!["null"]);
-    }
-
-    #[test]
-    fn test_json_array_to_pg_array() {
-        // atoms and escaping
-        let json = "[true, false, null, 42, \"foo\", \"bar\\\"-\\\\\"]";
-        let json: Value = serde_json::from_str(json).unwrap();
-        let pg_params = json_to_pg_text(vec![json]).unwrap();
-        assert_eq!(
-            pg_params,
-            vec!["{true,false,null,42,\"foo\",\"bar\\\"-\\\\\"}"]
-        );
-
-        // nested arrays
-        let json = "[[true, false], [null, 42], [\"foo\", \"bar\\\"-\\\\\"]]";
-        let json: Value = serde_json::from_str(json).unwrap();
-        let pg_params = json_to_pg_text(vec![json]).unwrap();
-        assert_eq!(
-            pg_params,
-            vec!["{{true,false},{null,42},{\"foo\",\"bar\\\"-\\\\\"}}"]
-        );
-    }
-
-    #[test]
-    fn test_atomic_types_parse() {
-        assert_eq!(
-            pg_text_to_json(Some("foo"), &Type::TEXT).unwrap(),
-            json!("foo")
-        );
-        assert_eq!(pg_text_to_json(None, &Type::TEXT).unwrap(), json!(null));
-        assert_eq!(pg_text_to_json(Some("42"), &Type::INT4).unwrap(), json!(42));
-        assert_eq!(pg_text_to_json(Some("42"), &Type::INT2).unwrap(), json!(42));
-        assert_eq!(
-            pg_text_to_json(Some("42"), &Type::INT8).unwrap(),
-            json!("42")
-        );
-        assert_eq!(
-            pg_text_to_json(Some("42.42"), &Type::FLOAT8).unwrap(),
-            json!(42.42)
-        );
-        assert_eq!(
-            pg_text_to_json(Some("42.42"), &Type::FLOAT4).unwrap(),
-            json!(42.42)
-        );
-        assert_eq!(
-            pg_text_to_json(Some("NaN"), &Type::FLOAT4).unwrap(),
-            json!("NaN")
-        );
-        assert_eq!(
-            pg_text_to_json(Some("Infinity"), &Type::FLOAT4).unwrap(),
-            json!("Infinity")
-        );
-        assert_eq!(
-            pg_text_to_json(Some("-Infinity"), &Type::FLOAT4).unwrap(),
-            json!("-Infinity")
-        );
-
-        let json: Value =
-            serde_json::from_str("{\"s\":\"str\",\"n\":42,\"f\":4.2,\"a\":[null,3,\"a\"]}")
-                .unwrap();
-        assert_eq!(
-            pg_text_to_json(
-                Some(r#"{"s":"str","n":42,"f":4.2,"a":[null,3,"a"]}"#),
-                &Type::JSONB
-            )
-            .unwrap(),
-            json
-        );
-    }
-
-    #[test]
-    fn test_pg_array_parse_text() {
-        fn pt(pg_arr: &str) -> Value {
-            pg_array_parse(pg_arr, &Type::TEXT).unwrap()
-        }
-        assert_eq!(
-            pt(r#"{"aa\"\\\,a",cha,"bbbb"}"#),
-            json!(["aa\"\\,a", "cha", "bbbb"])
-        );
-        assert_eq!(
-            pt(r#"{{"foo","bar"},{"bee","bop"}}"#),
-            json!([["foo", "bar"], ["bee", "bop"]])
-        );
-        assert_eq!(
-            pt(r#"{{{{"foo",NULL,"bop",bup}}}}"#),
-            json!([[[["foo", null, "bop", "bup"]]]])
-        );
-        assert_eq!(
-            pt(r#"{{"1",2,3},{4,NULL,6},{NULL,NULL,NULL}}"#),
-            json!([["1", "2", "3"], ["4", null, "6"], [null, null, null]])
-        );
-    }
-
-    #[test]
-    fn test_pg_array_parse_bool() {
-        fn pb(pg_arr: &str) -> Value {
-            pg_array_parse(pg_arr, &Type::BOOL).unwrap()
-        }
-        assert_eq!(pb(r#"{t,f,t}"#), json!([true, false, true]));
-        assert_eq!(pb(r#"{{t,f,t}}"#), json!([[true, false, true]]));
-        assert_eq!(
-            pb(r#"{{t,f},{f,t}}"#),
-            json!([[true, false], [false, true]])
-        );
-        assert_eq!(
-            pb(r#"{{t,NULL},{NULL,f}}"#),
-            json!([[true, null], [null, false]])
-        );
-    }
-
-    #[test]
-    fn test_pg_array_parse_numbers() {
-        fn pn(pg_arr: &str, ty: &Type) -> Value {
-            pg_array_parse(pg_arr, ty).unwrap()
-        }
-        assert_eq!(pn(r#"{1,2,3}"#, &Type::INT4), json!([1, 2, 3]));
-        assert_eq!(pn(r#"{1,2,3}"#, &Type::INT2), json!([1, 2, 3]));
-        assert_eq!(pn(r#"{1,2,3}"#, &Type::INT8), json!(["1", "2", "3"]));
-        assert_eq!(pn(r#"{1,2,3}"#, &Type::FLOAT4), json!([1.0, 2.0, 3.0]));
-        assert_eq!(pn(r#"{1,2,3}"#, &Type::FLOAT8), json!([1.0, 2.0, 3.0]));
-        assert_eq!(
-            pn(r#"{1.1,2.2,3.3}"#, &Type::FLOAT4),
-            json!([1.1, 2.2, 3.3])
-        );
-        assert_eq!(
-            pn(r#"{1.1,2.2,3.3}"#, &Type::FLOAT8),
-            json!([1.1, 2.2, 3.3])
-        );
-        assert_eq!(
-            pn(r#"{NaN,Infinity,-Infinity}"#, &Type::FLOAT4),
-            json!(["NaN", "Infinity", "-Infinity"])
-        );
-        assert_eq!(
-            pn(r#"{NaN,Infinity,-Infinity}"#, &Type::FLOAT8),
-            json!(["NaN", "Infinity", "-Infinity"])
-        );
-    }
-
-    #[test]
-    fn test_pg_array_with_decoration() {
-        fn p(pg_arr: &str) -> Value {
-            pg_array_parse(pg_arr, &Type::INT2).unwrap()
-        }
-        assert_eq!(
-            p(r#"[1:1][-2:-1][3:5]={{{1,2,3},{4,5,6}}}"#),
-            json!([[[1, 2, 3], [4, 5, 6]]])
-        );
-    }
-}

proxy/src/http/websocket.rs

@@ -4,17 +4,12 @@ use crate::{
 use bytes::{Buf, Bytes};
 use futures::{Sink, Stream, StreamExt};
 use hyper::{
-    server::{
-        accept,
-        conn::{AddrIncoming, AddrStream},
-    },
+    server::{accept, conn::AddrIncoming},
     upgrade::Upgraded,
-    Body, Method, Request, Response, StatusCode,
+    Body, Request, Response, StatusCode,
 };
 use hyper_tungstenite::{tungstenite::Message, HyperWebsocket, WebSocketStream};
 use pin_project_lite::pin_project;
-use serde_json::{json, Value};
-
 use std::{
     convert::Infallible,
     future::ready,
@@ -26,7 +21,6 @@ use tls_listener::TlsListener;
 use tokio::{
     io::{self, AsyncBufRead, AsyncRead, AsyncWrite, ReadBuf},
     net::TcpListener,
-    select,
 };
 use tokio_util::sync::CancellationToken;
 use tracing::{error, info, info_span, warn, Instrument};
@@ -36,8 +30,6 @@ use utils::http::{error::ApiError, json::json_response};
 // Tracking issue: https://github.com/rust-lang/rust/issues/98407.
 use sync_wrapper::SyncWrapper;
 
-use super::sql_over_http;
-
 pin_project! {
     /// This is a wrapper around a [`WebSocketStream`] that
     /// implements [`AsyncRead`] and [`AsyncWrite`].
@@ -167,7 +159,6 @@ async fn ws_handler(
     config: &'static ProxyConfig,
     cancel_map: Arc<CancelMap>,
     session_id: uuid::Uuid,
-    sni_hostname: Option<String>,
 ) -> Result<Response<Body>, ApiError> {
     let host = request
         .headers()
@@ -190,44 +181,8 @@ async fn ws_handler(
 
         // Return the response so the spawned future can continue.
         Ok(response)
-    // TODO: that deserves a refactor as now this function also handles http json client besides websockets.
-    // Right now I don't want to blow up sql-over-http patch with file renames and do that as a follow up instead.
-    } else if request.uri().path() == "/sql" && request.method() == Method::POST {
-        let result = select! {
-            _ = tokio::time::sleep(std::time::Duration::from_secs(10)) => {
-                Err(anyhow::anyhow!("Query timed out"))
-            }
-            response = sql_over_http::handle(config, request, sni_hostname) => {
-                response
-            }
-        };
-        let status_code = match result {
-            Ok(_) => StatusCode::OK,
-            Err(_) => StatusCode::BAD_REQUEST,
-        };
-        let json = match result {
-            Ok(r) => r,
-            Err(e) => {
-                let message = format!("{:?}", e);
-                let code = match e.downcast_ref::<tokio_postgres::Error>() {
-                    Some(e) => match e.code() {
-                        Some(e) => serde_json::to_value(e.code()).unwrap(),
-                        None => Value::Null,
-                    },
-                    None => Value::Null,
-                };
-                json!({ "message": message, "code": code })
-            }
-        };
-        json_response(status_code, json).map(|mut r| {
-            r.headers_mut().insert(
-                "Access-Control-Allow-Origin",
-                hyper::http::HeaderValue::from_static("*"),
-            );
-            r
-        })
     } else {
-        json_response(StatusCode::BAD_REQUEST, "query is not supported")
+        json_response(StatusCode::OK, "Connect with a websocket client")
     }
 }
 
@@ -261,27 +216,20 @@ pub async fn task_main(
         }
     });
 
-    let make_svc =
-        hyper::service::make_service_fn(|stream: &tokio_rustls::server::TlsStream<AddrStream>| {
-            let sni_name = stream.get_ref().1.sni_hostname().map(|s| s.to_string());
-
-            async move {
-                Ok::<_, Infallible>(hyper::service::service_fn(move |req: Request<Body>| {
-                    let sni_name = sni_name.clone();
-                    async move {
-                        let cancel_map = Arc::new(CancelMap::default());
-                        let session_id = uuid::Uuid::new_v4();
-
-                        ws_handler(req, config, cancel_map, session_id, sni_name)
-                            .instrument(info_span!(
-                                "ws-client",
-                                session = format_args!("{session_id}")
-                            ))
-                            .await
-                    }
-                }))
-            }
-        });
+    let make_svc = hyper::service::make_service_fn(|_stream| async move {
+        Ok::<_, Infallible>(hyper::service::service_fn(
+            move |req: Request<Body>| async move {
+                let cancel_map = Arc::new(CancelMap::default());
+                let session_id = uuid::Uuid::new_v4();
+                ws_handler(req, config, cancel_map, session_id)
+                    .instrument(info_span!(
+                        "ws-client",
+                        session = format_args!("{session_id}")
+                    ))
+                    .await
+            },
+        ))
+    });
 
     hyper::Server::builder(accept::from_stream(tls_listener))
         .serve(make_svc)

proxy/src/proxy.rs

@@ -4,7 +4,7 @@ mod tests;
 use crate::{
     auth::{self, backend::AuthSuccess},
     cancellation::{self, CancelMap},
-    compute::{self, ComputeNode, PostgresConnection},
+    compute::{self, PostgresConnection},
     config::{ProxyConfig, TlsConfig},
     console::{self, messages::MetricsAuxInfo},
     error::io_error,
@@ -155,7 +155,7 @@ pub async fn handle_ws_client(
         async { result }.or_else(|e| stream.throw_error(e)).await?
     };
 
-    let client = Client::new(stream, creds, &params, session_id);
+    let client = Client::new(stream, creds, &params, session_id, false);
     cancel_map
         .with_session(|session| client.connect_to_db(session, true))
         .await
@@ -194,7 +194,15 @@ async fn handle_client(
         async { result }.or_else(|e| stream.throw_error(e)).await?
     };
 
-    let client = Client::new(stream, creds, &params, session_id);
+    let allow_self_signed_compute = config.allow_self_signed_compute;
+
+    let client = Client::new(
+        stream,
+        creds,
+        &params,
+        session_id,
+        allow_self_signed_compute,
+    );
     cancel_map
         .with_session(|session| client.connect_to_db(session, false))
         .await
@@ -275,38 +283,61 @@ async fn handshake<S: AsyncRead + AsyncWrite + Unpin>(
     }
 }
 
+/// Try to connect to the compute node once.
+#[tracing::instrument(name = "connect_once", skip_all)]
+async fn connect_to_compute_once(
+    node_info: &console::CachedNodeInfo,
+) -> Result<PostgresConnection, compute::ConnectionError> {
+    // If we couldn't connect, a cached connection info might be to blame
+    // (e.g. the compute node's address might've changed at the wrong time).
+    // Invalidate the cache entry (if any) to prevent subsequent errors.
+    let invalidate_cache = |_: &compute::ConnectionError| {
+        let is_cached = node_info.cached();
+        if is_cached {
+            warn!("invalidating stalled compute node info cache entry");
+            node_info.invalidate();
+        }
+
+        let label = match is_cached {
+            true => "compute_cached",
+            false => "compute_uncached",
+        };
+        NUM_CONNECTION_FAILURES.with_label_values(&[label]).inc();
+    };
+
+    let allow_self_signed_compute = node_info.allow_self_signed_compute;
+
+    node_info
+        .config
+        .connect(allow_self_signed_compute)
+        .inspect_err(invalidate_cache)
+        .await
+}
+
 /// Try to connect to the compute node, retrying if necessary.
 /// This function might update `node_info`, so we take it by `&mut`.
 #[tracing::instrument(skip_all)]
 async fn connect_to_compute(
-    node_info: &mut ComputeNode,
+    node_info: &mut console::CachedNodeInfo,
     params: &StartupMessageParams,
     extra: &console::ConsoleReqExtra<'_>,
     creds: &auth::BackendType<'_, auth::ClientCredentials<'_>>,
 ) -> Result<PostgresConnection, compute::ConnectionError> {
-    let mut num_retries = NUM_RETRIES_WAKE_COMPUTE;
+    let mut num_retries: usize = NUM_RETRIES_WAKE_COMPUTE;
     loop {
-        let mut config = node_info.to_conn_config();
-        config.set_startup_params(params);
-        match config.connect().await {
+        // Apply startup params to the (possibly, cached) compute node info.
+        node_info.config.set_startup_params(params);
+        match connect_to_compute_once(node_info).await {
             Err(e) if num_retries > 0 => {
-                let label = match node_info.invalidate() {
-                    true => "compute_cached",
-                    false => "compute_uncached",
-                };
-                NUM_CONNECTION_FAILURES.with_label_values(&[label]).inc();
-
-                let res = creds.wake_compute(extra).map_err(io_error).await?;
-                match (res, &node_info) {
+                info!("compute node's state has changed; requesting a wake-up");
+                match creds.wake_compute(extra).map_err(io_error).await? {
                     // Update `node_info` and try one more time.
-                    (Some(new), ComputeNode::Static { password, .. }) => {
-                        *node_info = ComputeNode::Static {
-                            password: password.to_owned(),
-                            info: new,
-                        }
+                    Some(mut new) => {
+                        new.config.reuse_password(&node_info.config);
+                        *node_info = new;
                     }
                     // Link auth doesn't work that way, so we just exit.
-                    _ => return Err(e),
+                    None => return Err(e),
                 }
             }
             other => return other,
@@ -399,6 +430,8 @@ struct Client<'a, S> {
     params: &'a StartupMessageParams,
     /// Unique connection ID.
     session_id: uuid::Uuid,
+    /// Allow self-signed certificates (for testing).
+    allow_self_signed_compute: bool,
 }
 
 impl<'a, S> Client<'a, S> {
@@ -408,21 +441,20 @@ impl<'a, S> Client<'a, S> {
         creds: auth::BackendType<'a, auth::ClientCredentials<'a>>,
         params: &'a StartupMessageParams,
         session_id: uuid::Uuid,
+        allow_self_signed_compute: bool,
     ) -> Self {
         Self {
             stream,
             creds,
             params,
             session_id,
+            allow_self_signed_compute,
         }
     }
 }
 
 impl<S: AsyncRead + AsyncWrite + Unpin> Client<'_, S> {
     /// Let the client authenticate and connect to the designated compute node.
-    // Instrumentation logs endpoint name everywhere. Doesn't work for link
-    // auth; strictly speaking we don't know endpoint name in its case.
-    #[tracing::instrument(name = "", fields(ep = self.creds.get_endpoint().unwrap_or("".to_owned())), skip_all)]
     async fn connect_to_db(
         self,
         session: cancellation::Session<'_>,
@@ -433,6 +465,7 @@ impl<S: AsyncRead + AsyncWrite + Unpin> Client<'_, S> {
             mut creds,
             params,
             session_id,
+            allow_self_signed_compute,
         } = self;
 
         let extra = console::ConsoleReqExtra {
@@ -455,19 +488,19 @@ impl<S: AsyncRead + AsyncWrite + Unpin> Client<'_, S> {
             value: mut node_info,
         } = auth_result;
 
+        node_info.allow_self_signed_compute = allow_self_signed_compute;
+
         let mut node = connect_to_compute(&mut node_info, params, &extra, &creds)
             .or_else(|e| stream.throw_error(e))
             .await?;
 
         prepare_client_connection(&node, reported_auth_ok, session, &mut stream).await?;
-
         // Before proxy passing, forward to compute whatever data is left in the
         // PqStream input buffer. Normally there is none, but our serverless npm
         // driver in pipeline mode sends startup, password and first query
         // immediately after opening the connection.
         let (stream, read_buf) = stream.into_inner();
         node.stream.write_all(&read_buf).await?;
-
-        proxy_pass(stream, node.stream, node_info.metrics_aux_info()).await
+        proxy_pass(stream, node.stream, &node_info.aux).await
     }
 }

pyproject.toml

@@ -10,7 +10,7 @@ pytest = "^6.2.5"
 psycopg2-binary = "^2.9.1"
 typing-extensions = "^4.1.0"
 PyJWT = {version = "^2.1.0", extras = ["crypto"]}
-requests = "^2.31.0"
+requests = "^2.26.0"
 pytest-xdist = "^3.0.2"
 asyncpg = "^0.27.0"
 aiopg = "^1.3.1"

scripts/pr-comment-test-report.js

@@ -36,9 +36,11 @@ module.exports = async ({ github, context, fetch, report }) => {
     // Marker to find the comment in the subsequent runs
     const startMarker = `<!--AUTOMATIC COMMENT START #${context.payload.number}-->`
     // Let users know that the comment is updated automatically
-    const autoupdateNotice = `<div align="right"><sub>The comment gets automatically updated with the latest test results<br>${context.payload.pull_request.head.sha} at ${new Date().toISOString()} :recycle:</sub></div>`
+    const autoupdateNotice = `<div align="right"><sub>The comment gets automatically updated with the latest test results :recycle:</sub></div>`
     // 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`
 
@@ -72,6 +74,7 @@ module.exports = async ({ github, context, fetch, report }) => {
     let flakyTestsCount = 0
 
     const pgVersions = new Set()
+    const buildTypes = new Set()
 
     for (const parentSuite of suites.children) {
         for (const suite of parentSuite.children) {
@@ -89,29 +92,28 @@ module.exports = async ({ github, context, fetch, report }) => {
                 }
 
                 pgVersions.add(pgVersion)
+                buildTypes.add(buildType)
 
                 // Removing build type and PostgreSQL version from the test name to make it shorter
                 const testName = test.name.replace(new RegExp(`${buildType}-pg${pgVersion}-?`), "").replace("[]", "")
                 test.pytestName = `${parentSuite.name.replace(".", "/")}/${suite.name}.py::${testName}`
-                test.pgVersion = pgVersion
-                test.buildType = buildType
 
                 if (test.status === "passed") {
-                    passedTests[pgVersion][testName].push(test)
+                    passedTests[pgVersion][buildType].push(test)
                     passedTestsCount += 1
                 } else if (test.status === "failed" || test.status === "broken") {
-                    failedTests[pgVersion][testName].push(test)
+                    failedTests[pgVersion][buildType].push(test)
                     failedTestsCount += 1
                 } else if (test.status === "skipped") {
-                    skippedTests[pgVersion][testName].push(test)
+                    skippedTests[pgVersion][buildType].push(test)
                     skippedTestsCount += 1
                 }
 
                 if (test.retriesCount > 0) {
-                    retriedTests[pgVersion][testName].push(test)
+                    retriedTests[pgVersion][buildType].push(test)
 
                     if (test.retriesStatusChange) {
-                        flakyTests[pgVersion][testName].push(test)
+                        flakyTests[pgVersion][buildType].push(test)
                         flakyTestsCount += 1
                     }
                 }
@@ -120,44 +122,39 @@ module.exports = async ({ github, context, fetch, report }) => {
     }
 
     const totalTestsCount = failedTestsCount + passedTestsCount + skippedTestsCount
-    commentBody += `### ${totalTestsCount} tests run: ${passedTestsCount} passed, ${failedTestsCount} failed, ${skippedTestsCount} skipped ([full report](${reportUrl}))\n___\n`
+    commentBody += `### ${totalTestsCount} tests run: ${passedTestsCount} passed, ${failedTestsCount} failed, ${skippedTestsCount} skipped ([full report](${reportUrl}) for ${commitUrl})\n___\n`
 
-    // Print test resuls from the newest to the oldest Postgres version for release and debug builds.
+    // Print test resuls from the newest to the oldest PostgreSQL version for release and debug builds.
     for (const pgVersion of Array.from(pgVersions).sort().reverse()) {
-        if (Object.keys(failedTests[pgVersion]).length > 0) {
-            commentBody += `#### Failures on Posgres ${pgVersion}\n\n`
-            for (const [testName, tests] of Object.entries(failedTests[pgVersion])) {
-                const links = []
-                for (const test of tests) {
+        for (const buildType of Array.from(buildTypes).sort().reverse()) {
+            if (failedTests[pgVersion][buildType].length > 0) {
+                commentBody += `#### PostgreSQL ${pgVersion} (${buildType} build)\n\n`
+                commentBody += `Failed tests:\n`
+                for (const test of failedTests[pgVersion][buildType]) {
                     const allureLink = `${reportUrl}#suites/${test.parentUid}/${test.uid}`
-                    links.push(`[${test.buildType}](${allureLink})`)
+
+                    commentBody += `- [\`${test.pytestName}\`](${allureLink})`
+                    if (test.retriesCount > 0) {
+                        commentBody += ` (ran [${test.retriesCount + 1} times](${allureLink}/retries))`
+                    }
+                    commentBody += "\n"
                 }
-                commentBody += `- \`${testName}\`: ${links.join(", ")}\n`
+                commentBody += "\n"
             }
-
-            const testsToRerun = Object.values(failedTests[pgVersion]).map(x => x[0].name)
-            const command = `DEFAULT_PG_VERSION=${pgVersion} scripts/pytest -k "${testsToRerun.join(" or ")}"`
-
-            commentBody += "```\n"
-            commentBody += `# Run failed on Postgres ${pgVersion} tests locally:\n`
-            commentBody += `${command}\n`
-            commentBody += "```\n"
         }
     }
 
     if (flakyTestsCount > 0) {
-        commentBody += `<details>\n<summary>Flaky tests (${flakyTestsCount})</summary>\n\n`
+        commentBody += "<details>\n<summary>Flaky tests</summary>\n\n"
         for (const pgVersion of Array.from(pgVersions).sort().reverse()) {
-            if (Object.keys(flakyTests[pgVersion]).length > 0) {
-                commentBody += `#### Postgres ${pgVersion}\n\n`
-                for (const [testName, tests] of Object.entries(flakyTests[pgVersion])) {
-                    const links = []
-                    for (const test of tests) {
-                        const allureLink = `${reportUrl}#suites/${test.parentUid}/${test.uid}/retries`
+            for (const buildType of Array.from(buildTypes).sort().reverse()) {
+                if (flakyTests[pgVersion][buildType].length > 0) {
+                    commentBody += `#### PostgreSQL ${pgVersion} (${buildType} build)\n\n`
+                    for (const test of flakyTests[pgVersion][buildType]) {
                         const status = test.status === "passed" ? ":white_check_mark:" : ":x:"
-                        links.push(`[${status} ${test.buildType}](${allureLink})`)
+                        commentBody += `- ${status} [\`${test.pytestName}\`](${reportUrl}#suites/${test.parentUid}/${test.uid}/retries)\n`
                     }
-                    commentBody += `- \`${testName}\`: ${links.join(", ")}\n`
+                    commentBody += "\n"
                 }
             }
         }

test_runner/fixtures/neon_fixtures.py

@@ -149,7 +149,7 @@ def top_output_dir(base_dir: Path) -> Iterator[Path]:
 
 @pytest.fixture(scope="session")
 def versioned_pg_distrib_dir(pg_distrib_dir: Path, pg_version: PgVersion) -> Iterator[Path]:
-    versioned_dir = pg_distrib_dir / pg_version.v_prefixed
+    versioned_dir = pg_distrib_dir / f"v{pg_version}"
 
     psql_bin_path = versioned_dir / "bin/psql"
     postgres_bin_path = versioned_dir / "bin/postgres"
@@ -1745,8 +1745,8 @@ class PgBin:
     def __init__(self, log_dir: Path, pg_distrib_dir: Path, pg_version: PgVersion):
         self.log_dir = log_dir
         self.pg_version = pg_version
-        self.pg_bin_path = pg_distrib_dir / pg_version.v_prefixed / "bin"
-        self.pg_lib_dir = pg_distrib_dir / pg_version.v_prefixed / "lib"
+        self.pg_bin_path = pg_distrib_dir / f"v{pg_version}" / "bin"
+        self.pg_lib_dir = pg_distrib_dir / f"v{pg_version}" / "lib"
         self.env = os.environ.copy()
         self.env["LD_LIBRARY_PATH"] = str(self.pg_lib_dir)
 
@@ -1845,15 +1845,13 @@ class VanillaPostgres(PgProtocol):
             ]
         )
 
-    def configure(self, options: List[str]) -> "VanillaPostgres":
+    def configure(self, options: List[str]):
         """Append lines into postgresql.conf file."""
         assert not self.running
         with open(os.path.join(self.pgdatadir, "postgresql.conf"), "a") as conf_file:
             conf_file.write("\n".join(options))
 
-        return self
-
-    def start(self, log_path: Optional[str] = None) -> "VanillaPostgres":
+    def start(self, log_path: Optional[str] = None):
         assert not self.running
         self.running = True
 
@@ -1864,15 +1862,11 @@ class VanillaPostgres(PgProtocol):
             ["pg_ctl", "-w", "-D", str(self.pgdatadir), "-l", log_path, "start"]
         )
 
-        return self
-
-    def stop(self) -> "VanillaPostgres":
+    def stop(self):
         assert self.running
         self.running = False
         self.pg_bin.run_capture(["pg_ctl", "-w", "-D", str(self.pgdatadir), "stop"])
 
-        return self
-
     def get_subdir_size(self, subdir) -> int:
         """Return size of pgdatadir subdirectory in bytes."""
         return get_dir_size(os.path.join(self.pgdatadir, subdir))
@@ -2041,17 +2035,6 @@ class NeonProxy(PgProtocol):
                 *["--auth-endpoint", self.pg_conn_url],
             ]
 
-    @dataclass(frozen=True)
-    class Console(AuthBackend):
-        console_url: str
-
-        def extra_args(self) -> list[str]:
-            return [
-                # Postgres auth backend params
-                *["--auth-backend", "console"],
-                *["--auth-endpoint", self.console_url],
-            ]
-
     def __init__(
         self,
         neon_binpath: Path,
@@ -2059,19 +2042,15 @@ class NeonProxy(PgProtocol):
         proxy_port: int,
         http_port: int,
         mgmt_port: int,
-        external_http_port: int,
         auth_backend: NeonProxy.AuthBackend,
         metric_collection_endpoint: Optional[str] = None,
         metric_collection_interval: Optional[str] = None,
     ):
         host = "127.0.0.1"
-        domain = "proxy.localtest.me"  # resolves to 127.0.0.1
-        super().__init__(dsn=auth_backend.default_conn_url, host=domain, port=proxy_port)
+        super().__init__(dsn=auth_backend.default_conn_url, host=host, port=proxy_port)
 
-        self.domain = domain
         self.host = host
         self.http_port = http_port
-        self.external_http_port = external_http_port
         self.neon_binpath = neon_binpath
         self.test_output_dir = test_output_dir
         self.proxy_port = proxy_port
@@ -2083,42 +2062,11 @@ class NeonProxy(PgProtocol):
 
     def start(self) -> NeonProxy:
         assert self._popen is None
-
-        # generate key of it doesn't exist
-        crt_path = self.test_output_dir / "proxy.crt"
-        key_path = self.test_output_dir / "proxy.key"
-
-        if not key_path.exists():
-            r = subprocess.run(
-                [
-                    "openssl",
-                    "req",
-                    "-new",
-                    "-x509",
-                    "-days",
-                    "365",
-                    "-nodes",
-                    "-text",
-                    "-out",
-                    str(crt_path),
-                    "-keyout",
-                    str(key_path),
-                    "-subj",
-                    "/CN=*.localtest.me",
-                    "-addext",
-                    "subjectAltName = DNS:*.localtest.me",
-                ]
-            )
-            assert r.returncode == 0
-
         args = [
             str(self.neon_binpath / "proxy"),
             *["--http", f"{self.host}:{self.http_port}"],
             *["--proxy", f"{self.host}:{self.proxy_port}"],
             *["--mgmt", f"{self.host}:{self.mgmt_port}"],
-            *["--wss", f"{self.host}:{self.external_http_port}"],
-            *["-c", str(crt_path)],
-            *["-k", str(key_path)],
             *self.auth_backend.extra_args(),
         ]
 
@@ -2242,7 +2190,6 @@ def link_proxy(
     http_port = port_distributor.get_port()
     proxy_port = port_distributor.get_port()
     mgmt_port = port_distributor.get_port()
-    external_http_port = port_distributor.get_port()
 
     with NeonProxy(
         neon_binpath=neon_binpath,
@@ -2250,40 +2197,12 @@ def link_proxy(
         proxy_port=proxy_port,
         http_port=http_port,
         mgmt_port=mgmt_port,
-        external_http_port=external_http_port,
         auth_backend=NeonProxy.Link(),
     ) as proxy:
         proxy.start()
         yield proxy
 
 
-@pytest.fixture(scope="function")
-def console_proxy(
-    port_distributor: PortDistributor, neon_binpath: Path, test_output_dir: Path
-) -> Iterator[NeonProxy]:
-    """Neon proxy that routes through link auth."""
-
-    http_port = port_distributor.get_port()
-    proxy_port = port_distributor.get_port()
-    mgmt_port = port_distributor.get_port()
-    console_port = port_distributor.get_port()
-    external_http_port = port_distributor.get_port()
-
-    console_url = f"http://127.0.0.1:{console_port}"
-
-    with NeonProxy(
-        neon_binpath=neon_binpath,
-        test_output_dir=test_output_dir,
-        proxy_port=proxy_port,
-        http_port=http_port,
-        mgmt_port=mgmt_port,
-        external_http_port=external_http_port,
-        auth_backend=NeonProxy.Console(console_url),
-    ) as proxy:
-        proxy.start()
-        yield proxy
-
-
 @pytest.fixture(scope="function")
 def static_proxy(
     vanilla_pg: VanillaPostgres,
@@ -2305,7 +2224,6 @@ def static_proxy(
     proxy_port = port_distributor.get_port()
     mgmt_port = port_distributor.get_port()
     http_port = port_distributor.get_port()
-    external_http_port = port_distributor.get_port()
 
     with NeonProxy(
         neon_binpath=neon_binpath,
@@ -2313,7 +2231,6 @@ def static_proxy(
         proxy_port=proxy_port,
         http_port=http_port,
         mgmt_port=mgmt_port,
-        external_http_port=external_http_port,
         auth_backend=NeonProxy.Postgres(auth_endpoint),
     ) as proxy:
         proxy.start()

test_runner/fixtures/pageserver/http.py

@@ -149,16 +149,11 @@ class PageserverHttpClient(requests.Session):
         assert isinstance(res_json, list)
         return res_json
 
-    def tenant_create(
-        self, new_tenant_id: Optional[TenantId] = None, conf: Optional[Dict[str, Any]] = None
-    ) -> TenantId:
-        if conf is not None:
-            assert "new_tenant_id" not in conf.keys()
+    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,
-                **(conf or {}),
             },
         )
         self.verbose_error(res)

test_runner/fixtures/pg_version.py

@@ -27,12 +27,6 @@ class PgVersion(str, enum.Enum):
     def __repr__(self) -> str:
         return f"'{self.value}'"
 
-    # In GitHub workflows we use Postgres version with v-prefix (e.g. v14 instead of just 14),
-    # sometime we need to do so in tests.
-    @property
-    def v_prefixed(self) -> str:
-        return f"v{self.value}"
-
     @classmethod
     def _missing_(cls, value) -> Optional["PgVersion"]:
         known_values = {v.value for _, v in cls.__members__.items()}

test_runner/regress/test_compatibility.py

@@ -16,7 +16,7 @@ from fixtures.neon_fixtures import (
 )
 from fixtures.pageserver.http import PageserverHttpClient
 from fixtures.pageserver.utils import wait_for_last_record_lsn, wait_for_upload
-from fixtures.pg_version import PgVersion
+from fixtures.pg_version import PgVersion, skip_on_postgres
 from fixtures.types import Lsn
 from pytest import FixtureRequest
 
@@ -41,6 +41,7 @@ check_ondisk_data_compatibility_if_enabled = pytest.mark.skipif(
 )
 
 
+@skip_on_postgres(PgVersion.V15, "Compatibility tests doesn't support Postgres 15 yet")
 @pytest.mark.xdist_group("compatibility")
 @pytest.mark.order(before="test_forward_compatibility")
 def test_create_snapshot(
@@ -48,13 +49,12 @@ def test_create_snapshot(
     pg_bin: PgBin,
     top_output_dir: Path,
     test_output_dir: Path,
-    pg_version: PgVersion,
 ):
     # The test doesn't really test anything
     # it creates a new snapshot for releases after we tested the current version against the previous snapshot in `test_backward_compatibility`.
     #
     # There's no cleanup here, it allows to adjust the data in `test_backward_compatibility` itself without re-collecting it.
-    neon_env_builder.pg_version = pg_version
+    neon_env_builder.pg_version = PgVersion.V14
     neon_env_builder.num_safekeepers = 3
     neon_env_builder.enable_local_fs_remote_storage()
     neon_env_builder.preserve_database_files = True
@@ -90,14 +90,13 @@ def test_create_snapshot(
     env.pageserver.stop()
 
     # Directory `compatibility_snapshot_dir` is uploaded to S3 in a workflow, keep the name in sync with it
-    compatibility_snapshot_dir = (
-        top_output_dir / f"compatibility_snapshot_pg{pg_version.v_prefixed}"
-    )
+    compatibility_snapshot_dir = top_output_dir / "compatibility_snapshot_pg14"
     if compatibility_snapshot_dir.exists():
         shutil.rmtree(compatibility_snapshot_dir)
     shutil.copytree(test_output_dir, compatibility_snapshot_dir)
 
 
+@skip_on_postgres(PgVersion.V15, "Compatibility tests doesn't support Postgres 15 yet")
 @check_ondisk_data_compatibility_if_enabled
 @pytest.mark.xdist_group("compatibility")
 @pytest.mark.order(after="test_create_snapshot")
@@ -116,7 +115,7 @@ def test_backward_compatibility(
     compatibility_snapshot_dir_env = os.environ.get("COMPATIBILITY_SNAPSHOT_DIR")
     assert (
         compatibility_snapshot_dir_env is not None
-    ), f"COMPATIBILITY_SNAPSHOT_DIR is not set. It should be set to `compatibility_snapshot_pg{pg_version.v_prefixed}` path generateted by test_create_snapshot (ideally generated by the previous version of Neon)"
+    ), "COMPATIBILITY_SNAPSHOT_DIR is not set. It should be set to `compatibility_snapshot_pg14` path generateted by test_create_snapshot (ideally generated by the previous version of Neon)"
     compatibility_snapshot_dir = Path(compatibility_snapshot_dir_env).resolve()
 
     breaking_changes_allowed = (
@@ -156,6 +155,7 @@ def test_backward_compatibility(
     ), "Breaking changes are allowed by ALLOW_BACKWARD_COMPATIBILITY_BREAKAGE, but the test has passed without any breakage"
 
 
+@skip_on_postgres(PgVersion.V15, "Compatibility tests doesn't support Postgres 15 yet")
 @check_ondisk_data_compatibility_if_enabled
 @pytest.mark.xdist_group("compatibility")
 @pytest.mark.order(after="test_create_snapshot")
@@ -183,9 +183,7 @@ def test_forward_compatibility(
     ), "COMPATIBILITY_POSTGRES_DISTRIB_DIR is not set. It should be set to a pg_install directrory (ideally generated by the previous version of Neon)"
     compatibility_postgres_distrib_dir = Path(compatibility_postgres_distrib_dir_env).resolve()
 
-    compatibility_snapshot_dir = (
-        top_output_dir / f"compatibility_snapshot_pg{pg_version.v_prefixed}"
-    )
+    compatibility_snapshot_dir = top_output_dir / "compatibility_snapshot_pg14"
 
     breaking_changes_allowed = (
         os.environ.get("ALLOW_FORWARD_COMPATIBILITY_BREAKAGE", "false").lower() == "true"

test_runner/regress/test_disk_usage_eviction.py

@@ -118,11 +118,6 @@ class EvictionEnv:
 
         wait_until(10, 1, statvfs_called)
 
-        # these can sometimes happen during startup before any tenants have been
-        # loaded, so nothing can be evicted, we just wait for next iteration which
-        # is able to evict.
-        self.neon_env.pageserver.allowed_errors.append(".*WARN.* disk usage still high.*")
-
 
 @pytest.fixture
 def eviction_env(request, neon_env_builder: NeonEnvBuilder, pg_bin: PgBin) -> EvictionEnv:

test_runner/regress/test_hot_standby.py

@@ -1,7 +1,9 @@
 import pytest
 from fixtures.neon_fixtures import NeonEnv
+from fixtures.pg_version import PgVersion, xfail_on_postgres
 
 
+@xfail_on_postgres(PgVersion.V15, reason="https://github.com/neondatabase/neon/pull/4182")
 @pytest.mark.timeout(1800)
 def test_hot_standby(neon_simple_env: NeonEnv):
     env = neon_simple_env

test_runner/regress/test_metric_collection.py

@@ -204,7 +204,6 @@ def proxy_with_metric_collector(
     http_port = port_distributor.get_port()
     proxy_port = port_distributor.get_port()
     mgmt_port = port_distributor.get_port()
-    external_http_port = port_distributor.get_port()
 
     (host, port) = httpserver_listen_address
     metric_collection_endpoint = f"http://{host}:{port}/billing/api/v1/usage_events"
@@ -216,7 +215,6 @@ def proxy_with_metric_collector(
         proxy_port=proxy_port,
         http_port=http_port,
         mgmt_port=mgmt_port,
-        external_http_port=external_http_port,
         metric_collection_endpoint=metric_collection_endpoint,
         metric_collection_interval=metric_collection_interval,
         auth_backend=NeonProxy.Link(),

test_runner/regress/test_ondemand_download.py

@@ -20,7 +20,6 @@ from fixtures.pageserver.utils import (
     assert_tenant_state,
     wait_for_last_record_lsn,
     wait_for_upload,
-    wait_for_upload_queue_empty,
     wait_until_tenant_state,
 )
 from fixtures.types import Lsn
@@ -64,15 +63,12 @@ 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
-            # allow compaction with the checkpoint
             "compaction_threshold": "3",
             "compaction_target_size": f"{10 * 1024 ** 2}",  # 10 MB
-            # but don't run compaction in background or on restart
-            "compaction_period": "0s",
         }
     )
     env.initial_tenant = tenant
@@ -99,17 +95,9 @@ def test_ondemand_download_large_rel(
 
         current_lsn = Lsn(query_scalar(cur, "SELECT pg_current_wal_flush_lsn()"))
 
+    # wait until pageserver receives that data
     wait_for_last_record_lsn(client, tenant_id, timeline_id, current_lsn)
 
-    # stop endpoint before checkpoint to stop wal generation
-    endpoint.stop()
-
-    # stopping of safekeepers now will help us not to calculate logical size
-    # after startup, so page requests should be the only one on-demand
-    # downloading the layers
-    for sk in env.safekeepers:
-        sk.stop()
-
     # run checkpoint manually to be sure that data landed in remote storage
     client.timeline_checkpoint(tenant_id, timeline_id)
 
@@ -118,6 +106,7 @@ def test_ondemand_download_large_rel(
     log.info("uploads have finished")
 
     ##### Stop the first pageserver instance, erase all its data
+    endpoint.stop()
     env.pageserver.stop()
 
     # remove all the layer files
@@ -128,13 +117,8 @@ def test_ondemand_download_large_rel(
     ##### Second start, restore the data and ensure it's the same
     env.pageserver.start()
 
-    # start a readonly endpoint which we'll use to check the database.
-    # readonly (with lsn=) is required so that we don't try to connect to
-    # safekeepers, that have now been shut down.
-    endpoint = env.endpoints.create_start("main", lsn=current_lsn)
-
+    endpoint.start()
     before_downloads = get_num_downloaded_layers(client, tenant_id, timeline_id)
-    assert before_downloads != 0, "basebackup should on-demand non-zero layers"
 
     # Probe in the middle of the table. There's a high chance that the beginning
     # and end of the table was stored together in the same layer files with data
@@ -165,7 +149,6 @@ def test_ondemand_download_timetravel(
 
     ##### First start, insert data and upload it to the remote storage
     env = neon_env_builder.init_start()
-    pageserver_http = env.pageserver.http_client()
 
     # Override defaults, to create more layers
     tenant, _ = env.neon_cli.create_tenant(
@@ -242,8 +225,7 @@ def test_ondemand_download_timetravel(
     assert filled_current_physical == filled_size, "we don't yet do layer eviction"
 
     # Wait until generated image layers are uploaded to S3
-    if remote_storage_kind is not None:
-        wait_for_upload_queue_empty(pageserver_http, env.initial_tenant, timeline_id)
+    time.sleep(3)
 
     env.pageserver.stop()
 

test_runner/regress/test_pg_regress.py

@@ -5,6 +5,7 @@ from pathlib import Path
 
 import pytest
 from fixtures.neon_fixtures import NeonEnv, check_restored_datadir_content
+from fixtures.pg_version import PgVersion, xfail_on_postgres
 
 
 # Run the main PostgreSQL regression tests, in src/test/regress.
@@ -32,8 +33,8 @@ def test_pg_regress(
     (runpath / "testtablespace").mkdir(parents=True)
 
     # Compute all the file locations that pg_regress will need.
-    build_path = pg_distrib_dir / f"build/{env.pg_version.v_prefixed}/src/test/regress"
-    src_path = base_dir / f"vendor/postgres-{env.pg_version.v_prefixed}/src/test/regress"
+    build_path = pg_distrib_dir / f"build/v{env.pg_version}/src/test/regress"
+    src_path = base_dir / f"vendor/postgres-v{env.pg_version}/src/test/regress"
     bindir = pg_distrib_dir / f"v{env.pg_version}/bin"
     schedule = src_path / "parallel_schedule"
     pg_regress = build_path / "pg_regress"
@@ -71,6 +72,7 @@ def test_pg_regress(
 #
 # This runs for a long time, especially in debug mode, so use a larger-than-default
 # timeout.
+@xfail_on_postgres(PgVersion.V15, reason="https://github.com/neondatabase/neon/pull/4213")
 @pytest.mark.timeout(1800)
 def test_isolation(
     neon_simple_env: NeonEnv,
@@ -95,8 +97,8 @@ def test_isolation(
     (runpath / "testtablespace").mkdir(parents=True)
 
     # Compute all the file locations that pg_isolation_regress will need.
-    build_path = pg_distrib_dir / f"build/{env.pg_version.v_prefixed}/src/test/isolation"
-    src_path = base_dir / f"vendor/postgres-{env.pg_version.v_prefixed}/src/test/isolation"
+    build_path = pg_distrib_dir / f"build/v{env.pg_version}/src/test/isolation"
+    src_path = base_dir / f"vendor/postgres-v{env.pg_version}/src/test/isolation"
     bindir = pg_distrib_dir / f"v{env.pg_version}/bin"
     schedule = src_path / "isolation_schedule"
     pg_isolation_regress = build_path / "pg_isolation_regress"

test_runner/regress/test_proxy.py

@@ -1,13 +1,8 @@
-import json
-import logging
 import subprocess
-from typing import Any, List, cast
 
 import psycopg2
 import pytest
-import requests
-from aiohttp import web
-from fixtures.neon_fixtures import PSQL, NeonProxy, PortDistributor, VanillaPostgres
+from fixtures.neon_fixtures import PSQL, NeonProxy, VanillaPostgres
 
 
 def test_proxy_select_1(static_proxy: NeonProxy):
@@ -15,18 +10,9 @@ def test_proxy_select_1(static_proxy: NeonProxy):
     A simplest smoke test: check proxy against a local postgres instance.
     """
 
-    # no SNI, deprecated `options=project` syntax (before we had several endpoint in project)
-    out = static_proxy.safe_psql("select 1", sslsni=0, options="project=generic-project-name")
+    out = static_proxy.safe_psql("select 1", options="project=generic-project-name")
     assert out[0][0] == 1
 
-    # no SNI, new `options=endpoint` syntax
-    out = static_proxy.safe_psql("select 1", sslsni=0, options="endpoint=generic-project-name")
-    assert out[0][0] == 1
-
-    # with SNI
-    out = static_proxy.safe_psql("select 42", host="generic-project-name.localtest.me")
-    assert out[0][0] == 42
-
 
 def test_password_hack(static_proxy: NeonProxy):
     """
@@ -36,16 +22,13 @@ def test_password_hack(static_proxy: NeonProxy):
 
     user = "borat"
     password = "password"
-    static_proxy.safe_psql(f"create role {user} with login password '{password}'")
+    static_proxy.safe_psql(
+        f"create role {user} with login password '{password}'", options="project=irrelevant"
+    )
 
     # Note the format of `magic`!
     magic = f"project=irrelevant;{password}"
-    out = static_proxy.safe_psql("select 1", sslsni=0, user=user, password=magic)
-    assert out[0][0] == 1
-
-    magic = f"endpoint=irrelevant;{password}"
-    out = static_proxy.safe_psql("select 1", sslsni=0, user=user, password=magic)
-    assert out[0][0] == 1
+    static_proxy.safe_psql("select 1", sslsni=0, user=user, password=magic)
 
     # Must also check that invalid magic won't be accepted.
     with pytest.raises(psycopg2.OperationalError):
@@ -73,28 +56,18 @@ async def test_link_auth(vanilla_pg: VanillaPostgres, link_proxy: NeonProxy):
     assert out == "42"
 
 
-@pytest.mark.parametrize("option_name", ["project", "endpoint"])
-def test_proxy_options(static_proxy: NeonProxy, option_name: str):
+def test_proxy_options(static_proxy: NeonProxy):
     """
     Check that we pass extra `options` to the PostgreSQL server:
-    * `project=...` and `endpoint=...` shouldn't be passed at all
-    * (otherwise postgres will raise an error).
+    * `project=...` shouldn't be passed at all (otherwise postgres will raise an error).
     * everything else should be passed as-is.
     """
 
-    options = f"{option_name}=irrelevant -cproxytest.option=value"
-    out = static_proxy.safe_psql("show proxytest.option", options=options, sslsni=0)
-    assert out[0][0] == "value"
-
-    options = f"-c proxytest.foo=\\ str {option_name}=irrelevant"
-    out = static_proxy.safe_psql("show proxytest.foo", options=options, sslsni=0)
-    assert out[0][0] == " str"
-
-    options = "-cproxytest.option=value"
+    options = "project=irrelevant -cproxytest.option=value"
     out = static_proxy.safe_psql("show proxytest.option", options=options)
     assert out[0][0] == "value"
 
-    options = "-c proxytest.foo=\\ str"
+    options = "-c proxytest.foo=\\ str project=irrelevant"
     out = static_proxy.safe_psql("show proxytest.foo", options=options)
     assert out[0][0] == " str"
 
@@ -106,28 +79,28 @@ def test_auth_errors(static_proxy: NeonProxy):
 
     # User does not exist
     with pytest.raises(psycopg2.Error) as exprinfo:
-        static_proxy.connect(user="pinocchio")
+        static_proxy.connect(user="pinocchio", options="project=irrelevant")
     text = str(exprinfo.value).strip()
-    assert text.find("password authentication failed for user 'pinocchio'") != -1
+    assert text.endswith("password authentication failed for user 'pinocchio'")
 
     static_proxy.safe_psql(
-        "create role pinocchio with login password 'magic'",
+        "create role pinocchio with login password 'magic'", options="project=irrelevant"
     )
 
     # User exists, but password is missing
     with pytest.raises(psycopg2.Error) as exprinfo:
-        static_proxy.connect(user="pinocchio", password=None)
+        static_proxy.connect(user="pinocchio", password=None, options="project=irrelevant")
     text = str(exprinfo.value).strip()
-    assert text.find("password authentication failed for user 'pinocchio'") != -1
+    assert text.endswith("password authentication failed for user 'pinocchio'")
 
     # User exists, but password is wrong
     with pytest.raises(psycopg2.Error) as exprinfo:
-        static_proxy.connect(user="pinocchio", password="bad")
+        static_proxy.connect(user="pinocchio", password="bad", options="project=irrelevant")
     text = str(exprinfo.value).strip()
-    assert text.find("password authentication failed for user 'pinocchio'") != -1
+    assert text.endswith("password authentication failed for user 'pinocchio'")
 
     # Finally, check that the user can connect
-    with static_proxy.connect(user="pinocchio", password="magic"):
+    with static_proxy.connect(user="pinocchio", password="magic", options="project=irrelevant"):
         pass
 
 
@@ -157,7 +130,7 @@ def test_forward_params_to_client(static_proxy: NeonProxy):
         where name = any(%s)
     """
 
-    with static_proxy.connect() as conn:
+    with static_proxy.connect(options="project=irrelevant") as conn:
         with conn.cursor() as cur:
             cur.execute(query, (reported_params_subset,))
             for name, value in cur.fetchall():
@@ -169,136 +142,13 @@ def test_forward_params_to_client(static_proxy: NeonProxy):
 def test_close_on_connections_exit(static_proxy: NeonProxy):
     # Open two connections, send SIGTERM, then ensure that proxy doesn't exit
     # until after connections close.
-    with static_proxy.connect(), static_proxy.connect():
+    with static_proxy.connect(options="project=irrelevant"), static_proxy.connect(
+        options="project=irrelevant"
+    ):
         static_proxy.terminate()
         with pytest.raises(subprocess.TimeoutExpired):
             static_proxy.wait_for_exit(timeout=2)
         # Ensure we don't accept any more connections
         with pytest.raises(psycopg2.OperationalError):
-            static_proxy.connect()
+            static_proxy.connect(options="project=irrelevant")
     static_proxy.wait_for_exit()
-
-
-def test_sql_over_http(static_proxy: NeonProxy):
-    static_proxy.safe_psql("create role http with login password 'http' superuser")
-
-    def q(sql: str, params: List[Any] = []) -> Any:
-        connstr = f"postgresql://http:http@{static_proxy.domain}:{static_proxy.proxy_port}/postgres"
-        response = requests.post(
-            f"https://{static_proxy.domain}:{static_proxy.external_http_port}/sql",
-            data=json.dumps({"query": sql, "params": params}),
-            headers={"Content-Type": "application/sql", "Neon-Connection-String": connstr},
-            verify=str(static_proxy.test_output_dir / "proxy.crt"),
-        )
-        assert response.status_code == 200
-        return response.json()
-
-    rows = q("select 42 as answer")["rows"]
-    assert rows == [{"answer": 42}]
-
-    rows = q("select $1 as answer", [42])["rows"]
-    assert rows == [{"answer": "42"}]
-
-    rows = q("select $1 * 1 as answer", [42])["rows"]
-    assert rows == [{"answer": 42}]
-
-    rows = q("select $1::int[] as answer", [[1, 2, 3]])["rows"]
-    assert rows == [{"answer": [1, 2, 3]}]
-
-    rows = q("select $1::json->'a' as answer", [{"a": {"b": 42}}])["rows"]
-    assert rows == [{"answer": {"b": 42}}]
-
-    rows = q("select * from pg_class limit 1")["rows"]
-    assert len(rows) == 1
-
-    res = q("create table t(id serial primary key, val int)")
-    assert res["command"] == "CREATE"
-    assert res["rowCount"] is None
-
-    res = q("insert into t(val) values (10), (20), (30) returning id")
-    assert res["command"] == "INSERT"
-    assert res["rowCount"] == 3
-    assert res["rows"] == [{"id": 1}, {"id": 2}, {"id": 3}]
-
-    res = q("select * from t")
-    assert res["command"] == "SELECT"
-    assert res["rowCount"] == 3
-
-    res = q("drop table t")
-    assert res["command"] == "DROP"
-    assert res["rowCount"] is None
-
-
-@pytest.mark.asyncio
-async def test_compute_cache_invalidation(
-    port_distributor: PortDistributor, vanilla_pg: VanillaPostgres, console_proxy: NeonProxy
-):
-    console_url = cast(NeonProxy.Console, console_proxy.auth_backend).console_url
-    logging.info(f"mocked console's url is {console_url}")
-    console_port = int(console_url.split(":")[-1])
-    logging.info(f"mocked console's port is {console_port}")
-
-    routes = web.RouteTableDef()
-
-    @routes.get("/proxy_get_role_secret")
-    async def get_role_secret(request):
-        # corresponds to password "password"
-        secret = ":".join(
-            [
-                "SCRAM-SHA-256$4096",
-                "t33UQcz/cs1D+n9INqThsw==$1NYlCbuxtK7YF2sgECBDTv1Myf8PpHJCT3RgKSXlZL0=",
-                "9iLeGY91MqBQ4ez1389Smo7h+STsJJ5jvu7kNofxj08=",
-            ]
-        )
-
-        return web.json_response({"role_secret": secret})
-
-    wake_compute_called = 0
-    postgres_port = vanilla_pg.default_options["port"]
-
-    @routes.get("/proxy_wake_compute")
-    async def wake_compute(request):
-        nonlocal wake_compute_called
-        wake_compute_called += 1
-
-        nonlocal postgres_port
-        logging.info(f"compute's port is {postgres_port}")
-
-        return web.json_response(
-            {
-                "address": f"127.0.0.1:{postgres_port}",
-                "aux": {
-                    "endpoint_id": "",
-                    "project_id": "",
-                    "branch_id": "",
-                },
-            }
-        )
-
-    console = web.Application()
-    console.add_routes(routes)
-
-    runner = web.AppRunner(console)
-    await runner.setup()
-    await web.TCPSite(runner, "127.0.0.1", console_port).start()
-
-    # Create a user we're going to use in the test sequence
-    user, password = "borat", "password"
-    vanilla_pg.start().safe_psql(f"create role {user} with login password '{password}'")
-
-    async def try_connect():
-        await console_proxy.connect_async(user=user, password=password, dbname="postgres")
-
-    assert wake_compute_called == 0
-
-    # Try connecting to compute
-    await try_connect()
-    assert wake_compute_called == 1
-
-    # Change compute's port
-    postgres_port = port_distributor.get_port()
-    vanilla_pg.stop().configure([f"port = {postgres_port}"]).start()
-
-    # Try connecting to compute
-    await try_connect()
-    assert wake_compute_called == 2

test_runner/regress/test_remote_storage.py

@@ -83,7 +83,9 @@ def test_remote_storage_backup_and_restore(
     env.pageserver.allowed_errors.append(".*failed to load remote timeline.*")
     # we have a bunch of pytest.raises for these below
     env.pageserver.allowed_errors.append(".*tenant .*? already exists, state:.*")
-    env.pageserver.allowed_errors.append(".*tenant directory already exists.*")
+    env.pageserver.allowed_errors.append(
+        ".*Cannot attach tenant .*?, local tenant directory already exists.*"
+    )
     env.pageserver.allowed_errors.append(".*simulated failure of remote operation.*")
 
     pageserver_http = env.pageserver.http_client()

test_runner/regress/test_tenant_conf.py

@@ -151,7 +151,6 @@ eviction_policy = { "kind" = "LayerAccessThreshold", period = "20s", threshold =
         "eviction_policy": json.dumps(
             {"kind": "LayerAccessThreshold", "period": "80s", "threshold": "42h"}
         ),
-        "max_lsn_wal_lag": "13000000",
     }
     env.neon_cli.config_tenant(
         tenant_id=tenant,
@@ -207,7 +206,6 @@ eviction_policy = { "kind" = "LayerAccessThreshold", period = "20s", threshold =
     assert updated_effective_config["gc_horizon"] == 67108864
     assert updated_effective_config["image_creation_threshold"] == 2
     assert updated_effective_config["pitr_interval"] == "7days"
-    assert updated_effective_config["max_lsn_wal_lag"] == 13000000
 
     # restart the pageserver and ensure that the config is still correct
     env.pageserver.stop()
@@ -267,7 +265,6 @@ eviction_policy = { "kind" = "LayerAccessThreshold", period = "20s", threshold =
         "period": "20s",
         "threshold": "23h",
     }
-    assert final_effective_config["max_lsn_wal_lag"] == 10 * 1024 * 1024
 
     # restart the pageserver and ensure that the config is still correct
     env.pageserver.stop()

test_runner/regress/test_tenant_detach.py

@@ -685,10 +685,12 @@ def test_load_attach_negatives(
 
     pageserver_http.tenant_ignore(tenant_id)
 
-    env.pageserver.allowed_errors.append(".*tenant directory already exists.*")
+    env.pageserver.allowed_errors.append(
+        ".*Cannot attach tenant .*?, local tenant directory already exists.*"
+    )
     with pytest.raises(
         expected_exception=PageserverApiException,
-        match="tenant directory already exists",
+        match=f"Cannot attach tenant {tenant_id}, local tenant directory already exists",
     ):
         pageserver_http.tenant_attach(tenant_id)
 
@@ -732,10 +734,12 @@ def test_ignore_while_attaching(
     pageserver_http.tenant_ignore(tenant_id)
 
     # Cannot attach it due to some local files existing
-    env.pageserver.allowed_errors.append(".*tenant directory already exists.*")
+    env.pageserver.allowed_errors.append(
+        ".*Cannot attach tenant .*?, local tenant directory already exists.*"
+    )
     with pytest.raises(
         expected_exception=PageserverApiException,
-        match="tenant directory already exists",
+        match=f"Cannot attach tenant {tenant_id}, local tenant directory already exists",
     ):
         pageserver_http.tenant_attach(tenant_id)
 

test_runner/regress/test_tenants.py

@@ -217,16 +217,6 @@ def test_metrics_normal_work(neon_env_builder: NeonEnvBuilder):
             labels = ",".join([f'{key}="{value}"' for key, value in sample.labels.items()])
             log.info(f"{sample.name}{{{labels}}} {sample.value}")
 
-    # Test that we gather tenant create metric
-    storage_operation_metrics = [
-        "pageserver_storage_operations_seconds_global_bucket",
-        "pageserver_storage_operations_seconds_global_sum",
-        "pageserver_storage_operations_seconds_global_count",
-    ]
-    for metric in storage_operation_metrics:
-        value = ps_metrics.query_all(metric, filter={"operation": "create tenant"})
-        assert value
-
 
 @pytest.mark.parametrize(
     "remote_storage_kind",

workspace_hack/Cargo.toml

@@ -42,8 +42,7 @@ regex = { version = "1" }
 regex-syntax = { version = "0.6" }
 reqwest = { version = "0.11", default-features = false, features = ["blocking", "json", "multipart", "rustls-tls"] }
 ring = { version = "0.16", features = ["std"] }
-rustls-56bd22fc3884b12 = { package = "rustls", version = "0.20", features = ["dangerous_configuration"] }
-rustls-647d43efb71741da = { package = "rustls", version = "0.21" }
+rustls = { version = "0.20", features = ["dangerous_configuration"] }
 scopeguard = { version = "1" }
 serde = { version = "1", features = ["alloc", "derive"] }
 serde_json = { version = "1", features = ["raw_value"] }