Disable extra jobs

.github/workflows/build_and_test.yml

@@ -314,8 +314,7 @@ jobs:
           test_selection: performance
           run_in_parallel: false
           save_perf_report: ${{ github.ref_name == 'main' }}
-          # test_pageserver_max_throughput_getpage_at_latest_lsn is run in separate workflow periodic_pagebench.yml because it needs snapshots
-          extra_params: --splits 5 --group ${{ matrix.pytest_split_group }} --ignore=test_runner/performance/pageserver/pagebench/test_pageserver_max_throughput_getpage_at_latest_lsn.py
+          extra_params: --splits 5 --group ${{ matrix.pytest_split_group }}
           benchmark_durations: ${{ needs.get-benchmarks-durations.outputs.json }}
           pg_version: v16
           aws-oidc-role-arn: ${{ vars.DEV_AWS_OIDC_ROLE_ARN }}

.github/workflows/ingest_benchmark.yml

@@ -32,31 +32,31 @@ jobs:
       fail-fast: false # allow other variants to continue even if one fails
       matrix:
         include:
-          - target_project: new_empty_project_stripe_size_2048
-            stripe_size: 2048 # 16 MiB
-            postgres_version: 16
-            disable_sharding: false
-          - target_project: new_empty_project_stripe_size_32768
-            stripe_size: 32768 # 256 MiB # note that this is different from null because using null will shard_split the project only if it reaches the threshold
-                               # while here it is sharded from the beginning with a shard size of 256 MiB
-            disable_sharding: false
-            postgres_version: 16
-          - target_project: new_empty_project
-            stripe_size: null # run with neon defaults which will shard split only when reaching the threshold
-            disable_sharding: false
-            postgres_version: 16
+          # - target_project: new_empty_project_stripe_size_2048
+          #   stripe_size: 2048 # 16 MiB
+          #   postgres_version: 16
+          #   disable_sharding: false
+          # - target_project: new_empty_project_stripe_size_32768
+          #   stripe_size: 32768 # 256 MiB # note that this is different from null because using null will shard_split the project only if it reaches the threshold
+          #                      # while here it is sharded from the beginning with a shard size of 256 MiB
+          #   disable_sharding: false
+          #   postgres_version: 16
+          # - target_project: new_empty_project
+          #   stripe_size: null # run with neon defaults which will shard split only when reaching the threshold
+          #   disable_sharding: false
+          #   postgres_version: 16
           - target_project: new_empty_project
             stripe_size: null # run with neon defaults which will shard split only when reaching the threshold
             disable_sharding: false
             postgres_version: 17
-          - target_project: large_existing_project
-            stripe_size: null # cannot re-shared or choose different stripe size for existing, already sharded project
-            disable_sharding: false
-            postgres_version: 16
-          - target_project: new_empty_project_unsharded
-            stripe_size: null # run with neon defaults which will shard split only when reaching the threshold
-            disable_sharding: true
-            postgres_version: 16
+          # - target_project: large_existing_project
+          #   stripe_size: null # cannot re-shared or choose different stripe size for existing, already sharded project
+          #   disable_sharding: false
+          #   postgres_version: 16
+          # - target_project: new_empty_project_unsharded
+          #   stripe_size: null # run with neon defaults which will shard split only when reaching the threshold
+          #   disable_sharding: true
+          #   postgres_version: 16
       max-parallel: 1 # we want to run each stripe size sequentially to be able to compare the results
     permissions:
       contents: write
@@ -115,6 +115,9 @@ jobs:
         stripe_size: ${{ matrix.stripe_size }}
         disable_sharding: ${{ matrix.disable_sharding }}
 
+    - name: Sleep for 10 minutes
+      run: sleep 600
+
     - name: Initialize Neon project
       if: ${{ startsWith(matrix.target_project, 'new_empty_project') }}
       env:

.github/workflows/periodic_pagebench.yml

@@ -1,4 +1,4 @@
-name: Periodic pagebench performance test on unit-perf hetzner runner
+name: Periodic pagebench performance test on dedicated EC2 machine in eu-central-1 region
 
 on:
   schedule:
@@ -8,7 +8,7 @@ on:
     #        │   │ ┌───────────── day of the month (1 - 31)
     #        │   │ │ ┌───────────── month (1 - 12 or JAN-DEC)
     #        │   │ │ │ ┌───────────── day of the week (0 - 6 or SUN-SAT)
-    - cron: '0 */4 * * *' # Runs every 4 hours
+    - cron: '0 */3 * * *' # Runs every 3 hours
   workflow_dispatch: # Allows manual triggering of the workflow
     inputs:
       commit_hash:
@@ -16,11 +16,6 @@ on:
         description: 'The long neon repo commit hash for the system under test (pageserver) to be tested.'
         required: false
         default: ''
-      recreate_snapshots:
-        type: boolean
-        description: 'Recreate snapshots - !!!WARNING!!! We should only recreate snapshots if the previous ones are no longer compatible. Otherwise benchmarking results are not comparable across runs.'
-        required: false
-        default: false
 
 defaults:
   run:
@@ -34,13 +29,13 @@ permissions:
   contents: read
 
 jobs:
-  run_periodic_pagebench_test:
+  trigger_bench_on_ec2_machine_in_eu_central_1:
     permissions:
       id-token: write # aws-actions/configure-aws-credentials
       statuses: write
       contents: write
       pull-requests: write
-    runs-on: [ self-hosted, unit-perf ]
+    runs-on: [ self-hosted, small ]
     container:
       image: ghcr.io/neondatabase/build-tools:pinned-bookworm
       credentials:
@@ -49,13 +44,10 @@ jobs:
       options: --init
     timeout-minutes: 360  # Set the timeout to 6 hours
     env:
+      API_KEY: ${{ secrets.PERIODIC_PAGEBENCH_EC2_RUNNER_API_KEY }}
       RUN_ID: ${{ github.run_id }}
-      DEFAULT_PG_VERSION: 16
-      BUILD_TYPE: release
-      RUST_BACKTRACE: 1
-      # NEON_ENV_BUILDER_USE_OVERLAYFS_FOR_SNAPSHOTS: 1 - doesn't work without root in container
-      S3_BUCKET: neon-github-public-dev
-      PERF_TEST_RESULT_CONNSTR: "${{ secrets.PERF_TEST_RESULT_CONNSTR }}"
+      AWS_DEFAULT_REGION : "eu-central-1"
+      AWS_INSTANCE_ID : "i-02a59a3bf86bc7e74"
     steps:
     # we don't need the neon source code because we run everything remotely
     # however we still need the local github actions to run the allure step below
@@ -64,194 +56,99 @@ jobs:
       with:
         egress-policy: audit
 
-    - name: Set up the environment which depends on $RUNNER_TEMP on nvme drive
-      id: set-env
-      shell: bash -euxo pipefail {0}
-      run: |
-        {
-          echo "NEON_DIR=${RUNNER_TEMP}/neon"
-          echo "NEON_BIN=${RUNNER_TEMP}/neon/bin"
-          echo "POSTGRES_DISTRIB_DIR=${RUNNER_TEMP}/neon/pg_install"
-          echo "LD_LIBRARY_PATH=${RUNNER_TEMP}/neon/pg_install/v${DEFAULT_PG_VERSION}/lib"
-          echo "BACKUP_DIR=${RUNNER_TEMP}/instance_store/saved_snapshots"
-          echo "TEST_OUTPUT=${RUNNER_TEMP}/neon/test_output"
-          echo "PERF_REPORT_DIR=${RUNNER_TEMP}/neon/test_output/perf-report-local"
-          echo "ALLURE_DIR=${RUNNER_TEMP}/neon/test_output/allure-results"
-          echo "ALLURE_RESULTS_DIR=${RUNNER_TEMP}/neon/test_output/allure-results/results"
-        } >> "$GITHUB_ENV"
+    - uses: actions/checkout@11bd71901bbe5b1630ceea73d27597364c9af683 # v4.2.2
 
-        echo "allure_results_dir=${RUNNER_TEMP}/neon/test_output/allure-results/results" >> "$GITHUB_OUTPUT"
+    - name: Show my own (github runner) external IP address - usefull for IP allowlisting
+      run: curl https://ifconfig.me
 
-    - uses: aws-actions/configure-aws-credentials@e3dd6a429d7300a6a4c196c26e071d42e0343502 # v4.0.2
+    - name: Assume AWS OIDC role that allows to manage (start/stop/describe... EC machine)
+      uses: aws-actions/configure-aws-credentials@e3dd6a429d7300a6a4c196c26e071d42e0343502 # v4.0.2
       with:
         aws-region: eu-central-1
-        role-to-assume: ${{ vars.DEV_AWS_OIDC_ROLE_ARN }}
-        role-duration-seconds: 18000 # max 5 hours (needed in case commit hash is still being built)
+        role-to-assume: ${{ vars.DEV_AWS_OIDC_ROLE_MANAGE_BENCHMARK_EC2_VMS_ARN }}
+        role-duration-seconds: 3600
+
+    - name: Start EC2 instance and wait for the instance to boot up
+      run: |
+        aws ec2 start-instances --instance-ids $AWS_INSTANCE_ID
+        aws ec2 wait instance-running --instance-ids $AWS_INSTANCE_ID
+        sleep 60 # sleep some time to allow cloudinit and our API server to start up
+
+    - name: Determine public IP of the EC2 instance and set env variable EC2_MACHINE_URL_US
+      run: |
+        public_ip=$(aws ec2 describe-instances --instance-ids $AWS_INSTANCE_ID --query 'Reservations[*].Instances[*].PublicIpAddress' --output text)
+        echo "Public IP of the EC2 instance: $public_ip"
+        echo "EC2_MACHINE_URL_US=https://${public_ip}:8443" >> $GITHUB_ENV
+
     - name: Determine commit hash
-      id: commit_hash
-      shell: bash -euxo pipefail {0}
       env:
         INPUT_COMMIT_HASH: ${{ github.event.inputs.commit_hash }}
       run: |
-        if [[ -z "${INPUT_COMMIT_HASH}" ]]; then
-          COMMIT_HASH=$(curl -s https://api.github.com/repos/neondatabase/neon/commits/main | jq -r '.sha')
-          echo "COMMIT_HASH=$COMMIT_HASH" >> $GITHUB_ENV
-          echo "commit_hash=$COMMIT_HASH" >> "$GITHUB_OUTPUT"
+        if [ -z "$INPUT_COMMIT_HASH" ]; then
+          echo "COMMIT_HASH=$(curl -s https://api.github.com/repos/neondatabase/neon/commits/main | jq -r '.sha')" >> $GITHUB_ENV
           echo "COMMIT_HASH_TYPE=latest" >> $GITHUB_ENV
         else
-          COMMIT_HASH="${INPUT_COMMIT_HASH}"
-          echo "COMMIT_HASH=$COMMIT_HASH" >> $GITHUB_ENV
-          echo "commit_hash=$COMMIT_HASH" >> "$GITHUB_OUTPUT"
+          echo "COMMIT_HASH=$INPUT_COMMIT_HASH" >> $GITHUB_ENV
           echo "COMMIT_HASH_TYPE=manual" >> $GITHUB_ENV
         fi
-    - name: Checkout the neon repository at given commit hash
-      uses: actions/checkout@11bd71901bbe5b1630ceea73d27597364c9af683 # v4.2.2
-      with:
-        ref: ${{ steps.commit_hash.outputs.commit_hash }}
 
-    # does not reuse ./.github/actions/download because we need to download the artifact for the given commit hash
-    # example artifact
-    # s3://neon-github-public-dev/artifacts/48b870bc078bd2c450eb7b468e743b9c118549bf/15036827400/1/neon-Linux-X64-release-artifact.tar.zst /instance_store/artifacts/neon-Linux-release-artifact.tar.zst
-    - name: Determine artifact S3_KEY for given commit hash and download and extract artifact
-      id: artifact_prefix
-      shell: bash -euxo pipefail {0}
-      env:
-        ARCHIVE: ${{ runner.temp }}/downloads/neon-${{ runner.os }}-${{ runner.arch }}-release-artifact.tar.zst
-        COMMIT_HASH: ${{ env.COMMIT_HASH }}
-        COMMIT_HASH_TYPE: ${{ env.COMMIT_HASH_TYPE }}
+    - name: Start Bench with run_id
       run: |
-        attempt=0
-        max_attempts=24 # 5 minutes * 24 = 2 hours
+        curl -k -X 'POST' \
+        "${EC2_MACHINE_URL_US}/start_test/${GITHUB_RUN_ID}" \
+        -H 'accept: application/json' \
+        -H 'Content-Type: application/json' \
+        -H "Authorization: Bearer $API_KEY" \
+        -d "{\"neonRepoCommitHash\": \"${COMMIT_HASH}\", \"neonRepoCommitHashType\": \"${COMMIT_HASH_TYPE}\"}"
 
-        while [[ $attempt -lt $max_attempts ]]; do
-          # the following command will fail until the artifacts are available ...
-          S3_KEY=$(aws s3api list-objects-v2 --bucket "$S3_BUCKET" --prefix "artifacts/$COMMIT_HASH/" \
-            | jq -r '.Contents[]?.Key' \
-            | grep "neon-${{ runner.os }}-${{ runner.arch }}-release-artifact.tar.zst" \
-            | sort --version-sort \
-            | tail -1) || true # ... thus ignore errors from the command
-          if [[ -n "${S3_KEY}" ]]; then
-            echo "Artifact found: $S3_KEY"
-            echo "S3_KEY=$S3_KEY" >> $GITHUB_ENV
+    - name: Poll Test Status
+      id: poll_step
+      run: |
+        status=""
+        while [[ "$status" != "failure" && "$status" != "success" ]]; do
+          response=$(curl -k -X 'GET' \
+          "${EC2_MACHINE_URL_US}/test_status/${GITHUB_RUN_ID}" \
+          -H 'accept: application/json' \
+          -H "Authorization: Bearer $API_KEY")
+          echo "Response: $response"
+          set +x
+          status=$(echo $response | jq -r '.status')
+          echo "Test status: $status"
+          if [[ "$status" == "failure" ]]; then
+            echo "Test failed"
+            exit 1 # Fail the job step if status is failure
+          elif [[ "$status" == "success" || "$status" == "null" ]]; then
             break
+          elif [[ "$status" == "too_many_runs" ]]; then
+            echo "Too many runs already running"
+            echo "too_many_runs=true" >> "$GITHUB_OUTPUT"
+            exit 1
           fi
-          
-          # Increment attempt counter and sleep for 5 minutes
-          attempt=$((attempt + 1))
-          echo "Attempt $attempt of $max_attempts to find artifacts in S3 bucket s3://$S3_BUCKET/artifacts/$COMMIT_HASH failed. Retrying in 5 minutes..."
-          sleep 300 # Sleep for 5 minutes
+
+          sleep 60 # Poll every 60 seconds
         done
 
-        if [[ -z "${S3_KEY}" ]]; then
-          echo "Error: artifact not found in S3 bucket s3://$S3_BUCKET/artifacts/$COMMIT_HASH" after 2 hours
-        else
-          mkdir -p $(dirname $ARCHIVE)
-          time aws s3 cp --only-show-errors s3://$S3_BUCKET/${S3_KEY} ${ARCHIVE}
-          mkdir -p ${NEON_DIR}
-          time tar -xf ${ARCHIVE} -C ${NEON_DIR}
-          rm -f ${ARCHIVE}
-        fi
-
-    - name: Download snapshots from S3
-      if: ${{ github.event_name != 'workflow_dispatch' || github.event.inputs.recreate_snapshots == 'false' || github.event.inputs.recreate_snapshots == '' }}
-      id: download_snapshots
-      shell: bash -euxo pipefail {0}
+    - name: Retrieve Test Logs
+      if: always() && steps.poll_step.outputs.too_many_runs != 'true'
       run: |
-        # Download the snapshots from S3
-        mkdir -p ${TEST_OUTPUT}
-        mkdir -p $BACKUP_DIR
-        cd $BACKUP_DIR
-        mkdir parts
-        cd parts
-        PART=$(aws s3api list-objects-v2 --bucket $S3_BUCKET --prefix performance/pagebench/ \
-          | jq -r '.Contents[]?.Key' \
-          | grep -E 'shared-snapshots-[0-9]{4}-[0-9]{2}-[0-9]{2}' \
-          | sort \
-          | tail -1)
-        echo "Latest PART: $PART"
-        if [[ -z "$PART" ]]; then
-          echo "ERROR: No matching S3 key found" >&2
-          exit 1
-        fi
-        S3_KEY=$(dirname $PART)
-        time aws s3 cp --only-show-errors --recursive s3://${S3_BUCKET}/$S3_KEY/ .
-        cd $TEST_OUTPUT
-        time cat $BACKUP_DIR/parts/* | zstdcat | tar --extract --preserve-permissions
-        rm -rf ${BACKUP_DIR}
+        curl -k -X 'GET' \
+        "${EC2_MACHINE_URL_US}/test_log/${GITHUB_RUN_ID}" \
+        -H 'accept: application/gzip' \
+        -H "Authorization: Bearer $API_KEY" \
+        --output "test_log_${GITHUB_RUN_ID}.gz"
 
-    - name: Cache poetry deps
-      uses: actions/cache@v4
-      with:
-        path: ~/.cache/pypoetry/virtualenvs
-        key: v2-${{ runner.os }}-${{ runner.arch }}-python-deps-bookworm-${{ hashFiles('poetry.lock') }}
-
-    - name: Install Python deps
-      shell: bash -euxo pipefail {0}
-      run: ./scripts/pysync
-
-    # we need high number of open files for pagebench
-    - name: show ulimits
-      shell: bash -euxo pipefail {0}
+    - name: Unzip Test Log and Print it into this job's log
+      if: always() && steps.poll_step.outputs.too_many_runs != 'true'
       run: |
-        ulimit -a
-
-    - name: Run pagebench testcase
-      shell: bash -euxo pipefail {0}
-      env:
-        CI: false  # need to override this env variable set by github to enforce using snapshots
-      run: |
-        export PLATFORM=hetzner-unit-perf-${COMMIT_HASH_TYPE}
-        # report the commit hash of the neon repository in the revision of the test results
-        export GITHUB_SHA=${COMMIT_HASH}
-        rm -rf ${PERF_REPORT_DIR}
-        rm -rf ${ALLURE_RESULTS_DIR}
-        mkdir -p ${PERF_REPORT_DIR}
-        mkdir -p ${ALLURE_RESULTS_DIR}
-        PARAMS="--alluredir=${ALLURE_RESULTS_DIR} --tb=short --verbose -rA"
-        EXTRA_PARAMS="--out-dir ${PERF_REPORT_DIR} --durations-path $TEST_OUTPUT/benchmark_durations.json"
-        # run only two selected tests
-        # environment set by parent:
-        # RUST_BACKTRACE=1 DEFAULT_PG_VERSION=16 BUILD_TYPE=release
-        ./scripts/pytest ${PARAMS} test_runner/performance/pageserver/pagebench/test_pageserver_max_throughput_getpage_at_latest_lsn.py::test_pageserver_characterize_throughput_with_n_tenants ${EXTRA_PARAMS}
-        ./scripts/pytest ${PARAMS} test_runner/performance/pageserver/pagebench/test_pageserver_max_throughput_getpage_at_latest_lsn.py::test_pageserver_characterize_latencies_with_1_client_and_throughput_with_many_clients_one_tenant ${EXTRA_PARAMS}
-
-    - name: upload the performance metrics to the Neon performance database which is used by grafana dashboards to display the results
-      shell: bash -euxo pipefail {0}
-      run: |
-        export REPORT_FROM="$PERF_REPORT_DIR"
-        export GITHUB_SHA=${COMMIT_HASH}
-        time ./scripts/generate_and_push_perf_report.sh
-
-    - name: Upload test results
-      if: ${{ !cancelled() }}
-      uses: ./.github/actions/allure-report-store
-      with:
-        report-dir:  ${{ steps.set-env.outputs.allure_results_dir }}
-        unique-key: ${{ env.BUILD_TYPE }}-${{ env.DEFAULT_PG_VERSION }}-${{ runner.arch }}
-        aws-oidc-role-arn:  ${{ vars.DEV_AWS_OIDC_ROLE_ARN }}
+        gzip -d "test_log_${GITHUB_RUN_ID}.gz"
+        cat "test_log_${GITHUB_RUN_ID}"
 
     - name: Create Allure report
-      id: create-allure-report
       if: ${{ !cancelled() }}
       uses: ./.github/actions/allure-report-generate
       with:
         aws-oidc-role-arn: ${{ vars.DEV_AWS_OIDC_ROLE_ARN }}
 
-    - name: Upload snapshots
-      if: ${{ github.event_name == 'workflow_dispatch' && github.event.inputs.recreate_snapshots != 'false' && github.event.inputs.recreate_snapshots != '' }}
-      id: upload_snapshots
-      shell: bash -euxo pipefail {0}
-      run: |
-        mkdir -p $BACKUP_DIR
-        cd $TEST_OUTPUT
-        tar --create --preserve-permissions --file - shared-snapshots | zstd -o $BACKUP_DIR/shared_snapshots.tar.zst
-        cd $BACKUP_DIR
-        mkdir parts
-        split -b 1G shared_snapshots.tar.zst ./parts/shared_snapshots.tar.zst.part.
-        SNAPSHOT_DATE=$(date +%F)  # YYYY-MM-DD
-        cd parts
-        time aws s3 cp --recursive . s3://${S3_BUCKET}/performance/pagebench/shared-snapshots-${SNAPSHOT_DATE}/
-
     - name: Post to a Slack channel
       if: ${{ github.event.schedule && failure() }}
       uses: slackapi/slack-github-action@fcfb566f8b0aab22203f066d80ca1d7e4b5d05b3 # v1.27.1
@@ -260,22 +157,26 @@ jobs:
         slack-message: "Periodic pagebench testing on dedicated hardware: ${{ job.status }}\n${{ github.server_url }}/${{ github.repository }}/actions/runs/${{ github.run_id }}"
       env:
         SLACK_BOT_TOKEN: ${{ secrets.SLACK_BOT_TOKEN }}
-        
+
     - name: Cleanup Test Resources
       if: always()
-      shell: bash -euxo pipefail {0}
-      env:
-        ARCHIVE: ${{ runner.temp }}/downloads/neon-${{ runner.os }}-${{ runner.arch }}-release-artifact.tar.zst
       run: |
-        # Cleanup the test resources
-        if [[ -d "${BACKUP_DIR}" ]]; then
-          rm -rf ${BACKUP_DIR}
-        fi
-        if [[ -d "${TEST_OUTPUT}" ]]; then
-          rm -rf ${TEST_OUTPUT}
-        fi
-        if [[ -d "${NEON_DIR}" ]]; then
-          rm -rf ${NEON_DIR}
-        fi
-        rm -rf $(dirname $ARCHIVE)
+        curl -k -X 'POST' \
+        "${EC2_MACHINE_URL_US}/cleanup_test/${GITHUB_RUN_ID}" \
+        -H 'accept: application/json' \
+        -H "Authorization: Bearer $API_KEY" \
+        -d ''
 
+    - name: Assume AWS OIDC role that allows to manage (start/stop/describe... EC machine)
+      if: always() && steps.poll_step.outputs.too_many_runs != 'true'
+      uses: aws-actions/configure-aws-credentials@e3dd6a429d7300a6a4c196c26e071d42e0343502 # v4.0.2
+      with:
+        aws-region: eu-central-1
+        role-to-assume: ${{ vars.DEV_AWS_OIDC_ROLE_MANAGE_BENCHMARK_EC2_VMS_ARN }}
+        role-duration-seconds: 3600
+
+    - name: Stop EC2 instance and wait for the instance to be stopped
+      if: always() && steps.poll_step.outputs.too_many_runs != 'true'
+      run: |
+        aws ec2 stop-instances --instance-ids $AWS_INSTANCE_ID
+        aws ec2 wait instance-stopped --instance-ids $AWS_INSTANCE_ID

Cargo.lock

@@ -1276,7 +1276,7 @@ version = "0.1.0"
 dependencies = [
  "anyhow",
  "chrono",
- "indexmap 2.9.0",
+ "indexmap 2.0.1",
  "jsonwebtoken",
  "regex",
  "remote_storage",
@@ -1308,7 +1308,7 @@ dependencies = [
  "flate2",
  "futures",
  "http 1.1.0",
- "indexmap 2.9.0",
+ "indexmap 2.0.1",
  "itertools 0.10.5",
  "jsonwebtoken",
  "metrics",
@@ -2597,7 +2597,7 @@ dependencies = [
  "futures-sink",
  "futures-util",
  "http 0.2.9",
- "indexmap 2.9.0",
+ "indexmap 2.0.1",
  "slab",
  "tokio",
  "tokio-util",
@@ -2616,7 +2616,7 @@ dependencies = [
  "futures-sink",
  "futures-util",
  "http 1.1.0",
- "indexmap 2.9.0",
+ "indexmap 2.0.1",
  "slab",
  "tokio",
  "tokio-util",
@@ -2863,14 +2863,14 @@ dependencies = [
  "pprof",
  "regex",
  "routerify",
- "rustls 0.23.27",
+ "rustls 0.23.18",
  "rustls-pemfile 2.1.1",
  "serde",
  "serde_json",
  "serde_path_to_error",
  "thiserror 1.0.69",
  "tokio",
- "tokio-rustls 0.26.2",
+ "tokio-rustls 0.26.0",
  "tokio-stream",
  "tokio-util",
  "tracing",
@@ -3200,12 +3200,12 @@ dependencies = [
 
 [[package]]
 name = "indexmap"
-version = "2.9.0"
+version = "2.0.1"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "cea70ddb795996207ad57735b50c5982d8844f38ba9ee5f1aedcfb708a2aa11e"
+checksum = "ad227c3af19d4914570ad36d30409928b75967c298feb9ea1969db3a610bb14e"
 dependencies = [
  "equivalent",
- "hashbrown 0.15.2",
+ "hashbrown 0.14.5",
  "serde",
 ]
 
@@ -3228,7 +3228,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "232929e1d75fe899576a3d5c7416ad0d88dbfbb3c3d6aa00873a7408a50ddb88"
 dependencies = [
  "ahash",
- "indexmap 2.9.0",
+ "indexmap 2.0.1",
  "is-terminal",
  "itoa",
  "log",
@@ -3251,7 +3251,7 @@ dependencies = [
  "crossbeam-utils",
  "dashmap 6.1.0",
  "env_logger",
- "indexmap 2.9.0",
+ "indexmap 2.0.1",
  "itoa",
  "log",
  "num-format",
@@ -4112,7 +4112,7 @@ dependencies = [
  "opentelemetry-http",
  "opentelemetry-proto",
  "opentelemetry_sdk",
- "prost 0.13.5",
+ "prost 0.13.3",
  "reqwest",
  "thiserror 1.0.69",
 ]
@@ -4125,8 +4125,8 @@ checksum = "a6e05acbfada5ec79023c85368af14abd0b307c015e9064d249b2a950ef459a6"
 dependencies = [
  "opentelemetry",
  "opentelemetry_sdk",
- "prost 0.13.5",
- "tonic 0.12.3",
+ "prost 0.13.3",
+ "tonic",
 ]
 
 [[package]]
@@ -4321,7 +4321,6 @@ dependencies = [
  "pageserver_api",
  "pageserver_client",
  "pageserver_compaction",
- "pageserver_page_api",
  "pem",
  "pin-project-lite",
  "postgres-protocol",
@@ -4340,7 +4339,7 @@ dependencies = [
  "reqwest",
  "rpds",
  "rstest",
- "rustls 0.23.27",
+ "rustls 0.23.18",
  "scopeguard",
  "send-future",
  "serde",
@@ -4359,13 +4358,11 @@ dependencies = [
  "tokio-epoll-uring",
  "tokio-io-timeout",
  "tokio-postgres",
- "tokio-rustls 0.26.2",
+ "tokio-rustls 0.26.0",
  "tokio-stream",
  "tokio-tar",
  "tokio-util",
  "toml_edit",
- "tonic 0.13.1",
- "tonic-reflection",
  "tracing",
  "tracing-utils",
  "twox-hash",
@@ -4458,8 +4455,8 @@ dependencies = [
 name = "pageserver_page_api"
 version = "0.1.0"
 dependencies = [
- "prost 0.13.5",
- "tonic 0.13.1",
+ "prost 0.13.3",
+ "tonic",
  "tonic-build",
  "workspace_hack",
 ]
@@ -4840,14 +4837,14 @@ dependencies = [
  "bytes",
  "once_cell",
  "pq_proto",
- "rustls 0.23.27",
+ "rustls 0.23.18",
  "rustls-pemfile 2.1.1",
  "serde",
  "thiserror 1.0.69",
  "tokio",
  "tokio-postgres",
  "tokio-postgres-rustls",
- "tokio-rustls 0.26.2",
+ "tokio-rustls 0.26.0",
  "tokio-util",
  "tracing",
 ]
@@ -4954,7 +4951,7 @@ dependencies = [
  "inferno 0.12.0",
  "num",
  "paste",
- "prost 0.13.5",
+ "prost 0.13.3",
 ]
 
 [[package]]
@@ -5059,12 +5056,12 @@ dependencies = [
 
 [[package]]
 name = "prost"
-version = "0.13.5"
+version = "0.13.3"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "2796faa41db3ec313a31f7624d9286acf277b52de526150b7e69f3debf891ee5"
+checksum = "7b0487d90e047de87f984913713b85c601c05609aad5b0df4b4573fbf69aa13f"
 dependencies = [
  "bytes",
- "prost-derive 0.13.5",
+ "prost-derive 0.13.3",
 ]
 
 [[package]]
@@ -5102,7 +5099,7 @@ dependencies = [
  "once_cell",
  "petgraph",
  "prettyplease",
- "prost 0.13.5",
+ "prost 0.13.3",
  "prost-types 0.13.3",
  "regex",
  "syn 2.0.100",
@@ -5124,9 +5121,9 @@ dependencies = [
 
 [[package]]
 name = "prost-derive"
-version = "0.13.5"
+version = "0.13.3"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "8a56d757972c98b346a9b766e3f02746cde6dd1cd1d1d563472929fdd74bec4d"
+checksum = "e9552f850d5f0964a4e4d0bf306459ac29323ddfbae05e35a7c0d35cb0803cc5"
 dependencies = [
  "anyhow",
  "itertools 0.12.1",
@@ -5150,7 +5147,7 @@ version = "0.13.3"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "4759aa0d3a6232fb8dbdb97b61de2c20047c68aca932c7ed76da9d788508d670"
 dependencies = [
- "prost 0.13.5",
+ "prost 0.13.3",
 ]
 
 [[package]]
@@ -5198,7 +5195,7 @@ dependencies = [
  "hyper 0.14.30",
  "hyper 1.4.1",
  "hyper-util",
- "indexmap 2.9.0",
+ "indexmap 2.0.1",
  "ipnet",
  "itertools 0.10.5",
  "itoa",
@@ -5232,7 +5229,7 @@ dependencies = [
  "rsa",
  "rstest",
  "rustc-hash 1.1.0",
- "rustls 0.23.27",
+ "rustls 0.23.18",
  "rustls-native-certs 0.8.0",
  "rustls-pemfile 2.1.1",
  "scopeguard",
@@ -5251,7 +5248,7 @@ dependencies = [
  "tokio",
  "tokio-postgres",
  "tokio-postgres2",
- "tokio-rustls 0.26.2",
+ "tokio-rustls 0.26.0",
  "tokio-tungstenite 0.21.0",
  "tokio-util",
  "tracing",
@@ -5475,13 +5472,13 @@ dependencies = [
  "num-bigint",
  "percent-encoding",
  "pin-project-lite",
- "rustls 0.23.27",
+ "rustls 0.23.18",
  "rustls-native-certs 0.8.0",
  "ryu",
  "sha1_smol",
  "socket2",
  "tokio",
- "tokio-rustls 0.26.2",
+ "tokio-rustls 0.26.0",
  "tokio-util",
  "url",
 ]
@@ -5929,15 +5926,15 @@ dependencies = [
 
 [[package]]
 name = "rustls"
-version = "0.23.27"
+version = "0.23.18"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "730944ca083c1c233a75c09f199e973ca499344a2b7ba9e755c457e86fb4a321"
+checksum = "9c9cc1d47e243d655ace55ed38201c19ae02c148ae56412ab8750e8f0166ab7f"
 dependencies = [
  "log",
  "once_cell",
  "ring",
  "rustls-pki-types",
- "rustls-webpki 0.103.3",
+ "rustls-webpki 0.102.8",
  "subtle",
  "zeroize",
 ]
@@ -6026,17 +6023,6 @@ dependencies = [
  "untrusted",
 ]
 
-[[package]]
-name = "rustls-webpki"
-version = "0.103.3"
-source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "e4a72fe2bcf7a6ac6fd7d0b9e5cb68aeb7d4c0a0271730218b3e92d43b4eb435"
-dependencies = [
- "ring",
- "rustls-pki-types",
- "untrusted",
-]
-
 [[package]]
 name = "rustversion"
 version = "1.0.12"
@@ -6088,7 +6074,7 @@ dependencies = [
  "regex",
  "remote_storage",
  "reqwest",
- "rustls 0.23.27",
+ "rustls 0.23.18",
  "safekeeper_api",
  "safekeeper_client",
  "scopeguard",
@@ -6105,7 +6091,7 @@ dependencies = [
  "tokio",
  "tokio-io-timeout",
  "tokio-postgres",
- "tokio-rustls 0.26.2",
+ "tokio-rustls 0.26.0",
  "tokio-stream",
  "tokio-tar",
  "tokio-util",
@@ -6277,7 +6263,7 @@ checksum = "255914a8e53822abd946e2ce8baa41d4cded6b8e938913b7f7b9da5b7ab44335"
 dependencies = [
  "httpdate",
  "reqwest",
- "rustls 0.23.27",
+ "rustls 0.23.18",
  "sentry-backtrace",
  "sentry-contexts",
  "sentry-core",
@@ -6706,11 +6692,11 @@ dependencies = [
  "metrics",
  "once_cell",
  "parking_lot 0.12.1",
- "prost 0.13.5",
- "rustls 0.23.27",
+ "prost 0.13.3",
+ "rustls 0.23.18",
  "tokio",
- "tokio-rustls 0.26.2",
- "tonic 0.13.1",
+ "tokio-rustls 0.26.0",
+ "tonic",
  "tonic-build",
  "tracing",
  "utils",
@@ -6752,7 +6738,7 @@ dependencies = [
  "regex",
  "reqwest",
  "routerify",
- "rustls 0.23.27",
+ "rustls 0.23.18",
  "rustls-native-certs 0.8.0",
  "safekeeper_api",
  "safekeeper_client",
@@ -6767,7 +6753,7 @@ dependencies = [
  "tokio",
  "tokio-postgres",
  "tokio-postgres-rustls",
- "tokio-rustls 0.26.2",
+ "tokio-rustls 0.26.0",
  "tokio-util",
  "tracing",
  "utils",
@@ -6805,7 +6791,7 @@ dependencies = [
  "postgres_ffi",
  "remote_storage",
  "reqwest",
- "rustls 0.23.27",
+ "rustls 0.23.18",
  "rustls-native-certs 0.8.0",
  "serde",
  "serde_json",
@@ -7339,10 +7325,10 @@ source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "04fb792ccd6bbcd4bba408eb8a292f70fc4a3589e5d793626f45190e6454b6ab"
 dependencies = [
  "ring",
- "rustls 0.23.27",
+ "rustls 0.23.18",
  "tokio",
  "tokio-postgres",
- "tokio-rustls 0.26.2",
+ "tokio-rustls 0.26.0",
  "x509-certificate",
 ]
 
@@ -7386,11 +7372,12 @@ dependencies = [
 
 [[package]]
 name = "tokio-rustls"
-version = "0.26.2"
+version = "0.26.0"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "8e727b36a1a0e8b74c376ac2211e40c2c8af09fb4013c60d910495810f008e9b"
+checksum = "0c7bc40d0e5a97695bb96e27995cd3a08538541b0a846f65bba7a359f36700d4"
 dependencies = [
- "rustls 0.23.27",
+ "rustls 0.23.18",
+ "rustls-pki-types",
  "tokio",
 ]
 
@@ -7488,7 +7475,7 @@ version = "0.22.14"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "f21c7aaf97f1bd9ca9d4f9e73b0a6c74bd5afef56f2bc931943a6e1c37e04e38"
 dependencies = [
- "indexmap 2.9.0",
+ "indexmap 2.0.1",
  "serde",
  "serde_spanned",
  "toml_datetime",
@@ -7507,41 +7494,18 @@ dependencies = [
  "http 1.1.0",
  "http-body 1.0.0",
  "http-body-util",
- "percent-encoding",
- "pin-project",
- "prost 0.13.5",
- "tokio-stream",
- "tower-layer",
- "tower-service",
- "tracing",
-]
-
-[[package]]
-name = "tonic"
-version = "0.13.1"
-source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "7e581ba15a835f4d9ea06c55ab1bd4dce26fc53752c69a04aac00703bfb49ba9"
-dependencies = [
- "async-trait",
- "axum",
- "base64 0.22.1",
- "bytes",
- "h2 0.4.4",
- "http 1.1.0",
- "http-body 1.0.0",
- "http-body-util",
  "hyper 1.4.1",
  "hyper-timeout",
  "hyper-util",
  "percent-encoding",
  "pin-project",
- "prost 0.13.5",
+ "prost 0.13.3",
  "rustls-native-certs 0.8.0",
- "socket2",
+ "rustls-pemfile 2.1.1",
  "tokio",
- "tokio-rustls 0.26.2",
+ "tokio-rustls 0.26.0",
  "tokio-stream",
- "tower 0.5.2",
+ "tower 0.4.13",
  "tower-layer",
  "tower-service",
  "tracing",
@@ -7549,9 +7513,9 @@ dependencies = [
 
 [[package]]
 name = "tonic-build"
-version = "0.13.1"
+version = "0.12.3"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "eac6f67be712d12f0b41328db3137e0d0757645d8904b4cb7d51cd9c2279e847"
+checksum = "9557ce109ea773b399c9b9e5dca39294110b74f1f342cb347a80d1fce8c26a11"
 dependencies = [
  "prettyplease",
  "proc-macro2",
@@ -7561,19 +7525,6 @@ dependencies = [
  "syn 2.0.100",
 ]
 
-[[package]]
-name = "tonic-reflection"
-version = "0.13.1"
-source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "f9687bd5bfeafebdded2356950f278bba8226f0b32109537c4253406e09aafe1"
-dependencies = [
- "prost 0.13.5",
- "prost-types 0.13.3",
- "tokio",
- "tokio-stream",
- "tonic 0.13.1",
-]
-
 [[package]]
 name = "tower"
 version = "0.4.13"
@@ -7582,11 +7533,16 @@ checksum = "b8fa9be0de6cf49e536ce1851f987bd21a43b771b09473c3549a6c853db37c1c"
 dependencies = [
  "futures-core",
  "futures-util",
+ "indexmap 1.9.3",
  "pin-project",
  "pin-project-lite",
+ "rand 0.8.5",
+ "slab",
  "tokio",
+ "tokio-util",
  "tower-layer",
  "tower-service",
+ "tracing",
 ]
 
 [[package]]
@@ -7597,12 +7553,9 @@ checksum = "d039ad9159c98b70ecfd540b2573b97f7f52c3e8d9f8ad57a24b916a536975f9"
 dependencies = [
  "futures-core",
  "futures-util",
- "indexmap 2.9.0",
  "pin-project-lite",
- "slab",
  "sync_wrapper 1.0.1",
  "tokio",
- "tokio-util",
  "tower-layer",
  "tower-service",
  "tracing",
@@ -7930,7 +7883,7 @@ dependencies = [
  "base64 0.22.1",
  "log",
  "once_cell",
- "rustls 0.23.27",
+ "rustls 0.23.18",
  "rustls-pki-types",
  "url",
  "webpki-roots",
@@ -8125,7 +8078,7 @@ dependencies = [
  "pageserver_api",
  "postgres_ffi",
  "pprof",
- "prost 0.13.5",
+ "prost 0.13.3",
  "remote_storage",
  "serde",
  "serde_json",
@@ -8545,8 +8498,6 @@ dependencies = [
  "ahash",
  "anstream",
  "anyhow",
- "axum",
- "axum-core",
  "base64 0.13.1",
  "base64 0.21.7",
  "base64ct",
@@ -8583,7 +8534,8 @@ dependencies = [
  "hyper 0.14.30",
  "hyper 1.4.1",
  "hyper-util",
- "indexmap 2.9.0",
+ "indexmap 1.9.3",
+ "indexmap 2.0.1",
  "itertools 0.12.1",
  "lazy_static",
  "libc",
@@ -8605,16 +8557,16 @@ dependencies = [
  "percent-encoding",
  "prettyplease",
  "proc-macro2",
- "prost 0.13.5",
+ "prost 0.13.3",
  "quote",
  "rand 0.8.5",
  "regex",
  "regex-automata 0.4.3",
  "regex-syntax 0.8.2",
  "reqwest",
- "rustls 0.23.27",
+ "rustls 0.23.18",
  "rustls-pki-types",
- "rustls-webpki 0.103.3",
+ "rustls-webpki 0.102.8",
  "scopeguard",
  "sec1 0.7.3",
  "serde",
@@ -8632,11 +8584,12 @@ dependencies = [
  "time",
  "time-macros",
  "tokio",
- "tokio-rustls 0.26.2",
+ "tokio-rustls 0.26.0",
  "tokio-stream",
  "tokio-util",
  "toml_edit",
- "tower 0.5.2",
+ "tonic",
+ "tower 0.4.13",
  "tracing",
  "tracing-core",
  "tracing-log",

Cargo.toml

@@ -149,7 +149,7 @@ pin-project-lite = "0.2"
 pprof = { version = "0.14", features = ["criterion", "flamegraph", "frame-pointer", "prost-codec"] }
 procfs = "0.16"
 prometheus = {version = "0.13", default-features=false, features = ["process"]} # removes protobuf dependency
-prost = "0.13.5"
+prost = "0.13"
 rand = "0.8"
 redis = { version = "0.29.2", features = ["tokio-rustls-comp", "keep-alive"] }
 regex = "1.10.2"
@@ -199,8 +199,7 @@ tokio-tar = "0.3"
 tokio-util = { version = "0.7.10", features = ["io", "rt"] }
 toml = "0.8"
 toml_edit = "0.22"
-tonic = { version = "0.13.1", default-features = false, features = ["channel", "codegen", "prost", "router", "server", "tls-ring", "tls-native-roots"] }
-tonic-reflection = { version = "0.13.1", features = ["server"] }
+tonic = {version = "0.12.3", default-features = false, features = ["channel", "tls", "tls-roots"]}
 tower = { version = "0.5.2", default-features = false }
 tower-http = { version = "0.6.2", features = ["auth", "request-id", "trace"] }
 
@@ -281,7 +280,7 @@ criterion = "0.5.1"
 rcgen = "0.13"
 rstest = "0.18"
 camino-tempfile = "1.0.2"
-tonic-build = "0.13.1"
+tonic-build = "0.12"
 
 [patch.crates-io]
 

build-tools.Dockerfile

@@ -155,7 +155,7 @@ RUN set -e \
 
 # Keep the version the same as in compute/compute-node.Dockerfile and
 # test_runner/regress/test_compute_metrics.py.
-ENV SQL_EXPORTER_VERSION=0.17.3
+ENV SQL_EXPORTER_VERSION=0.17.0
 RUN curl -fsSL \
     "https://github.com/burningalchemist/sql_exporter/releases/download/${SQL_EXPORTER_VERSION}/sql_exporter-${SQL_EXPORTER_VERSION}.linux-$(case "$(uname -m)" in x86_64) echo amd64;; aarch64) echo arm64;; esac).tar.gz" \
     --output sql_exporter.tar.gz \

compute/compute-node.Dockerfile

@@ -582,38 +582,6 @@ RUN make -j $(getconf _NPROCESSORS_ONLN) && \
     make -j $(getconf _NPROCESSORS_ONLN) install && \
     echo 'trusted = true' >> /usr/local/pgsql/share/extension/hypopg.control
 
-#########################################################################################
-#
-# Layer "online_advisor-build"
-# compile online_advisor extension
-#
-#########################################################################################
-FROM build-deps AS online_advisor-src
-ARG PG_VERSION
-
-# online_advisor supports all Postgres version starting from PG14, but prior to PG17 has to be included in preload_shared_libraries
-# last release 1.0 - May 15, 2025
-WORKDIR /ext-src
-RUN case "${PG_VERSION:?}" in \
-    "v17") \
-        ;; \
-    *) \
-        echo "skipping the version of online_advistor for $PG_VERSION" && exit 0 \
-        ;; \
-    esac && \
-	wget https://github.com/knizhnik/online_advisor/archive/refs/tags/1.0.tar.gz -O online_advisor.tar.gz && \
-    echo "059b7d9e5a90013a58bdd22e9505b88406ce05790675eb2d8434e5b215652d54 online_advisor.tar.gz" | sha256sum --check && \
-    mkdir online_advisor-src && cd online_advisor-src && tar xzf ../online_advisor.tar.gz --strip-components=1 -C .
-
-FROM pg-build AS online_advisor-build
-COPY --from=online_advisor-src /ext-src/ /ext-src/
-WORKDIR /ext-src/
-RUN if [ -d online_advisor-src ]; then \
-	    cd online_advisor-src && \
-        make -j install && \
-        echo 'trusted = true' >> /usr/local/pgsql/share/extension/online_advisor.control; \
-    fi
-
 #########################################################################################
 #
 # Layer "pg_hashids-build"
@@ -1680,7 +1648,6 @@ COPY --from=pg_jsonschema-build /usr/local/pgsql/ /usr/local/pgsql/
 COPY --from=pg_graphql-build /usr/local/pgsql/ /usr/local/pgsql/
 COPY --from=pg_tiktoken-build /usr/local/pgsql/ /usr/local/pgsql/
 COPY --from=hypopg-build /usr/local/pgsql/ /usr/local/pgsql/
-COPY --from=online_advisor-build /usr/local/pgsql/ /usr/local/pgsql/
 COPY --from=pg_hashids-build /usr/local/pgsql/ /usr/local/pgsql/
 COPY --from=rum-build /usr/local/pgsql/ /usr/local/pgsql/
 COPY --from=pgtap-build /usr/local/pgsql/ /usr/local/pgsql/
@@ -1784,17 +1751,17 @@ ARG TARGETARCH
 RUN if [ "$TARGETARCH" = "amd64" ]; then\
         postgres_exporter_sha256='59aa4a7bb0f7d361f5e05732f5ed8c03cc08f78449cef5856eadec33a627694b';\
         pgbouncer_exporter_sha256='c9f7cf8dcff44f0472057e9bf52613d93f3ffbc381ad7547a959daa63c5e84ac';\
-        sql_exporter_sha256='9a41127a493e8bfebfe692bf78c7ed2872a58a3f961ee534d1b0da9ae584aaab';\
+        sql_exporter_sha256='38e439732bbf6e28ca4a94d7bc3686d3fa1abdb0050773d5617a9efdb9e64d08';\
     else\
         postgres_exporter_sha256='d1dedea97f56c6d965837bfd1fbb3e35a3b4a4556f8cccee8bd513d8ee086124';\
         pgbouncer_exporter_sha256='217c4afd7e6492ae904055bc14fe603552cf9bac458c063407e991d68c519da3';\
-        sql_exporter_sha256='530e6afc77c043497ed965532c4c9dfa873bc2a4f0b3047fad367715c0081d6a';\
+        sql_exporter_sha256='11918b00be6e2c3a67564adfdb2414fdcbb15a5db76ea17d1d1a944237a893c6';\
     fi\
     && curl -sL https://github.com/prometheus-community/postgres_exporter/releases/download/v0.17.1/postgres_exporter-0.17.1.linux-${TARGETARCH}.tar.gz\
      | tar xzf - --strip-components=1 -C.\
     && curl -sL https://github.com/prometheus-community/pgbouncer_exporter/releases/download/v0.10.2/pgbouncer_exporter-0.10.2.linux-${TARGETARCH}.tar.gz\
      | tar xzf - --strip-components=1 -C.\
-    && curl -sL https://github.com/burningalchemist/sql_exporter/releases/download/0.17.3/sql_exporter-0.17.3.linux-${TARGETARCH}.tar.gz\
+    && curl -sL https://github.com/burningalchemist/sql_exporter/releases/download/0.17.0/sql_exporter-0.17.0.linux-${TARGETARCH}.tar.gz\
      | tar xzf - --strip-components=1 -C.\
     && echo "${postgres_exporter_sha256} postgres_exporter" | sha256sum -c -\
     && echo "${pgbouncer_exporter_sha256} pgbouncer_exporter" | sha256sum -c -\
@@ -1856,7 +1823,6 @@ COPY --from=pgjwt-src /ext-src/ /ext-src/
 COPY --from=pg_graphql-src /ext-src/ /ext-src/
 #COPY --from=pg_tiktoken-src /ext-src/ /ext-src/
 COPY --from=hypopg-src /ext-src/ /ext-src/
-COPY --from=online_advisor-src /ext-src/ /ext-src/
 COPY --from=pg_hashids-src /ext-src/ /ext-src/
 COPY --from=rum-src /ext-src/ /ext-src/
 COPY --from=pgtap-src /ext-src/ /ext-src/

compute_tools/src/rsyslog.rs

@@ -27,40 +27,6 @@ fn get_rsyslog_pid() -> Option<String> {
     }
 }
 
-fn wait_for_rsyslog_pid() -> Result<String, anyhow::Error> {
-    const MAX_WAIT: Duration = Duration::from_secs(5);
-    const INITIAL_SLEEP: Duration = Duration::from_millis(2);
-
-    let mut sleep_duration = INITIAL_SLEEP;
-    let start = std::time::Instant::now();
-    let mut attempts = 1;
-
-    for attempt in 1.. {
-        attempts = attempt;
-        match get_rsyslog_pid() {
-            Some(pid) => return Ok(pid),
-            None => {
-                if start.elapsed() >= MAX_WAIT {
-                    break;
-                }
-                info!(
-                    "rsyslogd is not running, attempt {}. Sleeping for {} ms",
-                    attempt,
-                    sleep_duration.as_millis()
-                );
-                std::thread::sleep(sleep_duration);
-                sleep_duration *= 2;
-            }
-        }
-    }
-
-    Err(anyhow::anyhow!(
-        "rsyslogd is not running after waiting for {} seconds and {} attempts",
-        attempts,
-        start.elapsed().as_secs()
-    ))
-}
-
 // Restart rsyslogd to apply the new configuration.
 // This is necessary, because there is no other way to reload the rsyslog configuration.
 //
@@ -70,14 +36,14 @@ fn wait_for_rsyslog_pid() -> Result<String, anyhow::Error> {
 // TODO: test it properly
 //
 fn restart_rsyslog() -> Result<()> {
+    let old_pid = get_rsyslog_pid().context("rsyslogd is not running")?;
+    info!("rsyslogd is running with pid: {}, restart it", old_pid);
+
     // kill it to restart
     let _ = Command::new("pkill")
         .arg("rsyslogd")
         .output()
-        .context("Failed to restart rsyslogd")?;
-
-    // ensure rsyslogd is running
-    wait_for_rsyslog_pid()?;
+        .context("Failed to stop rsyslogd")?;
 
     Ok(())
 }
@@ -165,11 +131,15 @@ pub fn configure_postgres_logs_export(conf: PostgresLogsRsyslogConfig) -> Result
         return Ok(());
     }
 
-    // Nothing to configure
+    // When new config is empty we can simply remove the configuration file.
     if new_config.is_empty() {
-        // When the configuration is removed, PostgreSQL will stop sending data
-        // to the files watched by rsyslog, so restarting rsyslog is more effort
-        // than just ignoring this change.
+        info!("removing rsyslog config file: {}", POSTGRES_LOGS_CONF_PATH);
+        match std::fs::remove_file(POSTGRES_LOGS_CONF_PATH) {
+            Ok(_) => {}
+            Err(err) if err.kind() == ErrorKind::NotFound => {}
+            Err(err) => return Err(err.into()),
+        }
+        restart_rsyslog()?;
         return Ok(());
     }
 

control_plane/safekeepers.conf

@@ -2,10 +2,8 @@
 [pageserver]
 listen_pg_addr = '127.0.0.1:64000'
 listen_http_addr = '127.0.0.1:9898'
-listen_grpc_addr = '127.0.0.1:51051'
 pg_auth_type = 'Trust'
 http_auth_type = 'Trust'
-grpc_auth_type = 'Trust'
 
 [[safekeepers]]
 id = 1

control_plane/simple.conf

@@ -4,10 +4,8 @@
 id=1
 listen_pg_addr = '127.0.0.1:64000'
 listen_http_addr = '127.0.0.1:9898'
-listen_grpc_addr = '127.0.0.1:51051'
 pg_auth_type = 'Trust'
 http_auth_type = 'Trust'
-grpc_auth_type = 'Trust'
 
 [[safekeepers]]
 id = 1

control_plane/src/bin/neon_local.rs

@@ -32,7 +32,6 @@ use control_plane::storage_controller::{
 };
 use nix::fcntl::{Flock, FlockArg};
 use pageserver_api::config::{
-    DEFAULT_GRPC_LISTEN_PORT as DEFAULT_PAGESERVER_GRPC_PORT,
     DEFAULT_HTTP_LISTEN_PORT as DEFAULT_PAGESERVER_HTTP_PORT,
     DEFAULT_PG_LISTEN_PORT as DEFAULT_PAGESERVER_PG_PORT,
 };
@@ -1008,16 +1007,13 @@ fn handle_init(args: &InitCmdArgs) -> anyhow::Result<LocalEnv> {
                     let pageserver_id = NodeId(DEFAULT_PAGESERVER_ID.0 + i as u64);
                     let pg_port = DEFAULT_PAGESERVER_PG_PORT + i;
                     let http_port = DEFAULT_PAGESERVER_HTTP_PORT + i;
-                    let grpc_port = DEFAULT_PAGESERVER_GRPC_PORT + i;
                     NeonLocalInitPageserverConf {
                         id: pageserver_id,
                         listen_pg_addr: format!("127.0.0.1:{pg_port}"),
                         listen_http_addr: format!("127.0.0.1:{http_port}"),
                         listen_https_addr: None,
-                        listen_grpc_addr: Some(format!("127.0.0.1:{grpc_port}")),
                         pg_auth_type: AuthType::Trust,
                         http_auth_type: AuthType::Trust,
-                        grpc_auth_type: AuthType::Trust,
                         other: Default::default(),
                         // Typical developer machines use disks with slow fsync, and we don't care
                         // about data integrity: disable disk syncs.

control_plane/src/local_env.rs

@@ -278,10 +278,8 @@ pub struct PageServerConf {
     pub listen_pg_addr: String,
     pub listen_http_addr: String,
     pub listen_https_addr: Option<String>,
-    pub listen_grpc_addr: Option<String>,
     pub pg_auth_type: AuthType,
     pub http_auth_type: AuthType,
-    pub grpc_auth_type: AuthType,
     pub no_sync: bool,
 }
 
@@ -292,10 +290,8 @@ impl Default for PageServerConf {
             listen_pg_addr: String::new(),
             listen_http_addr: String::new(),
             listen_https_addr: None,
-            listen_grpc_addr: None,
             pg_auth_type: AuthType::Trust,
             http_auth_type: AuthType::Trust,
-            grpc_auth_type: AuthType::Trust,
             no_sync: false,
         }
     }
@@ -310,10 +306,8 @@ pub struct NeonLocalInitPageserverConf {
     pub listen_pg_addr: String,
     pub listen_http_addr: String,
     pub listen_https_addr: Option<String>,
-    pub listen_grpc_addr: Option<String>,
     pub pg_auth_type: AuthType,
     pub http_auth_type: AuthType,
-    pub grpc_auth_type: AuthType,
     #[serde(default, skip_serializing_if = "std::ops::Not::not")]
     pub no_sync: bool,
     #[serde(flatten)]
@@ -327,10 +321,8 @@ impl From<&NeonLocalInitPageserverConf> for PageServerConf {
             listen_pg_addr,
             listen_http_addr,
             listen_https_addr,
-            listen_grpc_addr,
             pg_auth_type,
             http_auth_type,
-            grpc_auth_type,
             no_sync,
             other: _,
         } = conf;
@@ -339,9 +331,7 @@ impl From<&NeonLocalInitPageserverConf> for PageServerConf {
             listen_pg_addr: listen_pg_addr.clone(),
             listen_http_addr: listen_http_addr.clone(),
             listen_https_addr: listen_https_addr.clone(),
-            listen_grpc_addr: listen_grpc_addr.clone(),
             pg_auth_type: *pg_auth_type,
-            grpc_auth_type: *grpc_auth_type,
             http_auth_type: *http_auth_type,
             no_sync: *no_sync,
         }
@@ -717,10 +707,8 @@ impl LocalEnv {
                     listen_pg_addr: String,
                     listen_http_addr: String,
                     listen_https_addr: Option<String>,
-                    listen_grpc_addr: Option<String>,
                     pg_auth_type: AuthType,
                     http_auth_type: AuthType,
-                    grpc_auth_type: AuthType,
                     #[serde(default)]
                     no_sync: bool,
                 }
@@ -744,10 +732,8 @@ impl LocalEnv {
                     listen_pg_addr,
                     listen_http_addr,
                     listen_https_addr,
-                    listen_grpc_addr,
                     pg_auth_type,
                     http_auth_type,
-                    grpc_auth_type,
                     no_sync,
                 } = config_toml;
                 let IdentityTomlSubset {
@@ -764,10 +750,8 @@ impl LocalEnv {
                     listen_pg_addr,
                     listen_http_addr,
                     listen_https_addr,
-                    listen_grpc_addr,
                     pg_auth_type,
                     http_auth_type,
-                    grpc_auth_type,
                     no_sync,
                 };
                 pageservers.push(conf);

control_plane/src/pageserver.rs

@@ -129,9 +129,7 @@ impl PageServerNode {
             ));
         }
 
-        if [conf.http_auth_type, conf.pg_auth_type, conf.grpc_auth_type]
-            .contains(&AuthType::NeonJWT)
-        {
+        if conf.http_auth_type != AuthType::Trust || conf.pg_auth_type != AuthType::Trust {
             // Keys are generated in the toplevel repo dir, pageservers' workdirs
             // are one level below that, so refer to keys with ../
             overrides.push("auth_validation_public_key_path='../auth_public_key.pem'".to_owned());
@@ -553,11 +551,6 @@ impl PageServerNode {
                 .map(|x| x.parse::<usize>())
                 .transpose()
                 .context("Falied to parse 'relsize_snapshot_cache_capacity' as integer")?,
-            basebackup_cache_enabled: settings
-                .remove("basebackup_cache_enabled")
-                .map(|x| x.parse::<bool>())
-                .transpose()
-                .context("Failed to parse 'basebackup_cache_enabled' as bool")?,
         };
         if !settings.is_empty() {
             bail!("Unrecognized tenant settings: {settings:?}")

libs/pageserver_api/src/config.rs

@@ -8,8 +8,6 @@ pub const DEFAULT_PG_LISTEN_PORT: u16 = 64000;
 pub const DEFAULT_PG_LISTEN_ADDR: &str = formatcp!("127.0.0.1:{DEFAULT_PG_LISTEN_PORT}");
 pub const DEFAULT_HTTP_LISTEN_PORT: u16 = 9898;
 pub const DEFAULT_HTTP_LISTEN_ADDR: &str = formatcp!("127.0.0.1:{DEFAULT_HTTP_LISTEN_PORT}");
-// TODO: gRPC is disabled by default for now, but the port is used in neon_local.
-pub const DEFAULT_GRPC_LISTEN_PORT: u16 = 51051; // storage-broker already uses 50051
 
 use std::collections::HashMap;
 use std::num::{NonZeroU64, NonZeroUsize};
@@ -106,7 +104,6 @@ pub struct ConfigToml {
     pub listen_pg_addr: String,
     pub listen_http_addr: String,
     pub listen_https_addr: Option<String>,
-    pub listen_grpc_addr: Option<String>,
     pub ssl_key_file: Utf8PathBuf,
     pub ssl_cert_file: Utf8PathBuf,
     #[serde(with = "humantime_serde")]
@@ -126,7 +123,6 @@ pub struct ConfigToml {
     pub http_auth_type: AuthType,
     #[serde_as(as = "serde_with::DisplayFromStr")]
     pub pg_auth_type: AuthType,
-    pub grpc_auth_type: AuthType,
     pub auth_validation_public_key_path: Option<Utf8PathBuf>,
     pub remote_storage: Option<RemoteStorageConfig>,
     pub tenant_config: TenantConfigToml,
@@ -187,8 +183,6 @@ pub struct ConfigToml {
     pub enable_tls_page_service_api: bool,
     pub dev_mode: bool,
     pub timeline_import_config: TimelineImportConfig,
-    #[serde(skip_serializing_if = "Option::is_none")]
-    pub basebackup_cache_config: Option<BasebackupCacheConfig>,
 }
 
 #[derive(Debug, Clone, PartialEq, Eq, serde::Serialize, serde::Deserialize)]
@@ -314,26 +308,6 @@ pub struct TimelineImportConfig {
     pub import_job_checkpoint_threshold: NonZeroUsize,
 }
 
-#[derive(Debug, Clone, PartialEq, Eq, serde::Serialize, serde::Deserialize)]
-#[serde(default)]
-pub struct BasebackupCacheConfig {
-    #[serde(with = "humantime_serde")]
-    pub cleanup_period: Duration,
-    // FIXME: Support max_size_bytes.
-    // pub max_size_bytes: usize,
-    pub max_size_entries: i64,
-}
-
-impl Default for BasebackupCacheConfig {
-    fn default() -> Self {
-        Self {
-            cleanup_period: Duration::from_secs(60),
-            // max_size_bytes: 1024 * 1024 * 1024, // 1 GiB
-            max_size_entries: 1000,
-        }
-    }
-}
-
 pub mod statvfs {
     pub mod mock {
         #[derive(Debug, Clone, PartialEq, Eq, serde::Serialize, serde::Deserialize)]
@@ -517,14 +491,8 @@ pub struct TenantConfigToml {
     /// Tenant level performance sampling ratio override. Controls the ratio of get page requests
     /// that will get perf sampling for the tenant.
     pub sampling_ratio: Option<Ratio>,
-
     /// Capacity of relsize snapshot cache (used by replicas).
     pub relsize_snapshot_cache_capacity: usize,
-
-    /// Enable preparing basebackup on XLOG_CHECKPOINT_SHUTDOWN and using it in basebackup requests.
-    // FIXME: Remove skip_serializing_if when the feature is stable.
-    #[serde(skip_serializing_if = "std::ops::Not::not")]
-    pub basebackup_cache_enabled: bool,
 }
 
 pub mod defaults {
@@ -592,7 +560,6 @@ impl Default for ConfigToml {
             listen_pg_addr: (DEFAULT_PG_LISTEN_ADDR.to_string()),
             listen_http_addr: (DEFAULT_HTTP_LISTEN_ADDR.to_string()),
             listen_https_addr: (None),
-            listen_grpc_addr: None, // TODO: default to 127.0.0.1:51051
             ssl_key_file: Utf8PathBuf::from(DEFAULT_SSL_KEY_FILE),
             ssl_cert_file: Utf8PathBuf::from(DEFAULT_SSL_CERT_FILE),
             ssl_cert_reload_period: Duration::from_secs(60),
@@ -609,7 +576,6 @@ impl Default for ConfigToml {
             pg_distrib_dir: None, // Utf8PathBuf::from("./pg_install"), // TODO: formely, this was std::env::current_dir()
             http_auth_type: (AuthType::Trust),
             pg_auth_type: (AuthType::Trust),
-            grpc_auth_type: (AuthType::Trust),
             auth_validation_public_key_path: (None),
             remote_storage: None,
             broker_endpoint: (storage_broker::DEFAULT_ENDPOINT
@@ -700,7 +666,6 @@ impl Default for ConfigToml {
                 import_job_soft_size_limit: NonZeroUsize::new(1024 * 1024 * 1024).unwrap(),
                 import_job_checkpoint_threshold: NonZeroUsize::new(128).unwrap(),
             },
-            basebackup_cache_config: None,
         }
     }
 }
@@ -826,7 +791,6 @@ impl Default for TenantConfigToml {
             gc_compaction_ratio_percent: DEFAULT_GC_COMPACTION_RATIO_PERCENT,
             sampling_ratio: None,
             relsize_snapshot_cache_capacity: DEFAULT_RELSIZE_SNAPSHOT_CACHE_CAPACITY,
-            basebackup_cache_enabled: false,
         }
     }
 }

libs/pageserver_api/src/models.rs

@@ -632,8 +632,6 @@ pub struct TenantConfigPatch {
     pub sampling_ratio: FieldPatch<Option<Ratio>>,
     #[serde(skip_serializing_if = "FieldPatch::is_noop")]
     pub relsize_snapshot_cache_capacity: FieldPatch<usize>,
-    #[serde(skip_serializing_if = "FieldPatch::is_noop")]
-    pub basebackup_cache_enabled: FieldPatch<bool>,
 }
 
 /// Like [`crate::config::TenantConfigToml`], but preserves the information
@@ -766,9 +764,6 @@ pub struct TenantConfig {
 
     #[serde(skip_serializing_if = "Option::is_none")]
     pub relsize_snapshot_cache_capacity: Option<usize>,
-
-    #[serde(skip_serializing_if = "Option::is_none")]
-    pub basebackup_cache_enabled: Option<bool>,
 }
 
 impl TenantConfig {
@@ -815,7 +810,6 @@ impl TenantConfig {
             mut gc_compaction_ratio_percent,
             mut sampling_ratio,
             mut relsize_snapshot_cache_capacity,
-            mut basebackup_cache_enabled,
         } = self;
 
         patch.checkpoint_distance.apply(&mut checkpoint_distance);
@@ -920,9 +914,6 @@ impl TenantConfig {
         patch
             .relsize_snapshot_cache_capacity
             .apply(&mut relsize_snapshot_cache_capacity);
-        patch
-            .basebackup_cache_enabled
-            .apply(&mut basebackup_cache_enabled);
 
         Ok(Self {
             checkpoint_distance,
@@ -963,7 +954,6 @@ impl TenantConfig {
             gc_compaction_ratio_percent,
             sampling_ratio,
             relsize_snapshot_cache_capacity,
-            basebackup_cache_enabled,
         })
     }
 
@@ -1075,9 +1065,6 @@ impl TenantConfig {
             relsize_snapshot_cache_capacity: self
                 .relsize_snapshot_cache_capacity
                 .unwrap_or(global_conf.relsize_snapshot_cache_capacity),
-            basebackup_cache_enabled: self
-                .basebackup_cache_enabled
-                .unwrap_or(global_conf.basebackup_cache_enabled),
         }
     }
 }

libs/proxy/postgres-protocol2/src/message/frontend.rs

@@ -25,7 +25,6 @@ where
     Ok(())
 }
 
-#[derive(Debug)]
 pub enum BindError {
     Conversion(Box<dyn Error + marker::Sync + Send>),
     Serialization(io::Error),
@@ -289,12 +288,6 @@ pub fn sync(buf: &mut BytesMut) {
     write_body(buf, |_| Ok::<(), io::Error>(())).unwrap();
 }
 
-#[inline]
-pub fn flush(buf: &mut BytesMut) {
-    buf.put_u8(b'H');
-    write_body(buf, |_| Ok::<(), io::Error>(())).unwrap();
-}
-
 #[inline]
 pub fn terminate(buf: &mut BytesMut) {
     buf.put_u8(b'X');

libs/proxy/postgres-types2/src/lib.rs

@@ -9,6 +9,7 @@ use std::error::Error;
 use std::fmt;
 use std::sync::Arc;
 
+use bytes::BytesMut;
 use fallible_iterator::FallibleIterator;
 #[doc(inline)]
 pub use postgres_protocol2::Oid;
@@ -26,6 +27,41 @@ macro_rules! accepts {
     )
 }
 
+/// Generates an implementation of `ToSql::to_sql_checked`.
+///
+/// All `ToSql` implementations should use this macro.
+macro_rules! to_sql_checked {
+    () => {
+        fn to_sql_checked(
+            &self,
+            ty: &$crate::Type,
+            out: &mut $crate::private::BytesMut,
+        ) -> ::std::result::Result<
+            $crate::IsNull,
+            Box<dyn ::std::error::Error + ::std::marker::Sync + ::std::marker::Send>,
+        > {
+            $crate::__to_sql_checked(self, ty, out)
+        }
+    };
+}
+
+// WARNING: this function is not considered part of this crate's public API.
+// It is subject to change at any time.
+#[doc(hidden)]
+pub fn __to_sql_checked<T>(
+    v: &T,
+    ty: &Type,
+    out: &mut BytesMut,
+) -> Result<IsNull, Box<dyn Error + Sync + Send>>
+where
+    T: ToSql,
+{
+    if !T::accepts(ty) {
+        return Err(Box::new(WrongType::new::<T>(ty.clone())));
+    }
+    v.to_sql(ty, out)
+}
+
 // mod pg_lsn;
 #[doc(hidden)]
 pub mod private;
@@ -106,7 +142,7 @@ pub enum Kind {
     /// An array type along with the type of its elements.
     Array(Type),
     /// A range type along with the type of its elements.
-    Range(Oid),
+    Range(Type),
     /// A multirange type along with the type of its elements.
     Multirange(Type),
     /// A domain type along with its underlying type.
@@ -341,6 +377,43 @@ pub enum IsNull {
     No,
 }
 
+/// A trait for types that can be converted into Postgres values.
+pub trait ToSql: fmt::Debug {
+    /// Converts the value of `self` into the binary format of the specified
+    /// Postgres `Type`, appending it to `out`.
+    ///
+    /// The caller of this method is responsible for ensuring that this type
+    /// is compatible with the Postgres `Type`.
+    ///
+    /// The return value indicates if this value should be represented as
+    /// `NULL`. If this is the case, implementations **must not** write
+    /// anything to `out`.
+    fn to_sql(&self, ty: &Type, out: &mut BytesMut) -> Result<IsNull, Box<dyn Error + Sync + Send>>
+    where
+        Self: Sized;
+
+    /// Determines if a value of this type can be converted to the specified
+    /// Postgres `Type`.
+    fn accepts(ty: &Type) -> bool
+    where
+        Self: Sized;
+
+    /// An adaptor method used internally by Rust-Postgres.
+    ///
+    /// *All* implementations of this method should be generated by the
+    /// `to_sql_checked!()` macro.
+    fn to_sql_checked(
+        &self,
+        ty: &Type,
+        out: &mut BytesMut,
+    ) -> Result<IsNull, Box<dyn Error + Sync + Send>>;
+
+    /// Specify the encode format
+    fn encode_format(&self, _ty: &Type) -> Format {
+        Format::Binary
+    }
+}
+
 /// Supported Postgres message format types
 ///
 /// Using Text format in a message assumes a Postgres `SERVER_ENCODING` of `UTF8`
@@ -351,3 +424,52 @@ pub enum Format {
     /// Compact, typed binary format
     Binary,
 }
+
+impl ToSql for &str {
+    fn to_sql(&self, ty: &Type, w: &mut BytesMut) -> Result<IsNull, Box<dyn Error + Sync + Send>> {
+        match *ty {
+            ref ty if ty.name() == "ltree" => types::ltree_to_sql(self, w),
+            ref ty if ty.name() == "lquery" => types::lquery_to_sql(self, w),
+            ref ty if ty.name() == "ltxtquery" => types::ltxtquery_to_sql(self, w),
+            _ => types::text_to_sql(self, w),
+        }
+        Ok(IsNull::No)
+    }
+
+    fn accepts(ty: &Type) -> bool {
+        match *ty {
+            Type::VARCHAR | Type::TEXT | Type::BPCHAR | Type::NAME | Type::UNKNOWN => true,
+            ref ty
+                if (ty.name() == "citext"
+                    || ty.name() == "ltree"
+                    || ty.name() == "lquery"
+                    || ty.name() == "ltxtquery") =>
+            {
+                true
+            }
+            _ => false,
+        }
+    }
+
+    to_sql_checked!();
+}
+
+macro_rules! simple_to {
+    ($t:ty, $f:ident, $($expected:ident),+) => {
+        impl ToSql for $t {
+            fn to_sql(&self,
+                      _: &Type,
+                      w: &mut BytesMut)
+                      -> Result<IsNull, Box<dyn Error + Sync + Send>> {
+                types::$f(*self, w);
+                Ok(IsNull::No)
+            }
+
+            accepts!($($expected),+);
+
+            to_sql_checked!();
+        }
+    }
+}
+
+simple_to!(u32, oid_to_sql, OID);

libs/proxy/postgres-types2/src/type_gen.rs

@@ -393,7 +393,7 @@ impl Inner {
         }
     }
 
-    pub const fn const_oid(&self) -> Oid {
+    pub fn oid(&self) -> Oid {
         match *self {
             Inner::Bool => 16,
             Inner::Bytea => 17,
@@ -580,14 +580,7 @@ impl Inner {
             Inner::TstzmultiRangeArray => 6153,
             Inner::DatemultiRangeArray => 6155,
             Inner::Int8multiRangeArray => 6157,
-            Inner::Other(_) => u32::MAX,
-        }
-    }
-
-    pub fn oid(&self) -> Oid {
-        match *self {
             Inner::Other(ref u) => u.oid,
-            _ => self.const_oid(),
         }
     }
 
@@ -734,17 +727,17 @@ impl Inner {
             Inner::JsonbArray => &Kind::Array(Type(Inner::Jsonb)),
             Inner::AnyRange => &Kind::Pseudo,
             Inner::EventTrigger => &Kind::Pseudo,
-            Inner::Int4Range => &const { Kind::Range(Inner::Int4.const_oid()) },
+            Inner::Int4Range => &Kind::Range(Type(Inner::Int4)),
             Inner::Int4RangeArray => &Kind::Array(Type(Inner::Int4Range)),
-            Inner::NumRange => &const { Kind::Range(Inner::Numeric.const_oid()) },
+            Inner::NumRange => &Kind::Range(Type(Inner::Numeric)),
             Inner::NumRangeArray => &Kind::Array(Type(Inner::NumRange)),
-            Inner::TsRange => &const { Kind::Range(Inner::Timestamp.const_oid()) },
+            Inner::TsRange => &Kind::Range(Type(Inner::Timestamp)),
             Inner::TsRangeArray => &Kind::Array(Type(Inner::TsRange)),
-            Inner::TstzRange => &const { Kind::Range(Inner::Timestamptz.const_oid()) },
+            Inner::TstzRange => &Kind::Range(Type(Inner::Timestamptz)),
             Inner::TstzRangeArray => &Kind::Array(Type(Inner::TstzRange)),
-            Inner::DateRange => &const { Kind::Range(Inner::Date.const_oid()) },
+            Inner::DateRange => &Kind::Range(Type(Inner::Date)),
             Inner::DateRangeArray => &Kind::Array(Type(Inner::DateRange)),
-            Inner::Int8Range => &const { Kind::Range(Inner::Int8.const_oid()) },
+            Inner::Int8Range => &Kind::Range(Type(Inner::Int8)),
             Inner::Int8RangeArray => &Kind::Array(Type(Inner::Int8Range)),
             Inner::Jsonpath => &Kind::Simple,
             Inner::JsonpathArray => &Kind::Array(Type(Inner::Jsonpath)),

libs/proxy/tokio-postgres2/src/client.rs

@@ -1,12 +1,14 @@
 use std::collections::HashMap;
 use std::fmt;
 use std::net::IpAddr;
+use std::sync::Arc;
 use std::task::{Context, Poll};
 use std::time::Duration;
 
 use bytes::BytesMut;
 use fallible_iterator::FallibleIterator;
 use futures_util::{TryStreamExt, future, ready};
+use parking_lot::Mutex;
 use postgres_protocol2::message::backend::Message;
 use postgres_protocol2::message::frontend;
 use serde::{Deserialize, Serialize};
@@ -14,52 +16,29 @@ use tokio::sync::mpsc;
 
 use crate::codec::{BackendMessages, FrontendMessage};
 use crate::config::{Host, SslMode};
+use crate::connection::{Request, RequestMessages};
 use crate::query::RowStream;
 use crate::simple_query::SimpleQueryStream;
 use crate::types::{Oid, Type};
 use crate::{
-    CancelToken, Error, ReadyForQueryStatus, SimpleQueryMessage, Transaction, TransactionBuilder,
-    query, simple_query,
+    CancelToken, Error, ReadyForQueryStatus, SimpleQueryMessage, Statement, Transaction,
+    TransactionBuilder, query, simple_query,
 };
 
 pub struct Responses {
-    /// new messages from conn
     receiver: mpsc::Receiver<BackendMessages>,
-    /// current batch of messages
     cur: BackendMessages,
-    /// number of total queries sent.
-    waiting: usize,
-    /// number of ReadyForQuery messages received.
-    received: usize,
 }
 
 impl Responses {
     pub fn poll_next(&mut self, cx: &mut Context<'_>) -> Poll<Result<Message, Error>> {
         loop {
-            // get the next saved message
-            if let Some(message) = self.cur.next().map_err(Error::parse)? {
-                let received = self.received;
-
-                // increase the query head if this is the last message.
-                if let Message::ReadyForQuery(_) = message {
-                    self.received += 1;
-                }
-
-                // check if the client has skipped this query.
-                if received + 1 < self.waiting {
-                    // grab the next message.
-                    continue;
-                }
-
-                // convenience: turn the error messaage into a proper error.
-                let res = match message {
-                    Message::ErrorResponse(body) => Err(Error::db(body)),
-                    message => Ok(message),
-                };
-                return Poll::Ready(res);
+            match self.cur.next().map_err(Error::parse)? {
+                Some(Message::ErrorResponse(body)) => return Poll::Ready(Err(Error::db(body))),
+                Some(message) => return Poll::Ready(Ok(message)),
+                None => {}
             }
 
-            // get the next batch of messages.
             match ready!(self.receiver.poll_recv(cx)) {
                 Some(messages) => self.cur = messages,
                 None => return Poll::Ready(Err(Error::closed())),
@@ -76,87 +55,44 @@ impl Responses {
 /// (corresponding to the queries in the [crate::prepare] module).
 #[derive(Default)]
 pub(crate) struct CachedTypeInfo {
+    /// A statement for basic information for a type from its
+    /// OID. Corresponds to [TYPEINFO_QUERY](crate::prepare::TYPEINFO_QUERY) (or its
+    /// fallback).
+    pub(crate) typeinfo: Option<Statement>,
+
     /// Cache of types already looked up.
     pub(crate) types: HashMap<Oid, Type>,
 }
 
 pub struct InnerClient {
-    sender: mpsc::UnboundedSender<FrontendMessage>,
-    responses: Responses,
+    sender: mpsc::UnboundedSender<Request>,
 
     /// A buffer to use when writing out postgres commands.
-    buffer: BytesMut,
+    buffer: Mutex<BytesMut>,
 }
 
 impl InnerClient {
-    pub fn start(&mut self) -> Result<PartialQuery, Error> {
-        self.responses.waiting += 1;
-        Ok(PartialQuery(Some(self)))
+    pub fn send(&self, messages: RequestMessages) -> Result<Responses, Error> {
+        let (sender, receiver) = mpsc::channel(1);
+        let request = Request { messages, sender };
+        self.sender.send(request).map_err(|_| Error::closed())?;
+
+        Ok(Responses {
+            receiver,
+            cur: BackendMessages::empty(),
+        })
     }
 
-    // pub fn send_with_sync<F>(&mut self, f: F) -> Result<&mut Responses, Error>
-    // where
-    //     F: FnOnce(&mut BytesMut) -> Result<(), Error>,
-    // {
-    //     self.start()?.send_with_sync(f)
-    // }
-
-    pub fn send_simple_query(&mut self, query: &str) -> Result<&mut Responses, Error> {
-        self.responses.waiting += 1;
-
-        self.buffer.clear();
-        // simple queries do not need sync.
-        frontend::query(query, &mut self.buffer).map_err(Error::encode)?;
-        let buf = self.buffer.split().freeze();
-        self.send_message(FrontendMessage::Raw(buf))
-    }
-
-    fn send_message(&mut self, messages: FrontendMessage) -> Result<&mut Responses, Error> {
-        self.sender.send(messages).map_err(|_| Error::closed())?;
-        Ok(&mut self.responses)
-    }
-}
-
-pub struct PartialQuery<'a>(Option<&'a mut InnerClient>);
-
-impl Drop for PartialQuery<'_> {
-    fn drop(&mut self) {
-        if let Some(client) = self.0.take() {
-            client.buffer.clear();
-            frontend::sync(&mut client.buffer);
-            let buf = client.buffer.split().freeze();
-            let _ = client.send_message(FrontendMessage::Raw(buf));
-        }
-    }
-}
-
-impl<'a> PartialQuery<'a> {
-    pub fn send_with_flush<F>(&mut self, f: F) -> Result<&mut Responses, Error>
+    /// Call the given function with a buffer to be used when writing out
+    /// postgres commands.
+    pub fn with_buf<F, R>(&self, f: F) -> R
     where
-        F: FnOnce(&mut BytesMut) -> Result<(), Error>,
+        F: FnOnce(&mut BytesMut) -> R,
     {
-        let client = self.0.as_deref_mut().unwrap();
-
-        client.buffer.clear();
-        f(&mut client.buffer)?;
-        frontend::flush(&mut client.buffer);
-        let buf = client.buffer.split().freeze();
-        client.send_message(FrontendMessage::Raw(buf))
-    }
-
-    pub fn send_with_sync<F>(mut self, f: F) -> Result<&'a mut Responses, Error>
-    where
-        F: FnOnce(&mut BytesMut) -> Result<(), Error>,
-    {
-        let client = self.0.as_deref_mut().unwrap();
-
-        client.buffer.clear();
-        f(&mut client.buffer)?;
-        frontend::sync(&mut client.buffer);
-        let buf = client.buffer.split().freeze();
-        let _ = client.send_message(FrontendMessage::Raw(buf));
-
-        Ok(&mut self.0.take().unwrap().responses)
+        let mut buffer = self.buffer.lock();
+        let r = f(&mut buffer);
+        buffer.clear();
+        r
     }
 }
 
@@ -173,7 +109,7 @@ pub struct SocketConfig {
 /// The client is one half of what is returned when a connection is established. Users interact with the database
 /// through this client object.
 pub struct Client {
-    inner: InnerClient,
+    inner: Arc<InnerClient>,
     cached_typeinfo: CachedTypeInfo,
 
     socket_config: SocketConfig,
@@ -184,24 +120,17 @@ pub struct Client {
 
 impl Client {
     pub(crate) fn new(
-        sender: mpsc::UnboundedSender<FrontendMessage>,
-        receiver: mpsc::Receiver<BackendMessages>,
+        sender: mpsc::UnboundedSender<Request>,
         socket_config: SocketConfig,
         ssl_mode: SslMode,
         process_id: i32,
         secret_key: i32,
     ) -> Client {
         Client {
-            inner: InnerClient {
+            inner: Arc::new(InnerClient {
                 sender,
-                responses: Responses {
-                    receiver,
-                    cur: BackendMessages::empty(),
-                    waiting: 0,
-                    received: 0,
-                },
                 buffer: Default::default(),
-            },
+            }),
             cached_typeinfo: Default::default(),
 
             socket_config,
@@ -216,29 +145,19 @@ impl Client {
         self.process_id
     }
 
-    pub(crate) fn inner_mut(&mut self) -> &mut InnerClient {
-        &mut self.inner
+    pub(crate) fn inner(&self) -> &Arc<InnerClient> {
+        &self.inner
     }
 
     /// Pass text directly to the Postgres backend to allow it to sort out typing itself and
     /// to save a roundtrip
-    pub async fn query_raw_txt<S, I>(
-        &mut self,
-        statement: &str,
-        params: I,
-    ) -> Result<RowStream, Error>
+    pub async fn query_raw_txt<S, I>(&self, statement: &str, params: I) -> Result<RowStream, Error>
     where
         S: AsRef<str>,
         I: IntoIterator<Item = Option<S>>,
         I::IntoIter: ExactSizeIterator,
     {
-        query::query_txt(
-            &mut self.inner,
-            &mut self.cached_typeinfo,
-            statement,
-            params,
-        )
-        .await
+        query::query_txt(&self.inner, statement, params).await
     }
 
     /// Executes a sequence of SQL statements using the simple query protocol, returning the resulting rows.
@@ -254,15 +173,12 @@ impl Client {
     /// Prepared statements should be use for any query which contains user-specified data, as they provided the
     /// functionality to safely embed that data in the request. Do not form statements via string concatenation and pass
     /// them to this method!
-    pub async fn simple_query(&mut self, query: &str) -> Result<Vec<SimpleQueryMessage>, Error> {
+    pub async fn simple_query(&self, query: &str) -> Result<Vec<SimpleQueryMessage>, Error> {
         self.simple_query_raw(query).await?.try_collect().await
     }
 
-    pub(crate) async fn simple_query_raw(
-        &mut self,
-        query: &str,
-    ) -> Result<SimpleQueryStream, Error> {
-        simple_query::simple_query(self.inner_mut(), query).await
+    pub(crate) async fn simple_query_raw(&self, query: &str) -> Result<SimpleQueryStream, Error> {
+        simple_query::simple_query(self.inner(), query).await
     }
 
     /// Executes a sequence of SQL statements using the simple query protocol.
@@ -275,11 +191,15 @@ impl Client {
     /// Prepared statements should be use for any query which contains user-specified data, as they provided the
     /// functionality to safely embed that data in the request. Do not form statements via string concatenation and pass
     /// them to this method!
-    pub async fn batch_execute(&mut self, query: &str) -> Result<ReadyForQueryStatus, Error> {
-        simple_query::batch_execute(self.inner_mut(), query).await
+    pub async fn batch_execute(&self, query: &str) -> Result<ReadyForQueryStatus, Error> {
+        simple_query::batch_execute(self.inner(), query).await
     }
 
     pub async fn discard_all(&mut self) -> Result<ReadyForQueryStatus, Error> {
+        // clear the prepared statements that are about to be nuked from the postgres session
+
+        self.cached_typeinfo.typeinfo = None;
+
         self.batch_execute("discard all").await
     }
 
@@ -288,7 +208,7 @@ impl Client {
     /// The transaction will roll back by default - use the `commit` method to commit it.
     pub async fn transaction(&mut self) -> Result<Transaction<'_>, Error> {
         struct RollbackIfNotDone<'me> {
-            client: &'me mut Client,
+            client: &'me Client,
             done: bool,
         }
 
@@ -298,7 +218,14 @@ impl Client {
                     return;
                 }
 
-                let _ = self.client.inner.send_simple_query("ROLLBACK");
+                let buf = self.client.inner().with_buf(|buf| {
+                    frontend::query("ROLLBACK", buf).unwrap();
+                    buf.split().freeze()
+                });
+                let _ = self
+                    .client
+                    .inner()
+                    .send(RequestMessages::Single(FrontendMessage::Raw(buf)));
             }
         }
 
@@ -312,7 +239,7 @@ impl Client {
                 client: self,
                 done: false,
             };
-            cleaner.client.batch_execute("BEGIN").await?;
+            self.batch_execute("BEGIN").await?;
             cleaner.done = true;
         }
 
@@ -338,6 +265,11 @@ impl Client {
         }
     }
 
+    /// Query for type information
+    pub(crate) async fn get_type_inner(&mut self, oid: Oid) -> Result<Type, Error> {
+        crate::prepare::get_type(&self.inner, &mut self.cached_typeinfo, oid).await
+    }
+
     /// Determines if the connection to the server has already closed.
     ///
     /// In that case, all future queries will fail.

libs/proxy/tokio-postgres2/src/codec.rs

@@ -1,16 +1,21 @@
 use std::io;
 
-use bytes::{Bytes, BytesMut};
+use bytes::{Buf, Bytes, BytesMut};
 use fallible_iterator::FallibleIterator;
 use postgres_protocol2::message::backend;
+use postgres_protocol2::message::frontend::CopyData;
 use tokio_util::codec::{Decoder, Encoder};
 
 pub enum FrontendMessage {
     Raw(Bytes),
+    CopyData(CopyData<Box<dyn Buf + Send>>),
 }
 
 pub enum BackendMessage {
-    Normal { messages: BackendMessages },
+    Normal {
+        messages: BackendMessages,
+        request_complete: bool,
+    },
     Async(backend::Message),
 }
 
@@ -39,6 +44,7 @@ impl Encoder<FrontendMessage> for PostgresCodec {
     fn encode(&mut self, item: FrontendMessage, dst: &mut BytesMut) -> io::Result<()> {
         match item {
             FrontendMessage::Raw(buf) => dst.extend_from_slice(&buf),
+            FrontendMessage::CopyData(data) => data.write(dst),
         }
 
         Ok(())
@@ -51,6 +57,7 @@ impl Decoder for PostgresCodec {
 
     fn decode(&mut self, src: &mut BytesMut) -> Result<Option<BackendMessage>, io::Error> {
         let mut idx = 0;
+        let mut request_complete = false;
 
         while let Some(header) = backend::Header::parse(&src[idx..])? {
             let len = header.len() as usize + 1;
@@ -75,6 +82,7 @@ impl Decoder for PostgresCodec {
             idx += len;
 
             if header.tag() == backend::READY_FOR_QUERY_TAG {
+                request_complete = true;
                 break;
             }
         }
@@ -84,6 +92,7 @@ impl Decoder for PostgresCodec {
         } else {
             Ok(Some(BackendMessage::Normal {
                 messages: BackendMessages(src.split_to(idx)),
+                request_complete,
             }))
         }
     }

libs/proxy/tokio-postgres2/src/connect.rs

@@ -59,11 +59,9 @@ where
         connect_timeout: config.connect_timeout,
     };
 
-    let (client_tx, conn_rx) = mpsc::unbounded_channel();
-    let (conn_tx, client_rx) = mpsc::channel(4);
+    let (sender, receiver) = mpsc::unbounded_channel();
     let client = Client::new(
-        client_tx,
-        client_rx,
+        sender,
         socket_config,
         config.ssl_mode,
         process_id,
@@ -76,7 +74,7 @@ where
         .map(|m| BackendMessage::Async(Message::NoticeResponse(m)))
         .collect();
 
-    let connection = Connection::new(stream, delayed, parameters, conn_tx, conn_rx);
+    let connection = Connection::new(stream, delayed, parameters, receiver);
 
     Ok((client, connection))
 }

libs/proxy/tokio-postgres2/src/connection.rs

@@ -4,6 +4,7 @@ use std::pin::Pin;
 use std::task::{Context, Poll};
 
 use bytes::BytesMut;
+use fallible_iterator::FallibleIterator;
 use futures_util::{Sink, Stream, ready};
 use postgres_protocol2::message::backend::Message;
 use postgres_protocol2::message::frontend;
@@ -18,12 +19,30 @@ use crate::error::DbError;
 use crate::maybe_tls_stream::MaybeTlsStream;
 use crate::{AsyncMessage, Error, Notification};
 
+pub enum RequestMessages {
+    Single(FrontendMessage),
+}
+
+pub struct Request {
+    pub messages: RequestMessages,
+    pub sender: mpsc::Sender<BackendMessages>,
+}
+
+pub struct Response {
+    sender: PollSender<BackendMessages>,
+}
+
 #[derive(PartialEq, Debug)]
 enum State {
     Active,
     Closing,
 }
 
+enum WriteReady {
+    Terminating,
+    WaitingOnRead,
+}
+
 /// A connection to a PostgreSQL database.
 ///
 /// This is one half of what is returned when a new connection is established. It performs the actual IO with the
@@ -37,11 +56,9 @@ pub struct Connection<S, T> {
     pub stream: Framed<MaybeTlsStream<S, T>, PostgresCodec>,
     /// HACK: we need this in the Neon Proxy to forward params.
     pub parameters: HashMap<String, String>,
-
-    sender: PollSender<BackendMessages>,
-    receiver: mpsc::UnboundedReceiver<FrontendMessage>,
-
+    receiver: mpsc::UnboundedReceiver<Request>,
     pending_responses: VecDeque<BackendMessage>,
+    responses: VecDeque<Response>,
     state: State,
 }
 
@@ -54,15 +71,14 @@ where
         stream: Framed<MaybeTlsStream<S, T>, PostgresCodec>,
         pending_responses: VecDeque<BackendMessage>,
         parameters: HashMap<String, String>,
-        sender: mpsc::Sender<BackendMessages>,
-        receiver: mpsc::UnboundedReceiver<FrontendMessage>,
+        receiver: mpsc::UnboundedReceiver<Request>,
     ) -> Connection<S, T> {
         Connection {
             stream,
             parameters,
-            sender: PollSender::new(sender),
             receiver,
             pending_responses,
+            responses: VecDeque::new(),
             state: State::Active,
         }
     }
@@ -94,7 +110,7 @@ where
                 }
             };
 
-            let messages = match message {
+            let (mut messages, request_complete) = match message {
                 BackendMessage::Async(Message::NoticeResponse(body)) => {
                     let error = DbError::parse(&mut body.fields()).map_err(Error::parse)?;
                     return Poll::Ready(Ok(AsyncMessage::Notice(error)));
@@ -115,19 +131,41 @@ where
                     continue;
                 }
                 BackendMessage::Async(_) => unreachable!(),
-                BackendMessage::Normal { messages } => messages,
+                BackendMessage::Normal {
+                    messages,
+                    request_complete,
+                } => (messages, request_complete),
             };
 
-            match self.sender.poll_reserve(cx) {
+            let mut response = match self.responses.pop_front() {
+                Some(response) => response,
+                None => match messages.next().map_err(Error::parse)? {
+                    Some(Message::ErrorResponse(error)) => {
+                        return Poll::Ready(Err(Error::db(error)));
+                    }
+                    _ => return Poll::Ready(Err(Error::unexpected_message())),
+                },
+            };
+
+            match response.sender.poll_reserve(cx) {
                 Poll::Ready(Ok(())) => {
-                    let _ = self.sender.send_item(messages);
+                    let _ = response.sender.send_item(messages);
+                    if !request_complete {
+                        self.responses.push_front(response);
+                    }
                 }
                 Poll::Ready(Err(_)) => {
-                    return Poll::Ready(Err(Error::closed()));
+                    // we need to keep paging through the rest of the messages even if the receiver's hung up
+                    if !request_complete {
+                        self.responses.push_front(response);
+                    }
                 }
                 Poll::Pending => {
-                    self.pending_responses
-                        .push_back(BackendMessage::Normal { messages });
+                    self.responses.push_front(response);
+                    self.pending_responses.push_back(BackendMessage::Normal {
+                        messages,
+                        request_complete,
+                    });
                     trace!("poll_read: waiting on sender");
                     return Poll::Pending;
                 }
@@ -136,7 +174,7 @@ where
     }
 
     /// Fetch the next client request and enqueue the response sender.
-    fn poll_request(&mut self, cx: &mut Context<'_>) -> Poll<Option<FrontendMessage>> {
+    fn poll_request(&mut self, cx: &mut Context<'_>) -> Poll<Option<RequestMessages>> {
         if self.receiver.is_closed() {
             return Poll::Ready(None);
         }
@@ -144,7 +182,10 @@ where
         match self.receiver.poll_recv(cx) {
             Poll::Ready(Some(request)) => {
                 trace!("polled new request");
-                Poll::Ready(Some(request))
+                self.responses.push_back(Response {
+                    sender: PollSender::new(request.sender),
+                });
+                Poll::Ready(Some(request.messages))
             }
             Poll::Ready(None) => Poll::Ready(None),
             Poll::Pending => Poll::Pending,
@@ -153,7 +194,7 @@ where
 
     /// Process client requests and write them to the postgres connection, flushing if necessary.
     /// client -> postgres
-    fn poll_write(&mut self, cx: &mut Context<'_>) -> Poll<Result<(), Error>> {
+    fn poll_write(&mut self, cx: &mut Context<'_>) -> Poll<Result<WriteReady, Error>> {
         loop {
             if Pin::new(&mut self.stream)
                 .poll_ready(cx)
@@ -168,14 +209,14 @@ where
 
             match self.poll_request(cx) {
                 // send the message to postgres
-                Poll::Ready(Some(request)) => {
+                Poll::Ready(Some(RequestMessages::Single(request))) => {
                     Pin::new(&mut self.stream)
                         .start_send(request)
                         .map_err(Error::io)?;
                 }
                 // No more messages from the client, and no more responses to wait for.
                 // Send a terminate message to postgres
-                Poll::Ready(None) => {
+                Poll::Ready(None) if self.responses.is_empty() => {
                     trace!("poll_write: at eof, terminating");
                     let mut request = BytesMut::new();
                     frontend::terminate(&mut request);
@@ -187,7 +228,16 @@ where
 
                     trace!("poll_write: sent eof, closing");
                     trace!("poll_write: done");
-                    return Poll::Ready(Ok(()));
+                    return Poll::Ready(Ok(WriteReady::Terminating));
+                }
+                // No more messages from the client, but there are still some responses to wait for.
+                Poll::Ready(None) => {
+                    trace!(
+                        "poll_write: at eof, pending responses {}",
+                        self.responses.len()
+                    );
+                    ready!(self.poll_flush(cx))?;
+                    return Poll::Ready(Ok(WriteReady::WaitingOnRead));
                 }
                 // Still waiting for a message from the client.
                 Poll::Pending => {
@@ -248,7 +298,7 @@ where
             // if the state is still active, try read from and write to postgres.
             let message = self.poll_read(cx)?;
             let closing = self.poll_write(cx)?;
-            if let Poll::Ready(()) = closing {
+            if let Poll::Ready(WriteReady::Terminating) = closing {
                 self.state = State::Closing;
             }
 

libs/proxy/tokio-postgres2/src/generic_client.rs

@@ -1,6 +1,9 @@
 #![allow(async_fn_in_trait)]
 
+use postgres_protocol2::Oid;
+
 use crate::query::RowStream;
+use crate::types::Type;
 use crate::{Client, Error, Transaction};
 
 mod private {
@@ -12,17 +15,20 @@ mod private {
 /// This trait is "sealed", and cannot be implemented outside of this crate.
 pub trait GenericClient: private::Sealed {
     /// Like `Client::query_raw_txt`.
-    async fn query_raw_txt<S, I>(&mut self, statement: &str, params: I) -> Result<RowStream, Error>
+    async fn query_raw_txt<S, I>(&self, statement: &str, params: I) -> Result<RowStream, Error>
     where
         S: AsRef<str> + Sync + Send,
         I: IntoIterator<Item = Option<S>> + Sync + Send,
         I::IntoIter: ExactSizeIterator + Sync + Send;
+
+    /// Query for type information
+    async fn get_type(&mut self, oid: Oid) -> Result<Type, Error>;
 }
 
 impl private::Sealed for Client {}
 
 impl GenericClient for Client {
-    async fn query_raw_txt<S, I>(&mut self, statement: &str, params: I) -> Result<RowStream, Error>
+    async fn query_raw_txt<S, I>(&self, statement: &str, params: I) -> Result<RowStream, Error>
     where
         S: AsRef<str> + Sync + Send,
         I: IntoIterator<Item = Option<S>> + Sync + Send,
@@ -30,12 +36,17 @@ impl GenericClient for Client {
     {
         self.query_raw_txt(statement, params).await
     }
+
+    /// Query for type information
+    async fn get_type(&mut self, oid: Oid) -> Result<Type, Error> {
+        self.get_type_inner(oid).await
+    }
 }
 
 impl private::Sealed for Transaction<'_> {}
 
 impl GenericClient for Transaction<'_> {
-    async fn query_raw_txt<S, I>(&mut self, statement: &str, params: I) -> Result<RowStream, Error>
+    async fn query_raw_txt<S, I>(&self, statement: &str, params: I) -> Result<RowStream, Error>
     where
         S: AsRef<str> + Sync + Send,
         I: IntoIterator<Item = Option<S>> + Sync + Send,
@@ -43,4 +54,9 @@ impl GenericClient for Transaction<'_> {
     {
         self.query_raw_txt(statement, params).await
     }
+
+    /// Query for type information
+    async fn get_type(&mut self, oid: Oid) -> Result<Type, Error> {
+        self.client_mut().get_type(oid).await
+    }
 }

libs/proxy/tokio-postgres2/src/lib.rs

@@ -18,6 +18,7 @@ pub use crate::statement::{Column, Statement};
 pub use crate::tls::NoTls;
 pub use crate::transaction::Transaction;
 pub use crate::transaction_builder::{IsolationLevel, TransactionBuilder};
+use crate::types::ToSql;
 
 /// After executing a query, the connection will be in one of these states
 #[derive(Clone, Copy, Debug, PartialEq)]
@@ -119,3 +120,9 @@ pub enum SimpleQueryMessage {
     /// The number of rows modified or selected is returned.
     CommandComplete(u64),
 }
+
+fn slice_iter<'a>(
+    s: &'a [&'a (dyn ToSql + Sync)],
+) -> impl ExactSizeIterator<Item = &'a (dyn ToSql + Sync)> + 'a {
+    s.iter().map(|s| *s as _)
+}

libs/proxy/tokio-postgres2/src/prepare.rs

@@ -1,14 +1,19 @@
-use bytes::BytesMut;
-use fallible_iterator::FallibleIterator;
-use postgres_protocol2::IsNull;
-use postgres_protocol2::message::backend::{Message, RowDescriptionBody};
-use postgres_protocol2::message::frontend;
-use postgres_protocol2::types::oid_to_sql;
-use postgres_types2::Format;
+use std::future::Future;
+use std::pin::Pin;
+use std::sync::Arc;
 
-use crate::client::{CachedTypeInfo, PartialQuery, Responses};
+use bytes::Bytes;
+use fallible_iterator::FallibleIterator;
+use futures_util::{TryStreamExt, pin_mut};
+use postgres_protocol2::message::backend::Message;
+use postgres_protocol2::message::frontend;
+use tracing::debug;
+
+use crate::client::{CachedTypeInfo, InnerClient};
+use crate::codec::FrontendMessage;
+use crate::connection::RequestMessages;
 use crate::types::{Kind, Oid, Type};
-use crate::{Column, Error, Row, Statement};
+use crate::{Column, Error, Statement, query, slice_iter};
 
 pub(crate) const TYPEINFO_QUERY: &str = "\
 SELECT t.typname, t.typtype, t.typelem, r.rngsubtype, t.typbasetype, n.nspname, t.typrelid
@@ -18,51 +23,22 @@ INNER JOIN pg_catalog.pg_namespace n ON t.typnamespace = n.oid
 WHERE t.oid = $1
 ";
 
-/// we need to make sure we close this prepared statement.
-struct CloseStmt<'a, 'b> {
-    client: Option<&'a mut PartialQuery<'b>>,
-    name: &'static str,
-}
-
-impl<'a> CloseStmt<'a, '_> {
-    fn close(mut self) -> Result<&'a mut Responses, Error> {
-        let client = self.client.take().unwrap();
-        client.send_with_flush(|buf| {
-            frontend::close(b'S', self.name, buf).map_err(Error::encode)?;
-            Ok(())
-        })
-    }
-}
-
-impl Drop for CloseStmt<'_, '_> {
-    fn drop(&mut self) {
-        if let Some(client) = self.client.take() {
-            let _ = client.send_with_flush(|buf| {
-                frontend::close(b'S', self.name, buf).map_err(Error::encode)?;
-                Ok(())
-            });
-        }
-    }
-}
-
 async fn prepare_typecheck(
-    client: &mut PartialQuery<'_>,
+    client: &Arc<InnerClient>,
     name: &'static str,
     query: &str,
+    types: &[Type],
 ) -> Result<Statement, Error> {
-    let responses = client.send_with_flush(|buf| {
-        frontend::parse(name, query, [], buf).map_err(Error::encode)?;
-        frontend::describe(b'S', name, buf).map_err(Error::encode)?;
-        Ok(())
-    })?;
+    let buf = encode(client, name, query, types)?;
+    let mut responses = client.send(RequestMessages::Single(FrontendMessage::Raw(buf)))?;
 
     match responses.next().await? {
         Message::ParseComplete => {}
         _ => return Err(Error::unexpected_message()),
     }
 
-    match responses.next().await? {
-        Message::ParameterDescription(_) => {}
+    let parameter_description = match responses.next().await? {
+        Message::ParameterDescription(body) => body,
         _ => return Err(Error::unexpected_message()),
     };
 
@@ -72,6 +48,13 @@ async fn prepare_typecheck(
         _ => return Err(Error::unexpected_message()),
     };
 
+    let mut parameters = vec![];
+    let mut it = parameter_description.parameters();
+    while let Some(oid) = it.next().map_err(Error::parse)? {
+        let type_ = Type::from_oid(oid).ok_or_else(Error::unexpected_message)?;
+        parameters.push(type_);
+    }
+
     let mut columns = vec![];
     if let Some(row_description) = row_description {
         let mut it = row_description.fields();
@@ -82,168 +65,98 @@ async fn prepare_typecheck(
         }
     }
 
-    Ok(Statement::new(name, columns))
+    Ok(Statement::new(client, name, parameters, columns))
 }
 
-fn try_from_cache(typecache: &CachedTypeInfo, oid: Oid) -> Option<Type> {
+fn encode(client: &InnerClient, name: &str, query: &str, types: &[Type]) -> Result<Bytes, Error> {
+    if types.is_empty() {
+        debug!("preparing query {}: {}", name, query);
+    } else {
+        debug!("preparing query {} with types {:?}: {}", name, types, query);
+    }
+
+    client.with_buf(|buf| {
+        frontend::parse(name, query, types.iter().map(Type::oid), buf).map_err(Error::encode)?;
+        frontend::describe(b'S', name, buf).map_err(Error::encode)?;
+        frontend::sync(buf);
+        Ok(buf.split().freeze())
+    })
+}
+
+pub async fn get_type(
+    client: &Arc<InnerClient>,
+    typecache: &mut CachedTypeInfo,
+    oid: Oid,
+) -> Result<Type, Error> {
     if let Some(type_) = Type::from_oid(oid) {
-        return Some(type_);
+        return Ok(type_);
     }
 
     if let Some(type_) = typecache.types.get(&oid) {
-        return Some(type_.clone());
+        return Ok(type_.clone());
     };
 
-    None
-}
+    let stmt = typeinfo_statement(client, typecache).await?;
 
-pub async fn parse_row_description(
-    client: &mut PartialQuery<'_>,
-    typecache: &mut CachedTypeInfo,
-    row_description: Option<RowDescriptionBody>,
-) -> Result<Vec<Column>, Error> {
-    let mut columns = vec![];
+    let rows = query::query(client, stmt, slice_iter(&[&oid])).await?;
+    pin_mut!(rows);
 
-    if let Some(row_description) = row_description {
-        let mut it = row_description.fields();
-        while let Some(field) = it.next().map_err(Error::parse)? {
-            let type_ = try_from_cache(typecache, field.type_oid()).unwrap_or(Type::UNKNOWN);
-            let column = Column::new(field.name().to_string(), type_, field);
-            columns.push(column);
-        }
-    }
-
-    let all_known = columns.iter().all(|c| c.type_ != Type::UNKNOWN);
-    if all_known {
-        // all known, return early.
-        return Ok(columns);
-    }
-
-    let typeinfo = "neon_proxy_typeinfo";
-
-    // make sure to close the typeinfo statement before exiting.
-    let mut guard = CloseStmt {
-        name: typeinfo,
-        client: None,
-    };
-    let client = guard.client.insert(client);
-
-    // get the typeinfo statement.
-    let stmt = prepare_typecheck(client, typeinfo, TYPEINFO_QUERY).await?;
-
-    for column in &mut columns {
-        column.type_ = get_type(client, typecache, &stmt, column.type_oid()).await?;
-    }
-
-    // cancel the close guard.
-    let responses = guard.close()?;
-
-    match responses.next().await? {
-        Message::CloseComplete => {}
-        _ => return Err(Error::unexpected_message()),
-    }
-
-    Ok(columns)
-}
-
-async fn get_type(
-    client: &mut PartialQuery<'_>,
-    typecache: &mut CachedTypeInfo,
-    stmt: &Statement,
-    mut oid: Oid,
-) -> Result<Type, Error> {
-    let mut stack = vec![];
-    let mut type_ = loop {
-        if let Some(type_) = try_from_cache(typecache, oid) {
-            break type_;
-        }
-
-        let row = exec(client, stmt, oid).await?;
-        if stack.len() > 8 {
-            return Err(Error::unexpected_message());
-        }
-
-        let name: String = row.try_get(0)?;
-        let type_: i8 = row.try_get(1)?;
-        let elem_oid: Oid = row.try_get(2)?;
-        let rngsubtype: Option<Oid> = row.try_get(3)?;
-        let basetype: Oid = row.try_get(4)?;
-        let schema: String = row.try_get(5)?;
-        let relid: Oid = row.try_get(6)?;
-
-        let kind = if type_ == b'e' as i8 {
-            Kind::Enum
-        } else if type_ == b'p' as i8 {
-            Kind::Pseudo
-        } else if basetype != 0 {
-            Kind::Domain(basetype)
-        } else if elem_oid != 0 {
-            stack.push((name, oid, schema));
-            oid = elem_oid;
-            continue;
-        } else if relid != 0 {
-            Kind::Composite(relid)
-        } else if let Some(rngsubtype) = rngsubtype {
-            Kind::Range(rngsubtype)
-        } else {
-            Kind::Simple
-        };
-
-        let type_ = Type::new(name, oid, kind, schema);
-        typecache.types.insert(oid, type_.clone());
-        break type_;
+    let row = match rows.try_next().await? {
+        Some(row) => row,
+        None => return Err(Error::unexpected_message()),
     };
 
-    while let Some((name, oid, schema)) = stack.pop() {
-        type_ = Type::new(name, oid, Kind::Array(type_), schema);
-        typecache.types.insert(oid, type_.clone());
-    }
+    let name: String = row.try_get(0)?;
+    let type_: i8 = row.try_get(1)?;
+    let elem_oid: Oid = row.try_get(2)?;
+    let rngsubtype: Option<Oid> = row.try_get(3)?;
+    let basetype: Oid = row.try_get(4)?;
+    let schema: String = row.try_get(5)?;
+    let relid: Oid = row.try_get(6)?;
+
+    let kind = if type_ == b'e' as i8 {
+        Kind::Enum
+    } else if type_ == b'p' as i8 {
+        Kind::Pseudo
+    } else if basetype != 0 {
+        Kind::Domain(basetype)
+    } else if elem_oid != 0 {
+        let type_ = get_type_rec(client, typecache, elem_oid).await?;
+        Kind::Array(type_)
+    } else if relid != 0 {
+        Kind::Composite(relid)
+    } else if let Some(rngsubtype) = rngsubtype {
+        let type_ = get_type_rec(client, typecache, rngsubtype).await?;
+        Kind::Range(type_)
+    } else {
+        Kind::Simple
+    };
+
+    let type_ = Type::new(name, oid, kind, schema);
+    typecache.types.insert(oid, type_.clone());
 
     Ok(type_)
 }
 
-/// exec the typeinfo statement returning one row.
-async fn exec(
-    client: &mut PartialQuery<'_>,
-    statement: &Statement,
-    param: Oid,
-) -> Result<Row, Error> {
-    let responses = client.send_with_flush(|buf| {
-        encode_bind(statement, param, "", buf);
-        frontend::execute("", 0, buf).map_err(Error::encode)?;
-        Ok(())
-    })?;
+fn get_type_rec<'a>(
+    client: &'a Arc<InnerClient>,
+    typecache: &'a mut CachedTypeInfo,
+    oid: Oid,
+) -> Pin<Box<dyn Future<Output = Result<Type, Error>> + Send + 'a>> {
+    Box::pin(get_type(client, typecache, oid))
+}
 
-    match responses.next().await? {
-        Message::BindComplete => {}
-        _ => return Err(Error::unexpected_message()),
+async fn typeinfo_statement(
+    client: &Arc<InnerClient>,
+    typecache: &mut CachedTypeInfo,
+) -> Result<Statement, Error> {
+    if let Some(stmt) = &typecache.typeinfo {
+        return Ok(stmt.clone());
     }
 
-    let row = match responses.next().await? {
-        Message::DataRow(body) => Row::new(statement.clone(), body, Format::Binary)?,
-        _ => return Err(Error::unexpected_message()),
-    };
+    let typeinfo = "neon_proxy_typeinfo";
+    let stmt = prepare_typecheck(client, typeinfo, TYPEINFO_QUERY, &[]).await?;
 
-    match responses.next().await? {
-        Message::CommandComplete(_) => {}
-        _ => return Err(Error::unexpected_message()),
-    };
-
-    Ok(row)
-}
-
-fn encode_bind(statement: &Statement, param: Oid, portal: &str, buf: &mut BytesMut) {
-    frontend::bind(
-        portal,
-        statement.name(),
-        [Format::Binary as i16],
-        [param],
-        |param, buf| {
-            oid_to_sql(param, buf);
-            Ok(IsNull::No)
-        },
-        [Format::Binary as i16],
-        buf,
-    )
-    .unwrap();
+    typecache.typeinfo = Some(stmt.clone());
+    Ok(stmt)
 }

libs/proxy/tokio-postgres2/src/query.rs

@@ -1,43 +1,76 @@
+use std::fmt;
+use std::marker::PhantomPinned;
 use std::pin::Pin;
+use std::sync::Arc;
 use std::task::{Context, Poll};
 
-use bytes::BufMut;
+use bytes::{BufMut, Bytes, BytesMut};
+use fallible_iterator::FallibleIterator;
 use futures_util::{Stream, ready};
+use pin_project_lite::pin_project;
 use postgres_protocol2::message::backend::Message;
 use postgres_protocol2::message::frontend;
-use postgres_types2::Format;
+use postgres_types2::{Format, ToSql, Type};
+use tracing::debug;
 
-use crate::client::{CachedTypeInfo, InnerClient, Responses};
-use crate::{Error, ReadyForQueryStatus, Row, Statement};
+use crate::client::{InnerClient, Responses};
+use crate::codec::FrontendMessage;
+use crate::connection::RequestMessages;
+use crate::types::IsNull;
+use crate::{Column, Error, ReadyForQueryStatus, Row, Statement};
 
-pub async fn query_txt<'a, S, I>(
-    client: &'a mut InnerClient,
-    typecache: &mut CachedTypeInfo,
+struct BorrowToSqlParamsDebug<'a>(&'a [&'a (dyn ToSql + Sync)]);
+
+impl fmt::Debug for BorrowToSqlParamsDebug<'_> {
+    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
+        f.debug_list().entries(self.0.iter()).finish()
+    }
+}
+
+pub async fn query<'a, I>(
+    client: &InnerClient,
+    statement: Statement,
+    params: I,
+) -> Result<RowStream, Error>
+where
+    I: IntoIterator<Item = &'a (dyn ToSql + Sync)>,
+    I::IntoIter: ExactSizeIterator,
+{
+    let buf = if tracing::enabled!(tracing::Level::DEBUG) {
+        let params = params.into_iter().collect::<Vec<_>>();
+        debug!(
+            "executing statement {} with parameters: {:?}",
+            statement.name(),
+            BorrowToSqlParamsDebug(params.as_slice()),
+        );
+        encode(client, &statement, params)?
+    } else {
+        encode(client, &statement, params)?
+    };
+    let responses = start(client, buf).await?;
+    Ok(RowStream {
+        statement,
+        responses,
+        command_tag: None,
+        status: ReadyForQueryStatus::Unknown,
+        output_format: Format::Binary,
+        _p: PhantomPinned,
+    })
+}
+
+pub async fn query_txt<S, I>(
+    client: &Arc<InnerClient>,
     query: &str,
     params: I,
-) -> Result<RowStream<'a>, Error>
+) -> Result<RowStream, Error>
 where
     S: AsRef<str>,
     I: IntoIterator<Item = Option<S>>,
     I::IntoIter: ExactSizeIterator,
 {
     let params = params.into_iter();
-    let mut client = client.start()?;
 
-    // Flow:
-    // 1. Parse the query
-    // 2. Inspect the row description for OIDs
-    // 3. If there's any OIDs we don't already know about, perform the typeinfo routine
-    // 4. Execute the query
-    // 5. Sync.
-    //
-    // The typeinfo routine:
-    // 1. Parse the typeinfo query
-    // 2. Execute the query on each OID
-    // 3. If the result does not match an OID we know, repeat 2.
-
-    // parse the query and get type info
-    let responses = client.send_with_flush(|buf| {
+    let buf = client.with_buf(|buf| {
         frontend::parse(
             "",                 // unnamed prepared statement
             query,              // query to parse
@@ -46,30 +79,7 @@ where
         )
         .map_err(Error::encode)?;
         frontend::describe(b'S', "", buf).map_err(Error::encode)?;
-        Ok(())
-    })?;
-
-    match responses.next().await? {
-        Message::ParseComplete => {}
-        _ => return Err(Error::unexpected_message()),
-    }
-
-    match responses.next().await? {
-        Message::ParameterDescription(_) => {}
-        _ => return Err(Error::unexpected_message()),
-    };
-
-    let row_description = match responses.next().await? {
-        Message::RowDescription(body) => Some(body),
-        Message::NoData => None,
-        _ => return Err(Error::unexpected_message()),
-    };
-
-    let columns =
-        crate::prepare::parse_row_description(&mut client, typecache, row_description).await?;
-
-    let responses = client.send_with_sync(|buf| {
-        // Bind, pass params as text, retrieve as text
+        // Bind, pass params as text, retrieve as binary
         match frontend::bind(
             "",                 // empty string selects the unnamed portal
             "",                 // unnamed prepared statement
@@ -92,55 +102,173 @@ where
 
         // Execute
         frontend::execute("", 0, buf).map_err(Error::encode)?;
+        // Sync
+        frontend::sync(buf);
 
-        Ok(())
+        Ok(buf.split().freeze())
     })?;
 
+    // now read the responses
+    let mut responses = client.send(RequestMessages::Single(FrontendMessage::Raw(buf)))?;
+
+    match responses.next().await? {
+        Message::ParseComplete => {}
+        _ => return Err(Error::unexpected_message()),
+    }
+
+    let parameter_description = match responses.next().await? {
+        Message::ParameterDescription(body) => body,
+        _ => return Err(Error::unexpected_message()),
+    };
+
+    let row_description = match responses.next().await? {
+        Message::RowDescription(body) => Some(body),
+        Message::NoData => None,
+        _ => return Err(Error::unexpected_message()),
+    };
+
     match responses.next().await? {
         Message::BindComplete => {}
         _ => return Err(Error::unexpected_message()),
     }
 
+    let mut parameters = vec![];
+    let mut it = parameter_description.parameters();
+    while let Some(oid) = it.next().map_err(Error::parse)? {
+        let type_ = Type::from_oid(oid).unwrap_or(Type::UNKNOWN);
+        parameters.push(type_);
+    }
+
+    let mut columns = vec![];
+    if let Some(row_description) = row_description {
+        let mut it = row_description.fields();
+        while let Some(field) = it.next().map_err(Error::parse)? {
+            let type_ = Type::from_oid(field.type_oid()).unwrap_or(Type::UNKNOWN);
+            let column = Column::new(field.name().to_string(), type_, field);
+            columns.push(column);
+        }
+    }
+
     Ok(RowStream {
+        statement: Statement::new_anonymous(parameters, columns),
         responses,
-        statement: Statement::new("", columns),
         command_tag: None,
         status: ReadyForQueryStatus::Unknown,
         output_format: Format::Text,
+        _p: PhantomPinned,
     })
 }
 
-/// A stream of table rows.
-pub struct RowStream<'a> {
-    responses: &'a mut Responses,
-    output_format: Format,
-    pub statement: Statement,
-    pub command_tag: Option<String>,
-    pub status: ReadyForQueryStatus,
+async fn start(client: &InnerClient, buf: Bytes) -> Result<Responses, Error> {
+    let mut responses = client.send(RequestMessages::Single(FrontendMessage::Raw(buf)))?;
+
+    match responses.next().await? {
+        Message::BindComplete => {}
+        _ => return Err(Error::unexpected_message()),
+    }
+
+    Ok(responses)
 }
 
-impl Stream for RowStream<'_> {
+pub fn encode<'a, I>(client: &InnerClient, statement: &Statement, params: I) -> Result<Bytes, Error>
+where
+    I: IntoIterator<Item = &'a (dyn ToSql + Sync)>,
+    I::IntoIter: ExactSizeIterator,
+{
+    client.with_buf(|buf| {
+        encode_bind(statement, params, "", buf)?;
+        frontend::execute("", 0, buf).map_err(Error::encode)?;
+        frontend::sync(buf);
+        Ok(buf.split().freeze())
+    })
+}
+
+pub fn encode_bind<'a, I>(
+    statement: &Statement,
+    params: I,
+    portal: &str,
+    buf: &mut BytesMut,
+) -> Result<(), Error>
+where
+    I: IntoIterator<Item = &'a (dyn ToSql + Sync)>,
+    I::IntoIter: ExactSizeIterator,
+{
+    let param_types = statement.params();
+    let params = params.into_iter();
+
+    assert!(
+        param_types.len() == params.len(),
+        "expected {} parameters but got {}",
+        param_types.len(),
+        params.len()
+    );
+
+    let (param_formats, params): (Vec<_>, Vec<_>) = params
+        .zip(param_types.iter())
+        .map(|(p, ty)| (p.encode_format(ty) as i16, p))
+        .unzip();
+
+    let params = params.into_iter();
+
+    let mut error_idx = 0;
+    let r = frontend::bind(
+        portal,
+        statement.name(),
+        param_formats,
+        params.zip(param_types).enumerate(),
+        |(idx, (param, ty)), buf| match param.to_sql_checked(ty, buf) {
+            Ok(IsNull::No) => Ok(postgres_protocol2::IsNull::No),
+            Ok(IsNull::Yes) => Ok(postgres_protocol2::IsNull::Yes),
+            Err(e) => {
+                error_idx = idx;
+                Err(e)
+            }
+        },
+        Some(1),
+        buf,
+    );
+    match r {
+        Ok(()) => Ok(()),
+        Err(frontend::BindError::Conversion(e)) => Err(Error::to_sql(e, error_idx)),
+        Err(frontend::BindError::Serialization(e)) => Err(Error::encode(e)),
+    }
+}
+
+pin_project! {
+    /// A stream of table rows.
+    pub struct RowStream {
+        statement: Statement,
+        responses: Responses,
+        command_tag: Option<String>,
+        output_format: Format,
+        status: ReadyForQueryStatus,
+        #[pin]
+        _p: PhantomPinned,
+    }
+}
+
+impl Stream for RowStream {
     type Item = Result<Row, Error>;
 
     fn poll_next(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Option<Self::Item>> {
-        let this = self.get_mut();
+        let this = self.project();
         loop {
             match ready!(this.responses.poll_next(cx)?) {
                 Message::DataRow(body) => {
                     return Poll::Ready(Some(Ok(Row::new(
                         this.statement.clone(),
                         body,
-                        this.output_format,
+                        *this.output_format,
                     )?)));
                 }
                 Message::EmptyQueryResponse | Message::PortalSuspended => {}
                 Message::CommandComplete(body) => {
                     if let Ok(tag) = body.tag() {
-                        this.command_tag = Some(tag.to_string());
+                        *this.command_tag = Some(tag.to_string());
                     }
                 }
                 Message::ReadyForQuery(status) => {
-                    this.status = status.into();
+                    *this.status = status.into();
                     return Poll::Ready(None);
                 }
                 _ => return Poll::Ready(Some(Err(Error::unexpected_message()))),
@@ -148,3 +276,24 @@ impl Stream for RowStream<'_> {
         }
     }
 }
+
+impl RowStream {
+    /// Returns information about the columns of data in the row.
+    pub fn columns(&self) -> &[Column] {
+        self.statement.columns()
+    }
+
+    /// Returns the command tag of this query.
+    ///
+    /// This is only available after the stream has been exhausted.
+    pub fn command_tag(&self) -> Option<String> {
+        self.command_tag.clone()
+    }
+
+    /// Returns if the connection is ready for querying, with the status of the connection.
+    ///
+    /// This might be available only after the stream has been exhausted.
+    pub fn ready_status(&self) -> ReadyForQueryStatus {
+        self.status
+    }
+}

libs/proxy/tokio-postgres2/src/simple_query.rs

@@ -1,14 +1,19 @@
+use std::marker::PhantomPinned;
 use std::pin::Pin;
 use std::sync::Arc;
 use std::task::{Context, Poll};
 
+use bytes::Bytes;
 use fallible_iterator::FallibleIterator;
 use futures_util::{Stream, ready};
 use pin_project_lite::pin_project;
 use postgres_protocol2::message::backend::Message;
+use postgres_protocol2::message::frontend;
 use tracing::debug;
 
 use crate::client::{InnerClient, Responses};
+use crate::codec::FrontendMessage;
+use crate::connection::RequestMessages;
 use crate::{Error, ReadyForQueryStatus, SimpleQueryMessage, SimpleQueryRow};
 
 /// Information about a column of a single query row.
@@ -28,28 +33,28 @@ impl SimpleColumn {
     }
 }
 
-pub async fn simple_query<'a>(
-    client: &'a mut InnerClient,
-    query: &str,
-) -> Result<SimpleQueryStream<'a>, Error> {
+pub async fn simple_query(client: &InnerClient, query: &str) -> Result<SimpleQueryStream, Error> {
     debug!("executing simple query: {}", query);
 
-    let responses = client.send_simple_query(query)?;
+    let buf = encode(client, query)?;
+    let responses = client.send(RequestMessages::Single(FrontendMessage::Raw(buf)))?;
 
     Ok(SimpleQueryStream {
         responses,
         columns: None,
         status: ReadyForQueryStatus::Unknown,
+        _p: PhantomPinned,
     })
 }
 
 pub async fn batch_execute(
-    client: &mut InnerClient,
+    client: &InnerClient,
     query: &str,
 ) -> Result<ReadyForQueryStatus, Error> {
     debug!("executing statement batch: {}", query);
 
-    let responses = client.send_simple_query(query)?;
+    let buf = encode(client, query)?;
+    let mut responses = client.send(RequestMessages::Single(FrontendMessage::Raw(buf)))?;
 
     loop {
         match responses.next().await? {
@@ -63,16 +68,25 @@ pub async fn batch_execute(
     }
 }
 
+pub(crate) fn encode(client: &InnerClient, query: &str) -> Result<Bytes, Error> {
+    client.with_buf(|buf| {
+        frontend::query(query, buf).map_err(Error::encode)?;
+        Ok(buf.split().freeze())
+    })
+}
+
 pin_project! {
     /// A stream of simple query results.
-    pub struct SimpleQueryStream<'a> {
-        responses: &'a mut Responses,
+    pub struct SimpleQueryStream {
+        responses: Responses,
         columns: Option<Arc<[SimpleColumn]>>,
         status: ReadyForQueryStatus,
+        #[pin]
+        _p: PhantomPinned,
     }
 }
 
-impl SimpleQueryStream<'_> {
+impl SimpleQueryStream {
     /// Returns if the connection is ready for querying, with the status of the connection.
     ///
     /// This might be available only after the stream has been exhausted.
@@ -81,7 +95,7 @@ impl SimpleQueryStream<'_> {
     }
 }
 
-impl Stream for SimpleQueryStream<'_> {
+impl Stream for SimpleQueryStream {
     type Item = Result<SimpleQueryMessage, Error>;
 
     fn poll_next(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Option<Self::Item>> {

libs/proxy/tokio-postgres2/src/statement.rs

@@ -1,15 +1,35 @@
 use std::fmt;
-use std::sync::Arc;
+use std::sync::{Arc, Weak};
 
-use crate::types::Type;
 use postgres_protocol2::Oid;
 use postgres_protocol2::message::backend::Field;
+use postgres_protocol2::message::frontend;
+
+use crate::client::InnerClient;
+use crate::codec::FrontendMessage;
+use crate::connection::RequestMessages;
+use crate::types::Type;
 
 struct StatementInner {
+    client: Weak<InnerClient>,
     name: &'static str,
+    params: Vec<Type>,
     columns: Vec<Column>,
 }
 
+impl Drop for StatementInner {
+    fn drop(&mut self) {
+        if let Some(client) = self.client.upgrade() {
+            let buf = client.with_buf(|buf| {
+                frontend::close(b'S', self.name, buf).unwrap();
+                frontend::sync(buf);
+                buf.split().freeze()
+            });
+            let _ = client.send(RequestMessages::Single(FrontendMessage::Raw(buf)));
+        }
+    }
+}
+
 /// A prepared statement.
 ///
 /// Prepared statements can only be used with the connection that created them.
@@ -17,14 +37,38 @@ struct StatementInner {
 pub struct Statement(Arc<StatementInner>);
 
 impl Statement {
-    pub(crate) fn new(name: &'static str, columns: Vec<Column>) -> Statement {
-        Statement(Arc::new(StatementInner { name, columns }))
+    pub(crate) fn new(
+        inner: &Arc<InnerClient>,
+        name: &'static str,
+        params: Vec<Type>,
+        columns: Vec<Column>,
+    ) -> Statement {
+        Statement(Arc::new(StatementInner {
+            client: Arc::downgrade(inner),
+            name,
+            params,
+            columns,
+        }))
+    }
+
+    pub(crate) fn new_anonymous(params: Vec<Type>, columns: Vec<Column>) -> Statement {
+        Statement(Arc::new(StatementInner {
+            client: Weak::new(),
+            name: "<anonymous>",
+            params,
+            columns,
+        }))
     }
 
     pub(crate) fn name(&self) -> &str {
         self.0.name
     }
 
+    /// Returns the expected types of the statement's parameters.
+    pub fn params(&self) -> &[Type] {
+        &self.0.params
+    }
+
     /// Returns information about the columns returned when the statement is queried.
     pub fn columns(&self) -> &[Column] {
         &self.0.columns
@@ -34,7 +78,7 @@ impl Statement {
 /// Information about a column of a query.
 pub struct Column {
     name: String,
-    pub(crate) type_: Type,
+    type_: Type,
 
     // raw fields from RowDescription
     table_oid: Oid,

libs/proxy/tokio-postgres2/src/transaction.rs

@@ -1,3 +1,7 @@
+use postgres_protocol2::message::frontend;
+
+use crate::codec::FrontendMessage;
+use crate::connection::RequestMessages;
 use crate::query::RowStream;
 use crate::{CancelToken, Client, Error, ReadyForQueryStatus};
 
@@ -16,7 +20,14 @@ impl Drop for Transaction<'_> {
             return;
         }
 
-        let _ = self.client.inner_mut().send_simple_query("ROLLBACK");
+        let buf = self.client.inner().with_buf(|buf| {
+            frontend::query("ROLLBACK", buf).unwrap();
+            buf.split().freeze()
+        });
+        let _ = self
+            .client
+            .inner()
+            .send(RequestMessages::Single(FrontendMessage::Raw(buf)));
     }
 }
 
@@ -43,11 +54,7 @@ impl<'a> Transaction<'a> {
     }
 
     /// Like `Client::query_raw_txt`.
-    pub async fn query_raw_txt<S, I>(
-        &mut self,
-        statement: &str,
-        params: I,
-    ) -> Result<RowStream, Error>
+    pub async fn query_raw_txt<S, I>(&self, statement: &str, params: I) -> Result<RowStream, Error>
     where
         S: AsRef<str>,
         I: IntoIterator<Item = Option<S>>,

pageserver/Cargo.toml

@@ -43,7 +43,6 @@ nix.workspace = true
 num_cpus.workspace = true
 num-traits.workspace = true
 once_cell.workspace = true
-pageserver_page_api.workspace = true
 pin-project-lite.workspace = true
 postgres_backend.workspace = true
 postgres-protocol.workspace = true
@@ -72,8 +71,6 @@ tokio-rustls.workspace = true
 tokio-stream.workspace = true
 tokio-util.workspace = true
 toml_edit = { workspace = true, features = [ "serde" ] }
-tonic.workspace = true
-tonic-reflection.workspace = true
 tracing.workspace = true
 tracing-utils.workspace = true
 url.workspace = true

pageserver/src/basebackup_cache.rs

@@ -1,518 +0,0 @@
-use std::{collections::HashMap, sync::Arc};
-
-use async_compression::tokio::write::GzipEncoder;
-use camino::{Utf8Path, Utf8PathBuf};
-use metrics::core::{AtomicU64, GenericCounter};
-use pageserver_api::{config::BasebackupCacheConfig, models::TenantState};
-use tokio::{
-    io::{AsyncWriteExt, BufWriter},
-    sync::mpsc::{UnboundedReceiver, UnboundedSender},
-};
-use tokio_util::sync::CancellationToken;
-use utils::{
-    id::{TenantId, TenantTimelineId, TimelineId},
-    lsn::Lsn,
-    shard::TenantShardId,
-};
-
-use crate::{
-    basebackup::send_basebackup_tarball,
-    context::{DownloadBehavior, RequestContext},
-    metrics::{BASEBACKUP_CACHE_ENTRIES, BASEBACKUP_CACHE_PREPARE, BASEBACKUP_CACHE_READ},
-    task_mgr::TaskKind,
-    tenant::{
-        Timeline,
-        mgr::{TenantManager, TenantSlot},
-    },
-};
-
-pub struct BasebackupPrepareRequest {
-    pub tenant_shard_id: TenantShardId,
-    pub timeline_id: TimelineId,
-    pub lsn: Lsn,
-}
-
-pub type BasebackupPrepareSender = UnboundedSender<BasebackupPrepareRequest>;
-pub type BasebackupPrepareReceiver = UnboundedReceiver<BasebackupPrepareRequest>;
-
-type BasebackupRemoveEntrySender = UnboundedSender<Utf8PathBuf>;
-type BasebackupRemoveEntryReceiver = UnboundedReceiver<Utf8PathBuf>;
-
-/// BasebackupCache stores cached basebackup archives for timelines on local disk.
-///
-/// The main purpose of this cache is to speed up the startup process of compute nodes
-/// after scaling to zero.
-/// Thus, the basebackup is stored only for the latest LSN of the timeline and with
-/// fixed set of parameters (gzip=true, full_backup=false, replica=false, prev_lsn=none).
-///
-/// The cache receives prepare requests through the `BasebackupPrepareSender` channel,
-/// generates a basebackup from the timeline in the background, and stores it on disk.
-///
-/// Basebackup requests are pretty rare. We expect ~thousands of entries in the cache
-/// and ~1 RPS for get requests.
-pub struct BasebackupCache {
-    data_dir: Utf8PathBuf,
-    config: BasebackupCacheConfig,
-    tenant_manager: Arc<TenantManager>,
-    remove_entry_sender: BasebackupRemoveEntrySender,
-
-    entries: std::sync::Mutex<HashMap<TenantTimelineId, Lsn>>,
-
-    cancel: CancellationToken,
-
-    read_hit_count: GenericCounter<AtomicU64>,
-    read_miss_count: GenericCounter<AtomicU64>,
-    read_err_count: GenericCounter<AtomicU64>,
-
-    prepare_ok_count: GenericCounter<AtomicU64>,
-    prepare_skip_count: GenericCounter<AtomicU64>,
-    prepare_err_count: GenericCounter<AtomicU64>,
-}
-
-impl BasebackupCache {
-    /// Creates a BasebackupCache and spawns the background task.
-    /// The initialization of the cache is performed in the background and does not
-    /// block the caller. The cache will return `None` for any get requests until
-    /// initialization is complete.
-    pub fn spawn(
-        runtime_handle: &tokio::runtime::Handle,
-        data_dir: Utf8PathBuf,
-        config: Option<BasebackupCacheConfig>,
-        prepare_receiver: BasebackupPrepareReceiver,
-        tenant_manager: Arc<TenantManager>,
-        cancel: CancellationToken,
-    ) -> Arc<Self> {
-        let (remove_entry_sender, remove_entry_receiver) = tokio::sync::mpsc::unbounded_channel();
-
-        let enabled = config.is_some();
-
-        let cache = Arc::new(BasebackupCache {
-            data_dir,
-            config: config.unwrap_or_default(),
-            tenant_manager,
-            remove_entry_sender,
-
-            entries: std::sync::Mutex::new(HashMap::new()),
-
-            cancel,
-
-            read_hit_count: BASEBACKUP_CACHE_READ.with_label_values(&["hit"]),
-            read_miss_count: BASEBACKUP_CACHE_READ.with_label_values(&["miss"]),
-            read_err_count: BASEBACKUP_CACHE_READ.with_label_values(&["error"]),
-
-            prepare_ok_count: BASEBACKUP_CACHE_PREPARE.with_label_values(&["ok"]),
-            prepare_skip_count: BASEBACKUP_CACHE_PREPARE.with_label_values(&["skip"]),
-            prepare_err_count: BASEBACKUP_CACHE_PREPARE.with_label_values(&["error"]),
-        });
-
-        if enabled {
-            runtime_handle.spawn(
-                cache
-                    .clone()
-                    .background(prepare_receiver, remove_entry_receiver),
-            );
-        }
-
-        cache
-    }
-
-    /// Gets a basebackup entry from the cache.
-    /// If the entry is found, opens a file with the basebackup archive and returns it.
-    /// The open file descriptor will prevent the file system from deleting the file
-    /// even if the entry is removed from the cache in the background.
-    pub async fn get(
-        &self,
-        tenant_id: TenantId,
-        timeline_id: TimelineId,
-        lsn: Lsn,
-    ) -> Option<tokio::fs::File> {
-        // Fast path. Check if the entry exists using the in-memory state.
-        let tti = TenantTimelineId::new(tenant_id, timeline_id);
-        if self.entries.lock().unwrap().get(&tti) != Some(&lsn) {
-            self.read_miss_count.inc();
-            return None;
-        }
-
-        let path = self.entry_path(tenant_id, timeline_id, lsn);
-
-        match tokio::fs::File::open(path).await {
-            Ok(file) => {
-                self.read_hit_count.inc();
-                Some(file)
-            }
-            Err(e) => {
-                if e.kind() == std::io::ErrorKind::NotFound {
-                    // We may end up here if the basebackup was concurrently removed by the cleanup task.
-                    self.read_miss_count.inc();
-                } else {
-                    self.read_err_count.inc();
-                    tracing::warn!("Unexpected error opening basebackup cache file: {:?}", e);
-                }
-                None
-            }
-        }
-    }
-
-    // Private methods.
-
-    fn entry_filename(tenant_id: TenantId, timeline_id: TimelineId, lsn: Lsn) -> String {
-        // The default format for LSN is 0/ABCDEF.
-        // The backslash is not filename friendly, so serialize it as plain hex.
-        let lsn = lsn.0;
-        format!("basebackup_{tenant_id}_{timeline_id}_{lsn:016X}.tar.gz")
-    }
-
-    fn entry_path(&self, tenant_id: TenantId, timeline_id: TimelineId, lsn: Lsn) -> Utf8PathBuf {
-        self.data_dir
-            .join(Self::entry_filename(tenant_id, timeline_id, lsn))
-    }
-
-    fn entry_tmp_path(
-        &self,
-        tenant_id: TenantId,
-        timeline_id: TimelineId,
-        lsn: Lsn,
-    ) -> Utf8PathBuf {
-        self.data_dir
-            .join("tmp")
-            .join(Self::entry_filename(tenant_id, timeline_id, lsn))
-    }
-
-    fn parse_entry_filename(filename: &str) -> Option<(TenantId, TimelineId, Lsn)> {
-        let parts: Vec<&str> = filename
-            .strip_prefix("basebackup_")?
-            .strip_suffix(".tar.gz")?
-            .split('_')
-            .collect();
-        if parts.len() != 3 {
-            return None;
-        }
-        let tenant_id = parts[0].parse::<TenantId>().ok()?;
-        let timeline_id = parts[1].parse::<TimelineId>().ok()?;
-        let lsn = Lsn(u64::from_str_radix(parts[2], 16).ok()?);
-
-        Some((tenant_id, timeline_id, lsn))
-    }
-
-    async fn cleanup(&self) -> anyhow::Result<()> {
-        // Cleanup tmp directory.
-        let tmp_dir = self.data_dir.join("tmp");
-        let mut tmp_dir = tokio::fs::read_dir(&tmp_dir).await?;
-        while let Some(dir_entry) = tmp_dir.next_entry().await? {
-            if let Err(e) = tokio::fs::remove_file(dir_entry.path()).await {
-                tracing::warn!("Failed to remove basebackup cache tmp file: {:#}", e);
-            }
-        }
-
-        // Remove outdated entries.
-        let entries_old = self.entries.lock().unwrap().clone();
-        let mut entries_new = HashMap::new();
-        for (tenant_shard_id, tenant_slot) in self.tenant_manager.list() {
-            if !tenant_shard_id.is_shard_zero() {
-                continue;
-            }
-            let TenantSlot::Attached(tenant) = tenant_slot else {
-                continue;
-            };
-            let tenant_id = tenant_shard_id.tenant_id;
-
-            for timeline in tenant.list_timelines() {
-                let tti = TenantTimelineId::new(tenant_id, timeline.timeline_id);
-                if let Some(&entry_lsn) = entries_old.get(&tti) {
-                    if timeline.get_last_record_lsn() <= entry_lsn {
-                        entries_new.insert(tti, entry_lsn);
-                    }
-                }
-            }
-        }
-
-        for (&tti, &lsn) in entries_old.iter() {
-            if !entries_new.contains_key(&tti) {
-                self.remove_entry_sender
-                    .send(self.entry_path(tti.tenant_id, tti.timeline_id, lsn))
-                    .unwrap();
-            }
-        }
-
-        BASEBACKUP_CACHE_ENTRIES.set(entries_new.len() as i64);
-        *self.entries.lock().unwrap() = entries_new;
-
-        Ok(())
-    }
-
-    async fn on_startup(&self) -> anyhow::Result<()> {
-        // Create data_dir and tmp directory if they do not exist.
-        tokio::fs::create_dir_all(&self.data_dir.join("tmp"))
-            .await
-            .map_err(|e| {
-                anyhow::anyhow!(
-                    "Failed to create basebackup cache data_dir {:?}: {:?}",
-                    self.data_dir,
-                    e
-                )
-            })?;
-
-        // Read existing entries from the data_dir and add them to in-memory state.
-        let mut entries = HashMap::new();
-        let mut dir = tokio::fs::read_dir(&self.data_dir).await?;
-        while let Some(dir_entry) = dir.next_entry().await? {
-            let filename = dir_entry.file_name();
-
-            if filename == "tmp" {
-                // Skip the tmp directory.
-                continue;
-            }
-
-            let parsed = Self::parse_entry_filename(filename.to_string_lossy().as_ref());
-            let Some((tenant_id, timeline_id, lsn)) = parsed else {
-                tracing::warn!("Invalid basebackup cache file name: {:?}", filename);
-                continue;
-            };
-
-            let tti = TenantTimelineId::new(tenant_id, timeline_id);
-
-            use std::collections::hash_map::Entry::*;
-
-            match entries.entry(tti) {
-                Occupied(mut entry) => {
-                    let entry_lsn = *entry.get();
-                    // Leave only the latest entry, remove the old one.
-                    if lsn < entry_lsn {
-                        self.remove_entry_sender.send(self.entry_path(
-                            tenant_id,
-                            timeline_id,
-                            lsn,
-                        ))?;
-                    } else if lsn > entry_lsn {
-                        self.remove_entry_sender.send(self.entry_path(
-                            tenant_id,
-                            timeline_id,
-                            entry_lsn,
-                        ))?;
-                        entry.insert(lsn);
-                    } else {
-                        // Two different filenames parsed to the same timline_id and LSN.
-                        // Should never happen.
-                        return Err(anyhow::anyhow!(
-                            "Duplicate basebackup cache entry with the same LSN: {:?}",
-                            filename
-                        ));
-                    }
-                }
-                Vacant(entry) => {
-                    entry.insert(lsn);
-                }
-            }
-        }
-
-        BASEBACKUP_CACHE_ENTRIES.set(entries.len() as i64);
-        *self.entries.lock().unwrap() = entries;
-
-        Ok(())
-    }
-
-    async fn background(
-        self: Arc<Self>,
-        mut prepare_receiver: BasebackupPrepareReceiver,
-        mut remove_entry_receiver: BasebackupRemoveEntryReceiver,
-    ) {
-        // Panic in the background is a safe fallback.
-        // It will drop receivers and the cache will be effectively disabled.
-        self.on_startup()
-            .await
-            .expect("Failed to initialize basebackup cache");
-
-        let mut cleanup_ticker = tokio::time::interval(self.config.cleanup_period);
-        cleanup_ticker.set_missed_tick_behavior(tokio::time::MissedTickBehavior::Skip);
-
-        loop {
-            tokio::select! {
-                Some(req) = prepare_receiver.recv() => {
-                    if let Err(err) = self.prepare_basebackup(
-                        req.tenant_shard_id,
-                        req.timeline_id,
-                        req.lsn,
-                    ).await {
-                        tracing::info!("Failed to prepare basebackup: {:#}", err);
-                        self.prepare_err_count.inc();
-                        continue;
-                    }
-                }
-                Some(req) = remove_entry_receiver.recv() => {
-                    if let Err(e) = tokio::fs::remove_file(req).await {
-                        tracing::warn!("Failed to remove basebackup cache file: {:#}", e);
-                    }
-                }
-                _ = cleanup_ticker.tick() => {
-                    self.cleanup().await.unwrap_or_else(|e| {
-                        tracing::warn!("Failed to clean up basebackup cache: {:#}", e);
-                    });
-                }
-                _ = self.cancel.cancelled() => {
-                    tracing::info!("BasebackupCache background task cancelled");
-                    break;
-                }
-            }
-        }
-    }
-
-    /// Prepare a basebackup for the given timeline.
-    ///
-    /// If the basebackup already exists with a higher LSN or the timeline already
-    /// has a higher last_record_lsn, skip the preparation.
-    ///
-    /// The basebackup is prepared in a temporary directory and then moved to the final
-    /// location to make the operation atomic.
-    async fn prepare_basebackup(
-        &self,
-        tenant_shard_id: TenantShardId,
-        timeline_id: TimelineId,
-        req_lsn: Lsn,
-    ) -> anyhow::Result<()> {
-        tracing::info!(
-            tenant_id = %tenant_shard_id.tenant_id,
-            %timeline_id,
-            %req_lsn,
-            "Preparing basebackup for timeline",
-        );
-
-        let tti = TenantTimelineId::new(tenant_shard_id.tenant_id, timeline_id);
-
-        {
-            let entries = self.entries.lock().unwrap();
-            if let Some(&entry_lsn) = entries.get(&tti) {
-                if entry_lsn >= req_lsn {
-                    tracing::info!(
-                        %timeline_id,
-                        %req_lsn,
-                        %entry_lsn,
-                        "Basebackup entry already exists for timeline with higher LSN, skipping basebackup",
-                    );
-                    self.prepare_skip_count.inc();
-                    return Ok(());
-                }
-            }
-
-            if entries.len() as i64 >= self.config.max_size_entries {
-                tracing::info!(
-                    %timeline_id,
-                    %req_lsn,
-                    "Basebackup cache is full, skipping basebackup",
-                );
-                self.prepare_skip_count.inc();
-                return Ok(());
-            }
-        }
-
-        let tenant = self
-            .tenant_manager
-            .get_attached_tenant_shard(tenant_shard_id)?;
-
-        let tenant_state = tenant.current_state();
-        if tenant_state != TenantState::Active {
-            anyhow::bail!(
-                "Tenant {} is not active, current state: {:?}",
-                tenant_shard_id.tenant_id,
-                tenant_state
-            )
-        }
-
-        let timeline = tenant.get_timeline(timeline_id, true)?;
-
-        let last_record_lsn = timeline.get_last_record_lsn();
-        if last_record_lsn > req_lsn {
-            tracing::info!(
-                %timeline_id,
-                %req_lsn,
-                %last_record_lsn,
-                "Timeline has a higher LSN than the requested one, skipping basebackup",
-            );
-            self.prepare_skip_count.inc();
-            return Ok(());
-        }
-
-        let entry_tmp_path = self.entry_tmp_path(tenant_shard_id.tenant_id, timeline_id, req_lsn);
-
-        let res = self
-            .prepare_basebackup_tmp(&entry_tmp_path, &timeline, req_lsn)
-            .await;
-
-        if let Err(err) = res {
-            tracing::info!("Failed to prepare basebackup tmp file: {:#}", err);
-            // Try to clean up tmp file. If we fail, the background clean up task will take care of it.
-            match tokio::fs::remove_file(&entry_tmp_path).await {
-                Ok(_) => {}
-                Err(e) if e.kind() == std::io::ErrorKind::NotFound => {}
-                Err(e) => {
-                    tracing::info!("Failed to remove basebackup tmp file: {:?}", e);
-                }
-            }
-            return Err(err);
-        }
-
-        // Move the tmp file to the final location atomically.
-        let entry_path = self.entry_path(tenant_shard_id.tenant_id, timeline_id, req_lsn);
-        tokio::fs::rename(&entry_tmp_path, &entry_path).await?;
-
-        let mut entries = self.entries.lock().unwrap();
-        if let Some(old_lsn) = entries.insert(tti, req_lsn) {
-            // Remove the old entry if it exists.
-            self.remove_entry_sender
-                .send(self.entry_path(tenant_shard_id.tenant_id, timeline_id, old_lsn))
-                .unwrap();
-        }
-        BASEBACKUP_CACHE_ENTRIES.set(entries.len() as i64);
-
-        self.prepare_ok_count.inc();
-        Ok(())
-    }
-
-    /// Prepares a basebackup in a temporary file.
-    async fn prepare_basebackup_tmp(
-        &self,
-        emptry_tmp_path: &Utf8Path,
-        timeline: &Arc<Timeline>,
-        req_lsn: Lsn,
-    ) -> anyhow::Result<()> {
-        let ctx = RequestContext::new(TaskKind::BasebackupCache, DownloadBehavior::Download);
-        let ctx = ctx.with_scope_timeline(timeline);
-
-        let file = tokio::fs::File::create(emptry_tmp_path).await?;
-        let mut writer = BufWriter::new(file);
-
-        let mut encoder = GzipEncoder::with_quality(
-            &mut writer,
-            // Level::Best because compression is not on the hot path of basebackup requests.
-            // The decompression is almost not affected by the compression level.
-            async_compression::Level::Best,
-        );
-
-        // We may receive a request before the WAL record is applied to the timeline.
-        // Wait for the requested LSN to be applied.
-        timeline
-            .wait_lsn(
-                req_lsn,
-                crate::tenant::timeline::WaitLsnWaiter::BaseBackupCache,
-                crate::tenant::timeline::WaitLsnTimeout::Default,
-                &ctx,
-            )
-            .await?;
-
-        send_basebackup_tarball(
-            &mut encoder,
-            timeline,
-            Some(req_lsn),
-            None,
-            false,
-            false,
-            &ctx,
-        )
-        .await?;
-
-        encoder.shutdown().await?;
-        writer.flush().await?;
-        writer.into_inner().sync_all().await?;
-
-        Ok(())
-    }
-}

pageserver/src/bin/pageserver.rs

@@ -16,7 +16,6 @@ use http_utils::tls_certs::ReloadingCertificateResolver;
 use metrics::launch_timestamp::{LaunchTimestamp, set_launch_timestamp_metric};
 use metrics::set_build_info_metric;
 use nix::sys::socket::{setsockopt, sockopt};
-use pageserver::basebackup_cache::BasebackupCache;
 use pageserver::config::{PageServerConf, PageserverIdentity, ignored_fields};
 use pageserver::controller_upcall_client::StorageControllerUpcallClient;
 use pageserver::deletion_queue::DeletionQueue;
@@ -388,30 +387,23 @@ fn start_pageserver(
     // We need to release the lock file only when the process exits.
     std::mem::forget(lock_file);
 
-    // Bind the HTTP, libpq, and gRPC ports early, to error out if they are
-    // already in use.
-    info!(
-        "Starting pageserver http handler on {} with auth {:#?}",
-        conf.listen_http_addr, conf.http_auth_type
-    );
-    let http_listener = tcp_listener::bind(&conf.listen_http_addr)?;
+    // Bind the HTTP and libpq ports early, so that if they are in use by some other
+    // process, we error out early.
+    let http_addr = &conf.listen_http_addr;
+    info!("Starting pageserver http handler on {http_addr}");
+    let http_listener = tcp_listener::bind(http_addr)?;
 
     let https_listener = match conf.listen_https_addr.as_ref() {
         Some(https_addr) => {
-            info!(
-                "Starting pageserver https handler on {https_addr} with auth {:#?}",
-                conf.http_auth_type
-            );
+            info!("Starting pageserver https handler on {https_addr}");
             Some(tcp_listener::bind(https_addr)?)
         }
         None => None,
     };
 
-    info!(
-        "Starting pageserver pg protocol handler on {} with auth {:#?}",
-        conf.listen_pg_addr, conf.pg_auth_type,
-    );
-    let pageserver_listener = tcp_listener::bind(&conf.listen_pg_addr)?;
+    let pg_addr = &conf.listen_pg_addr;
+    info!("Starting pageserver pg protocol handler on {pg_addr}");
+    let pageserver_listener = tcp_listener::bind(pg_addr)?;
 
     // Enable SO_KEEPALIVE on the socket, to detect dead connections faster.
     // These are configured via net.ipv4.tcp_keepalive_* sysctls.
@@ -420,15 +412,6 @@ fn start_pageserver(
     // support enabling keepalives while using the default OS sysctls.
     setsockopt(&pageserver_listener, sockopt::KeepAlive, &true)?;
 
-    let mut grpc_listener = None;
-    if let Some(grpc_addr) = &conf.listen_grpc_addr {
-        info!(
-            "Starting pageserver gRPC handler on {grpc_addr} with auth {:#?}",
-            conf.grpc_auth_type
-        );
-        grpc_listener = Some(tcp_listener::bind(grpc_addr).map_err(|e| anyhow!("{e}"))?);
-    }
-
     // Launch broker client
     // The storage_broker::connect call needs to happen inside a tokio runtime thread.
     let broker_client = WALRECEIVER_RUNTIME
@@ -456,8 +439,7 @@ fn start_pageserver(
     // Initialize authentication for incoming connections
     let http_auth;
     let pg_auth;
-    let grpc_auth;
-    if [conf.http_auth_type, conf.pg_auth_type, conf.grpc_auth_type].contains(&AuthType::NeonJWT) {
+    if conf.http_auth_type == AuthType::NeonJWT || conf.pg_auth_type == AuthType::NeonJWT {
         // unwrap is ok because check is performed when creating config, so path is set and exists
         let key_path = conf.auth_validation_public_key_path.as_ref().unwrap();
         info!("Loading public key(s) for verifying JWT tokens from {key_path:?}");
@@ -465,23 +447,20 @@ fn start_pageserver(
         let jwt_auth = JwtAuth::from_key_path(key_path)?;
         let auth: Arc<SwappableJwtAuth> = Arc::new(SwappableJwtAuth::new(jwt_auth));
 
-        http_auth = match conf.http_auth_type {
+        http_auth = match &conf.http_auth_type {
             AuthType::Trust => None,
             AuthType::NeonJWT => Some(auth.clone()),
         };
-        pg_auth = match conf.pg_auth_type {
-            AuthType::Trust => None,
-            AuthType::NeonJWT => Some(auth.clone()),
-        };
-        grpc_auth = match conf.grpc_auth_type {
+        pg_auth = match &conf.pg_auth_type {
             AuthType::Trust => None,
             AuthType::NeonJWT => Some(auth),
         };
     } else {
         http_auth = None;
         pg_auth = None;
-        grpc_auth = None;
     }
+    info!("Using auth for http API: {:#?}", conf.http_auth_type);
+    info!("Using auth for pg connections: {:#?}", conf.pg_auth_type);
 
     let tls_server_config = if conf.listen_https_addr.is_some() || conf.enable_tls_page_service_api
     {
@@ -562,8 +541,6 @@ fn start_pageserver(
         pageserver::l0_flush::L0FlushGlobalState::new(conf.l0_flush.clone());
 
     // Scan the local 'tenants/' directory and start loading the tenants
-    let (basebackup_prepare_sender, basebackup_prepare_receiver) =
-        tokio::sync::mpsc::unbounded_channel();
     let deletion_queue_client = deletion_queue.new_client();
     let background_purges = mgr::BackgroundPurges::default();
     let tenant_manager = BACKGROUND_RUNTIME.block_on(mgr::init_tenant_mgr(
@@ -574,22 +551,12 @@ fn start_pageserver(
             remote_storage: remote_storage.clone(),
             deletion_queue_client,
             l0_flush_global_state,
-            basebackup_prepare_sender,
         },
         order,
         shutdown_pageserver.clone(),
     ))?;
     let tenant_manager = Arc::new(tenant_manager);
 
-    let basebackup_cache = BasebackupCache::spawn(
-        BACKGROUND_RUNTIME.handle(),
-        conf.basebackup_cache_dir(),
-        conf.basebackup_cache_config.clone(),
-        basebackup_prepare_receiver,
-        Arc::clone(&tenant_manager),
-        shutdown_pageserver.child_token(),
-    );
-
     BACKGROUND_RUNTIME.spawn({
         let shutdown_pageserver = shutdown_pageserver.clone();
         let drive_init = async move {
@@ -796,27 +763,8 @@ fn start_pageserver(
         } else {
             None
         },
-        basebackup_cache.clone(),
     );
 
-    // Spawn a Pageserver gRPC server task. It will spawn separate tasks for
-    // each stream/request.
-    //
-    // TODO: this uses a separate Tokio runtime for the page service. If we want
-    // other gRPC services, they will need their own port and runtime. Is this
-    // necessary?
-    let mut page_service_grpc = None;
-    if let Some(grpc_listener) = grpc_listener {
-        page_service_grpc = Some(page_service::spawn_grpc(
-            conf,
-            tenant_manager.clone(),
-            grpc_auth,
-            otel_guard.as_ref().map(|g| g.dispatch.clone()),
-            grpc_listener,
-            basebackup_cache,
-        )?);
-    }
-
     // All started up! Now just sit and wait for shutdown signal.
     BACKGROUND_RUNTIME.block_on(async move {
         let signal_token = CancellationToken::new();
@@ -835,7 +783,6 @@ fn start_pageserver(
             http_endpoint_listener,
             https_endpoint_listener,
             page_service,
-            page_service_grpc,
             consumption_metrics_tasks,
             disk_usage_eviction_task,
             &tenant_manager,

pageserver/src/config.rs

@@ -58,16 +58,11 @@ pub struct PageServerConf {
     pub listen_http_addr: String,
     /// Example: 127.0.0.1:9899
     pub listen_https_addr: Option<String>,
-    /// If set, expose a gRPC API on this address.
-    /// Example: 127.0.0.1:51051
-    ///
-    /// EXPERIMENTAL: this protocol is unstable and under active development.
-    pub listen_grpc_addr: Option<String>,
 
-    /// Path to a file with certificate's private key for https and gRPC API.
+    /// Path to a file with certificate's private key for https API.
     /// Default: server.key
     pub ssl_key_file: Utf8PathBuf,
-    /// Path to a file with a X509 certificate for https and gRPC API.
+    /// Path to a file with a X509 certificate for https API.
     /// Default: server.crt
     pub ssl_cert_file: Utf8PathBuf,
     /// Period to reload certificate and private key from files.
@@ -105,8 +100,6 @@ pub struct PageServerConf {
     pub http_auth_type: AuthType,
     /// authentication method for libpq connections from compute
     pub pg_auth_type: AuthType,
-    /// authentication method for gRPC connections from compute
-    pub grpc_auth_type: AuthType,
     /// Path to a file or directory containing public key(s) for verifying JWT tokens.
     /// Used for both mgmt and compute auth, if enabled.
     pub auth_validation_public_key_path: Option<Utf8PathBuf>,
@@ -239,8 +232,6 @@ pub struct PageServerConf {
     pub dev_mode: bool,
 
     pub timeline_import_config: pageserver_api::config::TimelineImportConfig,
-
-    pub basebackup_cache_config: Option<pageserver_api::config::BasebackupCacheConfig>,
 }
 
 /// Token for authentication to safekeepers
@@ -270,10 +261,6 @@ impl PageServerConf {
         self.workdir.join("metadata.json")
     }
 
-    pub fn basebackup_cache_dir(&self) -> Utf8PathBuf {
-        self.workdir.join("basebackup_cache")
-    }
-
     pub fn deletion_list_path(&self, sequence: u64) -> Utf8PathBuf {
         // Encode a version in the filename, so that if we ever switch away from JSON we can
         // increment this.
@@ -362,7 +349,6 @@ impl PageServerConf {
             listen_pg_addr,
             listen_http_addr,
             listen_https_addr,
-            listen_grpc_addr,
             ssl_key_file,
             ssl_cert_file,
             ssl_cert_reload_period,
@@ -377,7 +363,6 @@ impl PageServerConf {
             pg_distrib_dir,
             http_auth_type,
             pg_auth_type,
-            grpc_auth_type,
             auth_validation_public_key_path,
             remote_storage,
             broker_endpoint,
@@ -422,7 +407,6 @@ impl PageServerConf {
             enable_tls_page_service_api,
             dev_mode,
             timeline_import_config,
-            basebackup_cache_config,
         } = config_toml;
 
         let mut conf = PageServerConf {
@@ -432,7 +416,6 @@ impl PageServerConf {
             listen_pg_addr,
             listen_http_addr,
             listen_https_addr,
-            listen_grpc_addr,
             ssl_key_file,
             ssl_cert_file,
             ssl_cert_reload_period,
@@ -445,7 +428,6 @@ impl PageServerConf {
             max_file_descriptors,
             http_auth_type,
             pg_auth_type,
-            grpc_auth_type,
             auth_validation_public_key_path,
             remote_storage_config: remote_storage,
             broker_endpoint,
@@ -479,7 +461,6 @@ impl PageServerConf {
             enable_tls_page_service_api,
             dev_mode,
             timeline_import_config,
-            basebackup_cache_config,
 
             // ------------------------------------------------------------
             // fields that require additional validation or custom handling
@@ -542,9 +523,7 @@ impl PageServerConf {
         // custom validation code that covers more than one field in isolation
         // ------------------------------------------------------------
 
-        if [conf.http_auth_type, conf.pg_auth_type, conf.grpc_auth_type]
-            .contains(&AuthType::NeonJWT)
-        {
+        if conf.http_auth_type == AuthType::NeonJWT || conf.pg_auth_type == AuthType::NeonJWT {
             let auth_validation_public_key_path = conf
                 .auth_validation_public_key_path
                 .get_or_insert_with(|| workdir.join("auth_public_key.pem"));

pageserver/src/consumption_metrics/metrics.rs

@@ -18,25 +18,12 @@ use crate::tenant::timeline::logical_size::CurrentLogicalSize;
 // management.
 #[derive(Debug, Clone, Copy, PartialEq, Eq, Hash, serde::Serialize, serde::Deserialize)]
 pub(super) enum Name {
-    /// Timeline last_record_lsn, absolute.
+    /// Timeline last_record_lsn, absolute
     #[serde(rename = "written_size")]
     WrittenSize,
     /// Timeline last_record_lsn, incremental
     #[serde(rename = "written_data_bytes_delta")]
     WrittenSizeDelta,
-    /// Written bytes only on this timeline (not including ancestors):
-    /// written_size - ancestor_lsn
-    ///
-    /// On the root branch, this is equivalent to `written_size`.
-    #[serde(rename = "written_size_since_parent")]
-    WrittenSizeSinceParent,
-    /// PITR history size only on this timeline (not including ancestors):
-    /// last_record_lsn - max(pitr_cutoff, ancestor_lsn).
-    ///
-    /// On the root branch, this is its entire PITR history size. Not emitted if GC hasn't computed
-    /// the PITR cutoff yet. 0 if PITR is disabled.
-    #[serde(rename = "pitr_history_size_since_parent")]
-    PitrHistorySizeSinceParent,
     /// Timeline logical size
     #[serde(rename = "timeline_logical_size")]
     LogicalSize,
@@ -170,32 +157,6 @@ impl MetricsKey {
         .incremental_values()
     }
 
-    /// `written_size` - `ancestor_lsn`.
-    const fn written_size_since_parent(
-        tenant_id: TenantId,
-        timeline_id: TimelineId,
-    ) -> AbsoluteValueFactory {
-        MetricsKey {
-            tenant_id,
-            timeline_id: Some(timeline_id),
-            metric: Name::WrittenSizeSinceParent,
-        }
-        .absolute_values()
-    }
-
-    /// `written_size` - max(`pitr_cutoff`, `ancestor_lsn`).
-    const fn pitr_history_size_since_parent(
-        tenant_id: TenantId,
-        timeline_id: TimelineId,
-    ) -> AbsoluteValueFactory {
-        MetricsKey {
-            tenant_id,
-            timeline_id: Some(timeline_id),
-            metric: Name::PitrHistorySizeSinceParent,
-        }
-        .absolute_values()
-    }
-
     /// Exact [`Timeline::get_current_logical_size`].
     ///
     /// [`Timeline::get_current_logical_size`]: crate::tenant::Timeline::get_current_logical_size
@@ -373,13 +334,7 @@ impl TenantSnapshot {
 struct TimelineSnapshot {
     loaded_at: (Lsn, SystemTime),
     last_record_lsn: Lsn,
-    ancestor_lsn: Lsn,
     current_exact_logical_size: Option<u64>,
-    /// Whether PITR is enabled (pitr_interval > 0).
-    pitr_enabled: bool,
-    /// The PITR cutoff LSN. None if not yet initialized. If PITR is disabled, this is approximately
-    /// Some(last_record_lsn), but may lag behind it since it's computed periodically.
-    pitr_cutoff: Option<Lsn>,
 }
 
 impl TimelineSnapshot {
@@ -399,9 +354,6 @@ impl TimelineSnapshot {
         } else {
             let loaded_at = t.loaded_at;
             let last_record_lsn = t.get_last_record_lsn();
-            let ancestor_lsn = t.get_ancestor_lsn();
-            let pitr_enabled = !t.get_pitr_interval().is_zero();
-            let pitr_cutoff = t.gc_info.read().unwrap().cutoffs.time;
 
             let current_exact_logical_size = {
                 let span = tracing::info_span!("collect_metrics_iteration", tenant_id = %t.tenant_shard_id.tenant_id, timeline_id = %t.timeline_id);
@@ -421,10 +373,7 @@ impl TimelineSnapshot {
             Ok(Some(TimelineSnapshot {
                 loaded_at,
                 last_record_lsn,
-                ancestor_lsn,
                 current_exact_logical_size,
-                pitr_enabled,
-                pitr_cutoff,
             }))
         }
     }
@@ -475,8 +424,6 @@ impl TimelineSnapshot {
 
         let up_to = now;
 
-        let written_size_last = written_size_now.value.max(prev.1); // don't regress
-
         if let Some(delta) = written_size_now.value.checked_sub(prev.1) {
             let key_value = written_size_delta_key.from_until(prev.0, up_to, delta);
             // written_size_delta
@@ -494,27 +441,6 @@ impl TimelineSnapshot {
             });
         }
 
-        // Compute the branch-local written size.
-        let written_size_since_parent_key =
-            MetricsKey::written_size_since_parent(tenant_id, timeline_id);
-        metrics.push(
-            written_size_since_parent_key
-                .at(now, written_size_last.saturating_sub(self.ancestor_lsn.0)),
-        );
-
-        // Compute the branch-local PITR history size. Not emitted if GC hasn't yet computed the
-        // PITR cutoff. 0 if PITR is disabled.
-        let pitr_history_size_since_parent_key =
-            MetricsKey::pitr_history_size_since_parent(tenant_id, timeline_id);
-        if !self.pitr_enabled {
-            metrics.push(pitr_history_size_since_parent_key.at(now, 0));
-        } else if let Some(pitr_cutoff) = self.pitr_cutoff {
-            metrics.push(pitr_history_size_since_parent_key.at(
-                now,
-                written_size_last.saturating_sub(pitr_cutoff.max(self.ancestor_lsn).0),
-            ));
-        }
-
         {
             let factory = MetricsKey::timeline_logical_size(tenant_id, timeline_id);
             let current_or_previous = self

pageserver/src/consumption_metrics/metrics/tests.rs

@@ -12,17 +12,12 @@ fn startup_collected_timeline_metrics_before_advancing() {
     let cache = HashMap::new();
 
     let initdb_lsn = Lsn(0x10000);
-    let pitr_cutoff = Lsn(0x11000);
     let disk_consistent_lsn = Lsn(initdb_lsn.0 * 2);
-    let logical_size = 0x42000;
 
     let snap = TimelineSnapshot {
         loaded_at: (disk_consistent_lsn, SystemTime::now()),
         last_record_lsn: disk_consistent_lsn,
-        ancestor_lsn: Lsn(0),
-        current_exact_logical_size: Some(logical_size),
-        pitr_enabled: true,
-        pitr_cutoff: Some(pitr_cutoff),
+        current_exact_logical_size: Some(0x42000),
     };
 
     let now = DateTime::<Utc>::from(SystemTime::now());
@@ -38,11 +33,7 @@ fn startup_collected_timeline_metrics_before_advancing() {
                 0
             ),
             MetricsKey::written_size(tenant_id, timeline_id).at(now, disk_consistent_lsn.0),
-            MetricsKey::written_size_since_parent(tenant_id, timeline_id)
-                .at(now, disk_consistent_lsn.0),
-            MetricsKey::pitr_history_size_since_parent(tenant_id, timeline_id)
-                .at(now, disk_consistent_lsn.0 - pitr_cutoff.0),
-            MetricsKey::timeline_logical_size(tenant_id, timeline_id).at(now, logical_size)
+            MetricsKey::timeline_logical_size(tenant_id, timeline_id).at(now, 0x42000)
         ]
     );
 }
@@ -58,9 +49,7 @@ fn startup_collected_timeline_metrics_second_round() {
     let before = DateTime::<Utc>::from(before);
 
     let initdb_lsn = Lsn(0x10000);
-    let pitr_cutoff = Lsn(0x11000);
     let disk_consistent_lsn = Lsn(initdb_lsn.0 * 2);
-    let logical_size = 0x42000;
 
     let mut metrics = Vec::new();
     let cache = HashMap::from([MetricsKey::written_size(tenant_id, timeline_id)
@@ -70,10 +59,7 @@ fn startup_collected_timeline_metrics_second_round() {
     let snap = TimelineSnapshot {
         loaded_at: (disk_consistent_lsn, init),
         last_record_lsn: disk_consistent_lsn,
-        ancestor_lsn: Lsn(0),
-        current_exact_logical_size: Some(logical_size),
-        pitr_enabled: true,
-        pitr_cutoff: Some(pitr_cutoff),
+        current_exact_logical_size: Some(0x42000),
     };
 
     snap.to_metrics(tenant_id, timeline_id, now, &mut metrics, &cache);
@@ -83,11 +69,7 @@ fn startup_collected_timeline_metrics_second_round() {
         &[
             MetricsKey::written_size_delta(tenant_id, timeline_id).from_until(before, now, 0),
             MetricsKey::written_size(tenant_id, timeline_id).at(now, disk_consistent_lsn.0),
-            MetricsKey::written_size_since_parent(tenant_id, timeline_id)
-                .at(now, disk_consistent_lsn.0),
-            MetricsKey::pitr_history_size_since_parent(tenant_id, timeline_id)
-                .at(now, disk_consistent_lsn.0 - pitr_cutoff.0),
-            MetricsKey::timeline_logical_size(tenant_id, timeline_id).at(now, logical_size)
+            MetricsKey::timeline_logical_size(tenant_id, timeline_id).at(now, 0x42000)
         ]
     );
 }
@@ -104,9 +86,7 @@ fn startup_collected_timeline_metrics_nth_round_at_same_lsn() {
     let before = DateTime::<Utc>::from(before);
 
     let initdb_lsn = Lsn(0x10000);
-    let pitr_cutoff = Lsn(0x11000);
     let disk_consistent_lsn = Lsn(initdb_lsn.0 * 2);
-    let logical_size = 0x42000;
 
     let mut metrics = Vec::new();
     let cache = HashMap::from([
@@ -123,10 +103,7 @@ fn startup_collected_timeline_metrics_nth_round_at_same_lsn() {
     let snap = TimelineSnapshot {
         loaded_at: (disk_consistent_lsn, init),
         last_record_lsn: disk_consistent_lsn,
-        ancestor_lsn: Lsn(0),
-        current_exact_logical_size: Some(logical_size),
-        pitr_enabled: true,
-        pitr_cutoff: Some(pitr_cutoff),
+        current_exact_logical_size: Some(0x42000),
     };
 
     snap.to_metrics(tenant_id, timeline_id, now, &mut metrics, &cache);
@@ -136,18 +113,16 @@ fn startup_collected_timeline_metrics_nth_round_at_same_lsn() {
         &[
             MetricsKey::written_size_delta(tenant_id, timeline_id).from_until(just_before, now, 0),
             MetricsKey::written_size(tenant_id, timeline_id).at(now, disk_consistent_lsn.0),
-            MetricsKey::written_size_since_parent(tenant_id, timeline_id)
-                .at(now, disk_consistent_lsn.0),
-            MetricsKey::pitr_history_size_since_parent(tenant_id, timeline_id)
-                .at(now, disk_consistent_lsn.0 - pitr_cutoff.0),
-            MetricsKey::timeline_logical_size(tenant_id, timeline_id).at(now, logical_size)
+            MetricsKey::timeline_logical_size(tenant_id, timeline_id).at(now, 0x42000)
         ]
     );
 }
 
-/// Tests that written sizes do not regress across restarts.
 #[test]
 fn post_restart_written_sizes_with_rolled_back_last_record_lsn() {
+    // it can happen that we lose the inmemorylayer but have previously sent metrics and we
+    // should never go backwards
+
     let tenant_id = TenantId::generate();
     let timeline_id = TimelineId::generate();
 
@@ -165,10 +140,7 @@ fn post_restart_written_sizes_with_rolled_back_last_record_lsn() {
     let snap = TimelineSnapshot {
         loaded_at: (Lsn(50), at_restart),
         last_record_lsn: Lsn(50),
-        ancestor_lsn: Lsn(0),
         current_exact_logical_size: None,
-        pitr_enabled: true,
-        pitr_cutoff: Some(Lsn(20)),
     };
 
     let mut cache = HashMap::from([
@@ -197,8 +169,6 @@ fn post_restart_written_sizes_with_rolled_back_last_record_lsn() {
                 0
             ),
             MetricsKey::written_size(tenant_id, timeline_id).at(now, 100),
-            MetricsKey::written_size_since_parent(tenant_id, timeline_id).at(now, 100),
-            MetricsKey::pitr_history_size_since_parent(tenant_id, timeline_id).at(now, 80),
         ]
     );
 
@@ -213,157 +183,6 @@ fn post_restart_written_sizes_with_rolled_back_last_record_lsn() {
         &[
             MetricsKey::written_size_delta(tenant_id, timeline_id).from_until(now, later, 0),
             MetricsKey::written_size(tenant_id, timeline_id).at(later, 100),
-            MetricsKey::written_size_since_parent(tenant_id, timeline_id).at(later, 100),
-            MetricsKey::pitr_history_size_since_parent(tenant_id, timeline_id).at(later, 80),
-        ]
-    );
-}
-
-/// Tests that written sizes do not regress across restarts, even on child branches.
-#[test]
-fn post_restart_written_sizes_with_rolled_back_last_record_lsn_and_ancestor_lsn() {
-    let tenant_id = TenantId::generate();
-    let timeline_id = TimelineId::generate();
-
-    let [later, now, at_restart] = time_backwards();
-
-    // FIXME: tests would be so much easier if we did not need to juggle back and forth
-    // SystemTime and DateTime::<Utc> ... Could do the conversion only at upload time?
-    let now = DateTime::<Utc>::from(now);
-    let later = DateTime::<Utc>::from(later);
-    let before_restart = at_restart - std::time::Duration::from_secs(5 * 60);
-    let way_before = before_restart - std::time::Duration::from_secs(10 * 60);
-    let before_restart = DateTime::<Utc>::from(before_restart);
-    let way_before = DateTime::<Utc>::from(way_before);
-
-    let snap = TimelineSnapshot {
-        loaded_at: (Lsn(50), at_restart),
-        last_record_lsn: Lsn(50),
-        ancestor_lsn: Lsn(40),
-        current_exact_logical_size: None,
-        pitr_enabled: true,
-        pitr_cutoff: Some(Lsn(20)),
-    };
-
-    let mut cache = HashMap::from([
-        MetricsKey::written_size(tenant_id, timeline_id)
-            .at(before_restart, 100)
-            .to_kv_pair(),
-        MetricsKey::written_size_delta(tenant_id, timeline_id)
-            .from_until(
-                way_before,
-                before_restart,
-                // not taken into account, but the timestamps are important
-                999_999_999,
-            )
-            .to_kv_pair(),
-    ]);
-
-    let mut metrics = Vec::new();
-    snap.to_metrics(tenant_id, timeline_id, now, &mut metrics, &cache);
-
-    assert_eq!(
-        metrics,
-        &[
-            MetricsKey::written_size_delta(tenant_id, timeline_id).from_until(
-                before_restart,
-                now,
-                0
-            ),
-            MetricsKey::written_size(tenant_id, timeline_id).at(now, 100),
-            MetricsKey::written_size_since_parent(tenant_id, timeline_id).at(now, 60),
-            MetricsKey::pitr_history_size_since_parent(tenant_id, timeline_id).at(now, 60),
-        ]
-    );
-
-    // now if we cache these metrics, and re-run while "still in recovery"
-    cache.extend(metrics.drain(..).map(|x| x.to_kv_pair()));
-
-    // "still in recovery", because our snapshot did not change
-    snap.to_metrics(tenant_id, timeline_id, later, &mut metrics, &cache);
-
-    assert_eq!(
-        metrics,
-        &[
-            MetricsKey::written_size_delta(tenant_id, timeline_id).from_until(now, later, 0),
-            MetricsKey::written_size(tenant_id, timeline_id).at(later, 100),
-            MetricsKey::written_size_since_parent(tenant_id, timeline_id).at(later, 60),
-            MetricsKey::pitr_history_size_since_parent(tenant_id, timeline_id).at(later, 60),
-        ]
-    );
-}
-
-/// Tests that written sizes do not regress across restarts, even on child branches and
-/// with a PITR cutoff after the branch point.
-#[test]
-fn post_restart_written_sizes_with_rolled_back_last_record_lsn_and_ancestor_lsn_and_pitr_cutoff() {
-    let tenant_id = TenantId::generate();
-    let timeline_id = TimelineId::generate();
-
-    let [later, now, at_restart] = time_backwards();
-
-    // FIXME: tests would be so much easier if we did not need to juggle back and forth
-    // SystemTime and DateTime::<Utc> ... Could do the conversion only at upload time?
-    let now = DateTime::<Utc>::from(now);
-    let later = DateTime::<Utc>::from(later);
-    let before_restart = at_restart - std::time::Duration::from_secs(5 * 60);
-    let way_before = before_restart - std::time::Duration::from_secs(10 * 60);
-    let before_restart = DateTime::<Utc>::from(before_restart);
-    let way_before = DateTime::<Utc>::from(way_before);
-
-    let snap = TimelineSnapshot {
-        loaded_at: (Lsn(50), at_restart),
-        last_record_lsn: Lsn(50),
-        ancestor_lsn: Lsn(30),
-        current_exact_logical_size: None,
-        pitr_enabled: true,
-        pitr_cutoff: Some(Lsn(40)),
-    };
-
-    let mut cache = HashMap::from([
-        MetricsKey::written_size(tenant_id, timeline_id)
-            .at(before_restart, 100)
-            .to_kv_pair(),
-        MetricsKey::written_size_delta(tenant_id, timeline_id)
-            .from_until(
-                way_before,
-                before_restart,
-                // not taken into account, but the timestamps are important
-                999_999_999,
-            )
-            .to_kv_pair(),
-    ]);
-
-    let mut metrics = Vec::new();
-    snap.to_metrics(tenant_id, timeline_id, now, &mut metrics, &cache);
-
-    assert_eq!(
-        metrics,
-        &[
-            MetricsKey::written_size_delta(tenant_id, timeline_id).from_until(
-                before_restart,
-                now,
-                0
-            ),
-            MetricsKey::written_size(tenant_id, timeline_id).at(now, 100),
-            MetricsKey::written_size_since_parent(tenant_id, timeline_id).at(now, 70),
-            MetricsKey::pitr_history_size_since_parent(tenant_id, timeline_id).at(now, 60),
-        ]
-    );
-
-    // now if we cache these metrics, and re-run while "still in recovery"
-    cache.extend(metrics.drain(..).map(|x| x.to_kv_pair()));
-
-    // "still in recovery", because our snapshot did not change
-    snap.to_metrics(tenant_id, timeline_id, later, &mut metrics, &cache);
-
-    assert_eq!(
-        metrics,
-        &[
-            MetricsKey::written_size_delta(tenant_id, timeline_id).from_until(now, later, 0),
-            MetricsKey::written_size(tenant_id, timeline_id).at(later, 100),
-            MetricsKey::written_size_since_parent(tenant_id, timeline_id).at(later, 70),
-            MetricsKey::pitr_history_size_since_parent(tenant_id, timeline_id).at(later, 60),
         ]
     );
 }
@@ -382,10 +201,7 @@ fn post_restart_current_exact_logical_size_uses_cached() {
     let snap = TimelineSnapshot {
         loaded_at: (Lsn(50), at_restart),
         last_record_lsn: Lsn(50),
-        ancestor_lsn: Lsn(0),
         current_exact_logical_size: None,
-        pitr_enabled: true,
-        pitr_cutoff: None,
     };
 
     let cache = HashMap::from([MetricsKey::timeline_logical_size(tenant_id, timeline_id)
@@ -470,101 +286,16 @@ fn time_backwards<const N: usize>() -> [std::time::SystemTime; N] {
     times
 }
 
-/// Tests that disabled PITR history does not yield any history size, even when the PITR cutoff
-/// indicates otherwise.
-#[test]
-fn pitr_disabled_yields_no_history_size() {
-    let tenant_id = TenantId::generate();
-    let timeline_id = TimelineId::generate();
-
-    let mut metrics = Vec::new();
-    let cache = HashMap::new();
-
-    let initdb_lsn = Lsn(0x10000);
-    let pitr_cutoff = Lsn(0x11000);
-    let disk_consistent_lsn = Lsn(initdb_lsn.0 * 2);
-
-    let snap = TimelineSnapshot {
-        loaded_at: (disk_consistent_lsn, SystemTime::now()),
-        last_record_lsn: disk_consistent_lsn,
-        ancestor_lsn: Lsn(0),
-        current_exact_logical_size: None,
-        pitr_enabled: false,
-        pitr_cutoff: Some(pitr_cutoff),
-    };
-
-    let now = DateTime::<Utc>::from(SystemTime::now());
-
-    snap.to_metrics(tenant_id, timeline_id, now, &mut metrics, &cache);
-
-    assert_eq!(
-        metrics,
-        &[
-            MetricsKey::written_size_delta(tenant_id, timeline_id).from_until(
-                snap.loaded_at.1.into(),
-                now,
-                0
-            ),
-            MetricsKey::written_size(tenant_id, timeline_id).at(now, disk_consistent_lsn.0),
-            MetricsKey::written_size_since_parent(tenant_id, timeline_id)
-                .at(now, disk_consistent_lsn.0),
-            MetricsKey::pitr_history_size_since_parent(tenant_id, timeline_id).at(now, 0),
-        ]
-    );
-}
-
-/// Tests that uninitialized PITR cutoff does not emit any history size metric at all.
-#[test]
-fn pitr_uninitialized_does_not_emit_history_size() {
-    let tenant_id = TenantId::generate();
-    let timeline_id = TimelineId::generate();
-
-    let mut metrics = Vec::new();
-    let cache = HashMap::new();
-
-    let initdb_lsn = Lsn(0x10000);
-    let disk_consistent_lsn = Lsn(initdb_lsn.0 * 2);
-
-    let snap = TimelineSnapshot {
-        loaded_at: (disk_consistent_lsn, SystemTime::now()),
-        last_record_lsn: disk_consistent_lsn,
-        ancestor_lsn: Lsn(0),
-        current_exact_logical_size: None,
-        pitr_enabled: true,
-        pitr_cutoff: None,
-    };
-
-    let now = DateTime::<Utc>::from(SystemTime::now());
-
-    snap.to_metrics(tenant_id, timeline_id, now, &mut metrics, &cache);
-
-    assert_eq!(
-        metrics,
-        &[
-            MetricsKey::written_size_delta(tenant_id, timeline_id).from_until(
-                snap.loaded_at.1.into(),
-                now,
-                0
-            ),
-            MetricsKey::written_size(tenant_id, timeline_id).at(now, disk_consistent_lsn.0),
-            MetricsKey::written_size_since_parent(tenant_id, timeline_id)
-                .at(now, disk_consistent_lsn.0),
-        ]
-    );
-}
-
 pub(crate) const fn metric_examples_old(
     tenant_id: TenantId,
     timeline_id: TimelineId,
     now: DateTime<Utc>,
     before: DateTime<Utc>,
-) -> [RawMetric; 7] {
+) -> [RawMetric; 5] {
     [
         MetricsKey::written_size(tenant_id, timeline_id).at_old_format(now, 0),
         MetricsKey::written_size_delta(tenant_id, timeline_id)
             .from_until_old_format(before, now, 0),
-        MetricsKey::written_size_since_parent(tenant_id, timeline_id).at_old_format(now, 0),
-        MetricsKey::pitr_history_size_since_parent(tenant_id, timeline_id).at_old_format(now, 0),
         MetricsKey::timeline_logical_size(tenant_id, timeline_id).at_old_format(now, 0),
         MetricsKey::remote_storage_size(tenant_id).at_old_format(now, 0),
         MetricsKey::synthetic_size(tenant_id).at_old_format(now, 1),
@@ -576,12 +307,10 @@ pub(crate) const fn metric_examples(
     timeline_id: TimelineId,
     now: DateTime<Utc>,
     before: DateTime<Utc>,
-) -> [NewRawMetric; 7] {
+) -> [NewRawMetric; 5] {
     [
         MetricsKey::written_size(tenant_id, timeline_id).at(now, 0),
         MetricsKey::written_size_delta(tenant_id, timeline_id).from_until(before, now, 0),
-        MetricsKey::written_size_since_parent(tenant_id, timeline_id).at(now, 0),
-        MetricsKey::pitr_history_size_since_parent(tenant_id, timeline_id).at(now, 0),
         MetricsKey::timeline_logical_size(tenant_id, timeline_id).at(now, 0),
         MetricsKey::remote_storage_size(tenant_id).at(now, 0),
         MetricsKey::synthetic_size(tenant_id).at(now, 1),

pageserver/src/consumption_metrics/upload.rs

@@ -513,14 +513,6 @@ mod tests {
                 line!(),
                 r#"{"type":"incremental","start_time":"2023-09-14T00:00:00.123456789Z","stop_time":"2023-09-15T00:00:00.123456789Z","metric":"written_data_bytes_delta","idempotency_key":"2023-09-15 00:00:00.123456789 UTC-1-0000","value":0,"tenant_id":"00000000000000000000000000000000","timeline_id":"ffffffffffffffffffffffffffffffff"}"#,
             ),
-            (
-                line!(),
-                r#"{"type":"absolute","time":"2023-09-15T00:00:00.123456789Z","metric":"written_size_since_parent","idempotency_key":"2023-09-15 00:00:00.123456789 UTC-1-0000","value":0,"tenant_id":"00000000000000000000000000000000","timeline_id":"ffffffffffffffffffffffffffffffff"}"#,
-            ),
-            (
-                line!(),
-                r#"{"type":"absolute","time":"2023-09-15T00:00:00.123456789Z","metric":"pitr_history_size_since_parent","idempotency_key":"2023-09-15 00:00:00.123456789 UTC-1-0000","value":0,"tenant_id":"00000000000000000000000000000000","timeline_id":"ffffffffffffffffffffffffffffffff"}"#,
-            ),
             (
                 line!(),
                 r#"{"type":"absolute","time":"2023-09-15T00:00:00.123456789Z","metric":"timeline_logical_size","idempotency_key":"2023-09-15 00:00:00.123456789 UTC-1-0000","value":0,"tenant_id":"00000000000000000000000000000000","timeline_id":"ffffffffffffffffffffffffffffffff"}"#,
@@ -568,7 +560,7 @@ mod tests {
         assert_eq!(upgraded_samples, new_samples);
     }
 
-    fn metric_samples_old() -> [RawMetric; 7] {
+    fn metric_samples_old() -> [RawMetric; 5] {
         let tenant_id = TenantId::from_array([0; 16]);
         let timeline_id = TimelineId::from_array([0xff; 16]);
 
@@ -580,7 +572,7 @@ mod tests {
         super::super::metrics::metric_examples_old(tenant_id, timeline_id, now, before)
     }
 
-    fn metric_samples() -> [NewRawMetric; 7] {
+    fn metric_samples() -> [NewRawMetric; 5] {
         let tenant_id = TenantId::from_array([0; 16]);
         let timeline_id = TimelineId::from_array([0xff; 16]);
 

pageserver/src/lib.rs

@@ -3,7 +3,6 @@
 
 mod auth;
 pub mod basebackup;
-pub mod basebackup_cache;
 pub mod config;
 pub mod consumption_metrics;
 pub mod context;
@@ -84,7 +83,6 @@ pub async fn shutdown_pageserver(
     http_listener: HttpEndpointListener,
     https_listener: Option<HttpsEndpointListener>,
     page_service: page_service::Listener,
-    grpc_task: Option<CancellableTask>,
     consumption_metrics_worker: ConsumptionMetricsTasks,
     disk_usage_eviction_task: Option<DiskUsageEvictionTask>,
     tenant_manager: &TenantManager,
@@ -178,16 +176,6 @@ pub async fn shutdown_pageserver(
     )
     .await;
 
-    // Shut down the gRPC server task, including request handlers.
-    if let Some(grpc_task) = grpc_task {
-        timed(
-            grpc_task.shutdown(),
-            "shutdown gRPC PageRequestHandler",
-            Duration::from_secs(3),
-        )
-        .await;
-    }
-
     // Shut down all the tenants. This flushes everything to disk and kills
     // the checkpoint and GC tasks.
     timed(

pageserver/src/metrics.rs

@@ -4359,42 +4359,6 @@ pub(crate) fn set_tokio_runtime_setup(setup: &str, num_threads: NonZeroUsize) {
         .set(u64::try_from(num_threads.get()).unwrap());
 }
 
-pub(crate) static BASEBACKUP_CACHE_READ: Lazy<IntCounterVec> = Lazy::new(|| {
-    register_int_counter_vec!(
-        "pageserver_basebackup_cache_read_total",
-        "Number of read accesses to the basebackup cache grouped by hit/miss/error",
-        &["result"]
-    )
-    .expect("failed to define a metric")
-});
-
-pub(crate) static BASEBACKUP_CACHE_PREPARE: Lazy<IntCounterVec> = Lazy::new(|| {
-    register_int_counter_vec!(
-        "pageserver_basebackup_cache_prepare_total",
-        "Number of prepare requests processed by the basebackup cache grouped by ok/skip/error",
-        &["result"]
-    )
-    .expect("failed to define a metric")
-});
-
-pub(crate) static BASEBACKUP_CACHE_ENTRIES: Lazy<IntGauge> = Lazy::new(|| {
-    register_int_gauge!(
-        "pageserver_basebackup_cache_entries_total",
-        "Number of entries in the basebackup cache"
-    )
-    .expect("failed to define a metric")
-});
-
-// FIXME: Support basebackup cache size metrics.
-#[allow(dead_code)]
-pub(crate) static BASEBACKUP_CACHE_SIZE: Lazy<IntGauge> = Lazy::new(|| {
-    register_int_gauge!(
-        "pageserver_basebackup_cache_size_bytes",
-        "Total size of all basebackup cache entries on disk in bytes"
-    )
-    .expect("failed to define a metric")
-});
-
 static PAGESERVER_CONFIG_IGNORED_ITEMS: Lazy<UIntGaugeVec> = Lazy::new(|| {
     register_uint_gauge_vec!(
         "pageserver_config_ignored_items",

pageserver/src/page_service.rs

@@ -4,16 +4,16 @@
 use std::borrow::Cow;
 use std::num::NonZeroUsize;
 use std::os::fd::AsRawFd;
-use std::pin::Pin;
 use std::str::FromStr;
 use std::sync::Arc;
 use std::time::{Duration, Instant, SystemTime};
 use std::{io, str};
 
+use crate::PERF_TRACE_TARGET;
 use anyhow::{Context, bail};
 use async_compression::tokio::write::GzipEncoder;
 use bytes::Buf;
-use futures::{FutureExt, Stream};
+use futures::FutureExt;
 use itertools::Itertools;
 use jsonwebtoken::TokenData;
 use once_cell::sync::OnceCell;
@@ -31,7 +31,6 @@ use pageserver_api::models::{
 };
 use pageserver_api::reltag::SlruKind;
 use pageserver_api::shard::TenantShardId;
-use pageserver_page_api::proto;
 use postgres_backend::{
     AuthType, PostgresBackend, PostgresBackendReader, QueryError, is_expected_io_error,
 };
@@ -54,8 +53,7 @@ use utils::sync::gate::{Gate, GateGuard};
 use utils::sync::spsc_fold;
 
 use crate::auth::check_permission;
-use crate::basebackup::{self, BasebackupError};
-use crate::basebackup_cache::BasebackupCache;
+use crate::basebackup::BasebackupError;
 use crate::config::PageServerConf;
 use crate::context::{
     DownloadBehavior, PerfInstrumentFutureExt, RequestContext, RequestContextBuilder,
@@ -76,7 +74,7 @@ use crate::tenant::mgr::{
 use crate::tenant::storage_layer::IoConcurrency;
 use crate::tenant::timeline::{self, WaitLsnError};
 use crate::tenant::{GetTimelineError, PageReconstructError, Timeline};
-use crate::{CancellableTask, PERF_TRACE_TARGET, timed_after_cancellation};
+use crate::{basebackup, timed_after_cancellation};
 
 /// How long we may wait for a [`crate::tenant::mgr::TenantSlot::InProgress`]` and/or a [`crate::tenant::TenantShard`] which
 /// is not yet in state [`TenantState::Active`].
@@ -87,26 +85,6 @@ const ACTIVE_TENANT_TIMEOUT: Duration = Duration::from_millis(30000);
 /// Threshold at which to log slow GetPage requests.
 const LOG_SLOW_GETPAGE_THRESHOLD: Duration = Duration::from_secs(30);
 
-/// The idle time before sending TCP keepalive probes for gRPC connections. The
-/// interval and timeout between each probe is configured via sysctl. This
-/// allows detecting dead connections sooner.
-const GRPC_TCP_KEEPALIVE_TIME: Duration = Duration::from_secs(60);
-
-/// Whether to enable TCP nodelay for gRPC connections. This disables Nagle's
-/// algorithm, which can cause latency spikes for small messages.
-const GRPC_TCP_NODELAY: bool = true;
-
-/// The interval between HTTP2 keepalive pings. This allows shutting down server
-/// tasks when clients are unresponsive.
-const GRPC_HTTP2_KEEPALIVE_INTERVAL: Duration = Duration::from_secs(30);
-
-/// The timeout for HTTP2 keepalive pings. Should be <= GRPC_KEEPALIVE_INTERVAL.
-const GRPC_HTTP2_KEEPALIVE_TIMEOUT: Duration = Duration::from_secs(20);
-
-/// Number of concurrent gRPC streams per TCP connection. We expect something
-/// like 8 GetPage streams per connections, plus any unary requests.
-const GRPC_MAX_CONCURRENT_STREAMS: u32 = 256;
-
 ///////////////////////////////////////////////////////////////////////////////
 
 pub struct Listener {
@@ -129,7 +107,6 @@ pub fn spawn(
     perf_trace_dispatch: Option<Dispatch>,
     tcp_listener: tokio::net::TcpListener,
     tls_config: Option<Arc<rustls::ServerConfig>>,
-    basebackup_cache: Arc<BasebackupCache>,
 ) -> Listener {
     let cancel = CancellationToken::new();
     let libpq_ctx = RequestContext::todo_child(
@@ -151,7 +128,6 @@ pub fn spawn(
             conf.pg_auth_type,
             tls_config,
             conf.page_service_pipelining.clone(),
-            basebackup_cache,
             libpq_ctx,
             cancel.clone(),
         )
@@ -161,83 +137,6 @@ pub fn spawn(
     Listener { cancel, task }
 }
 
-/// Spawns a gRPC server for the page service.
-///
-/// TODO: this doesn't support TLS. We need TLS reloading via ReloadingCertificateResolver, so we
-/// need to reimplement the TCP+TLS accept loop ourselves.
-pub fn spawn_grpc(
-    conf: &'static PageServerConf,
-    tenant_manager: Arc<TenantManager>,
-    auth: Option<Arc<SwappableJwtAuth>>,
-    perf_trace_dispatch: Option<Dispatch>,
-    listener: std::net::TcpListener,
-    basebackup_cache: Arc<BasebackupCache>,
-) -> anyhow::Result<CancellableTask> {
-    let cancel = CancellationToken::new();
-    let ctx = RequestContextBuilder::new(TaskKind::PageRequestHandler)
-        .download_behavior(DownloadBehavior::Download)
-        .perf_span_dispatch(perf_trace_dispatch)
-        .detached_child();
-    let gate = Gate::default();
-
-    // Set up the TCP socket. We take a preconfigured TcpListener to bind the
-    // port early during startup.
-    let incoming = {
-        let _runtime = COMPUTE_REQUEST_RUNTIME.enter(); // required by TcpListener::from_std
-        listener.set_nonblocking(true)?;
-        tonic::transport::server::TcpIncoming::from(tokio::net::TcpListener::from_std(listener)?)
-            .with_nodelay(Some(GRPC_TCP_NODELAY))
-            .with_keepalive(Some(GRPC_TCP_KEEPALIVE_TIME))
-    };
-
-    // Set up the gRPC server.
-    //
-    // TODO: consider tuning window sizes.
-    // TODO: wire up tracing.
-    let mut server = tonic::transport::Server::builder()
-        .http2_keepalive_interval(Some(GRPC_HTTP2_KEEPALIVE_INTERVAL))
-        .http2_keepalive_timeout(Some(GRPC_HTTP2_KEEPALIVE_TIMEOUT))
-        .max_concurrent_streams(Some(GRPC_MAX_CONCURRENT_STREAMS));
-
-    // Main page service.
-    let page_service = proto::PageServiceServer::new(PageServerHandler::new(
-        tenant_manager,
-        auth,
-        PageServicePipeliningConfig::Serial, // TODO: unused with gRPC
-        conf.get_vectored_concurrent_io,
-        ConnectionPerfSpanFields::default(),
-        basebackup_cache,
-        ctx,
-        cancel.clone(),
-        gate.enter().expect("just created"),
-    ));
-    let server = server.add_service(page_service);
-
-    // Reflection service for use with e.g. grpcurl.
-    let reflection_service = tonic_reflection::server::Builder::configure()
-        .register_encoded_file_descriptor_set(proto::FILE_DESCRIPTOR_SET)
-        .build_v1()?;
-    let server = server.add_service(reflection_service);
-
-    // Spawn server task.
-    let task_cancel = cancel.clone();
-    let task = COMPUTE_REQUEST_RUNTIME.spawn(task_mgr::exit_on_panic_or_error(
-        "grpc listener",
-        async move {
-            let result = server
-                .serve_with_incoming_shutdown(incoming, task_cancel.cancelled())
-                .await;
-            if result.is_ok() {
-                // TODO: revisit shutdown logic once page service is implemented.
-                gate.close().await;
-            }
-            result
-        },
-    ));
-
-    Ok(CancellableTask { task, cancel })
-}
-
 impl Listener {
     pub async fn stop_accepting(self) -> Connections {
         self.cancel.cancel();
@@ -287,7 +186,6 @@ pub async fn libpq_listener_main(
     auth_type: AuthType,
     tls_config: Option<Arc<rustls::ServerConfig>>,
     pipelining_config: PageServicePipeliningConfig,
-    basebackup_cache: Arc<BasebackupCache>,
     listener_ctx: RequestContext,
     listener_cancel: CancellationToken,
 ) -> Connections {
@@ -331,7 +229,6 @@ pub async fn libpq_listener_main(
                     auth_type,
                     tls_config.clone(),
                     pipelining_config.clone(),
-                    Arc::clone(&basebackup_cache),
                     connection_ctx,
                     connections_cancel.child_token(),
                     gate_guard,
@@ -357,7 +254,7 @@ type ConnectionHandlerResult = anyhow::Result<()>;
 
 /// Perf root spans start at the per-request level, after shard routing.
 /// This struct carries connection-level information to the root perf span definition.
-#[derive(Clone, Default)]
+#[derive(Clone)]
 struct ConnectionPerfSpanFields {
     peer_addr: String,
     application_name: Option<String>,
@@ -374,7 +271,6 @@ async fn page_service_conn_main(
     auth_type: AuthType,
     tls_config: Option<Arc<rustls::ServerConfig>>,
     pipelining_config: PageServicePipeliningConfig,
-    basebackup_cache: Arc<BasebackupCache>,
     connection_ctx: RequestContext,
     cancel: CancellationToken,
     gate_guard: GateGuard,
@@ -440,7 +336,6 @@ async fn page_service_conn_main(
         pipelining_config,
         conf.get_vectored_concurrent_io,
         perf_span_fields,
-        basebackup_cache,
         connection_ctx,
         cancel.clone(),
         gate_guard,
@@ -475,11 +370,6 @@ async fn page_service_conn_main(
     }
 }
 
-/// Page service connection handler.
-///
-/// TODO: for gRPC, this will be shared by all requests from all connections.
-/// Decompose it into global state and per-connection/request state, and make
-/// libpq-specific options (e.g. pipelining) separate.
 struct PageServerHandler {
     auth: Option<Arc<SwappableJwtAuth>>,
     claims: Option<Claims>,
@@ -500,8 +390,6 @@ struct PageServerHandler {
     pipelining_config: PageServicePipeliningConfig,
     get_vectored_concurrent_io: GetVectoredConcurrentIo,
 
-    basebackup_cache: Arc<BasebackupCache>,
-
     gate_guard: GateGuard,
 }
 
@@ -961,7 +849,6 @@ impl PageServerHandler {
         pipelining_config: PageServicePipeliningConfig,
         get_vectored_concurrent_io: GetVectoredConcurrentIo,
         perf_span_fields: ConnectionPerfSpanFields,
-        basebackup_cache: Arc<BasebackupCache>,
         connection_ctx: RequestContext,
         cancel: CancellationToken,
         gate_guard: GateGuard,
@@ -975,7 +862,6 @@ impl PageServerHandler {
             cancel,
             pipelining_config,
             get_vectored_concurrent_io,
-            basebackup_cache,
             gate_guard,
         }
     }
@@ -2607,8 +2493,6 @@ impl PageServerHandler {
             .map_err(QueryError::Disconnected)?;
         self.flush_cancellable(pgb, &self.cancel).await?;
 
-        let mut from_cache = false;
-
         // Send a tarball of the latest layer on the timeline. Compress if not
         // fullbackup. TODO Compress in that case too (tests need to be updated)
         if full_backup {
@@ -2626,33 +2510,7 @@ impl PageServerHandler {
             .map_err(map_basebackup_error)?;
         } else {
             let mut writer = BufWriter::new(pgb.copyout_writer());
-
-            let cached = {
-                // Basebackup is cached only for this combination of parameters.
-                if timeline.is_basebackup_cache_enabled()
-                    && gzip
-                    && lsn.is_some()
-                    && prev_lsn.is_none()
-                {
-                    self.basebackup_cache
-                        .get(tenant_id, timeline_id, lsn.unwrap())
-                        .await
-                } else {
-                    None
-                }
-            };
-
-            if let Some(mut cached) = cached {
-                from_cache = true;
-                tokio::io::copy(&mut cached, &mut writer)
-                    .await
-                    .map_err(|e| {
-                        map_basebackup_error(BasebackupError::Client(
-                            e,
-                            "handle_basebackup_request,cached,copy",
-                        ))
-                    })?;
-            } else if gzip {
+            if gzip {
                 let mut encoder = GzipEncoder::with_quality(
                     &mut writer,
                     // NOTE using fast compression because it's on the critical path
@@ -2711,7 +2569,6 @@ impl PageServerHandler {
         info!(
             lsn_await_millis = lsn_awaited_after.as_millis(),
             basebackup_millis = basebackup_after.as_millis(),
-            %from_cache,
             "basebackup complete"
         );
 
@@ -3220,60 +3077,6 @@ where
     }
 }
 
-/// Implements the page service over gRPC.
-///
-/// TODO: not yet implemented, all methods return unimplemented.
-#[tonic::async_trait]
-impl proto::PageService for PageServerHandler {
-    type GetBaseBackupStream = Pin<
-        Box<dyn Stream<Item = Result<proto::GetBaseBackupResponseChunk, tonic::Status>> + Send>,
-    >;
-    type GetPagesStream =
-        Pin<Box<dyn Stream<Item = Result<proto::GetPageResponse, tonic::Status>> + Send>>;
-
-    async fn check_rel_exists(
-        &self,
-        _: tonic::Request<proto::CheckRelExistsRequest>,
-    ) -> Result<tonic::Response<proto::CheckRelExistsResponse>, tonic::Status> {
-        Err(tonic::Status::unimplemented("not implemented"))
-    }
-
-    async fn get_base_backup(
-        &self,
-        _: tonic::Request<proto::GetBaseBackupRequest>,
-    ) -> Result<tonic::Response<Self::GetBaseBackupStream>, tonic::Status> {
-        Err(tonic::Status::unimplemented("not implemented"))
-    }
-
-    async fn get_db_size(
-        &self,
-        _: tonic::Request<proto::GetDbSizeRequest>,
-    ) -> Result<tonic::Response<proto::GetDbSizeResponse>, tonic::Status> {
-        Err(tonic::Status::unimplemented("not implemented"))
-    }
-
-    async fn get_pages(
-        &self,
-        _: tonic::Request<tonic::Streaming<proto::GetPageRequest>>,
-    ) -> Result<tonic::Response<Self::GetPagesStream>, tonic::Status> {
-        Err(tonic::Status::unimplemented("not implemented"))
-    }
-
-    async fn get_rel_size(
-        &self,
-        _: tonic::Request<proto::GetRelSizeRequest>,
-    ) -> Result<tonic::Response<proto::GetRelSizeResponse>, tonic::Status> {
-        Err(tonic::Status::unimplemented("not implemented"))
-    }
-
-    async fn get_slru_segment(
-        &self,
-        _: tonic::Request<proto::GetSlruSegmentRequest>,
-    ) -> Result<tonic::Response<proto::GetSlruSegmentResponse>, tonic::Status> {
-        Err(tonic::Status::unimplemented("not implemented"))
-    }
-}
-
 impl From<GetActiveTenantError> for QueryError {
     fn from(e: GetActiveTenantError) -> Self {
         match e {

pageserver/src/task_mgr.rs

@@ -276,10 +276,9 @@ pub enum TaskKind {
     // HTTP endpoint listener.
     HttpEndpointListener,
 
-    /// Task that handles a single page service connection. A PageRequestHandler
-    /// task starts detached from any particular tenant or timeline, but it can
-    /// be associated with one later, after receiving a command from the client.
-    /// Also used for the gRPC page service API, including the main server task.
+    // Task that handles a single connection. A PageRequestHandler task
+    // starts detached from any particular tenant or timeline, but it can be
+    // associated with one later, after receiving a command from the client.
     PageRequestHandler,
 
     /// Manages the WAL receiver connection for one timeline.
@@ -381,10 +380,6 @@ pub enum TaskKind {
     DetachAncestor,
 
     ImportPgdata,
-
-    /// Background task of [`crate::basebackup_cache::BasebackupCache`].
-    /// Prepares basebackups and clears outdated entries.
-    BasebackupCache,
 }
 
 #[derive(Default)]

pageserver/src/tenant.rs

@@ -78,7 +78,6 @@ use self::timeline::uninit::{TimelineCreateGuard, TimelineExclusionError, Uninit
 use self::timeline::{
     EvictionTaskTenantState, GcCutoffs, TimelineDeleteProgress, TimelineResources, WaitLsnError,
 };
-use crate::basebackup_cache::BasebackupPrepareSender;
 use crate::config::PageServerConf;
 use crate::context;
 use crate::context::RequestContextBuilder;
@@ -158,7 +157,6 @@ pub struct TenantSharedResources {
     pub remote_storage: GenericRemoteStorage,
     pub deletion_queue_client: DeletionQueueClient,
     pub l0_flush_global_state: L0FlushGlobalState,
-    pub basebackup_prepare_sender: BasebackupPrepareSender,
 }
 
 /// A [`TenantShard`] is really an _attached_ tenant.  The configuration
@@ -319,15 +317,12 @@ pub struct TenantShard {
     gc_cs: tokio::sync::Mutex<()>,
     walredo_mgr: Option<Arc<WalRedoManager>>,
 
-    /// Provides access to timeline data sitting in the remote storage.
+    // provides access to timeline data sitting in the remote storage
     pub(crate) remote_storage: GenericRemoteStorage,
 
-    /// Access to global deletion queue for when this tenant wants to schedule a deletion.
+    // Access to global deletion queue for when this tenant wants to schedule a deletion
     deletion_queue_client: DeletionQueueClient,
 
-    /// A channel to send async requests to prepare a basebackup for the basebackup cache.
-    basebackup_prepare_sender: BasebackupPrepareSender,
-
     /// Cached logical sizes updated updated on each [`TenantShard::gather_size_inputs`].
     cached_logical_sizes: tokio::sync::Mutex<HashMap<(TimelineId, Lsn), u64>>,
     cached_synthetic_tenant_size: Arc<AtomicU64>,
@@ -1291,7 +1286,6 @@ impl TenantShard {
             remote_storage,
             deletion_queue_client,
             l0_flush_global_state,
-            basebackup_prepare_sender,
         } = resources;
 
         let attach_mode = attached_conf.location.attach_mode;
@@ -1307,7 +1301,6 @@ impl TenantShard {
             remote_storage.clone(),
             deletion_queue_client,
             l0_flush_global_state,
-            basebackup_prepare_sender,
         ));
 
         // The attach task will carry a GateGuard, so that shutdown() reliably waits for it to drop out if
@@ -4246,7 +4239,6 @@ impl TenantShard {
         remote_storage: GenericRemoteStorage,
         deletion_queue_client: DeletionQueueClient,
         l0_flush_global_state: L0FlushGlobalState,
-        basebackup_prepare_sender: BasebackupPrepareSender,
     ) -> TenantShard {
         assert!(!attached_conf.location.generation.is_none());
 
@@ -4350,7 +4342,6 @@ impl TenantShard {
             ongoing_timeline_detach: std::sync::Mutex::default(),
             gc_block: Default::default(),
             l0_flush_global_state,
-            basebackup_prepare_sender,
         }
     }
 
@@ -5270,7 +5261,6 @@ impl TenantShard {
             pagestream_throttle_metrics: self.pagestream_throttle_metrics.clone(),
             l0_compaction_trigger: self.l0_compaction_trigger.clone(),
             l0_flush_global_state: self.l0_flush_global_state.clone(),
-            basebackup_prepare_sender: self.basebackup_prepare_sender.clone(),
         }
     }
 
@@ -5853,8 +5843,6 @@ pub(crate) mod harness {
         ) -> anyhow::Result<Arc<TenantShard>> {
             let walredo_mgr = Arc::new(WalRedoManager::from(TestRedoManager));
 
-            let (basebackup_requst_sender, _) = tokio::sync::mpsc::unbounded_channel();
-
             let tenant = Arc::new(TenantShard::new(
                 TenantState::Attaching,
                 self.conf,
@@ -5872,7 +5860,6 @@ pub(crate) mod harness {
                 self.deletion_queue.new_client(),
                 // TODO: ideally we should run all unit tests with both configs
                 L0FlushGlobalState::new(L0FlushConfig::default()),
-                basebackup_requst_sender,
             ));
 
             let preload = tenant

pageserver/src/tenant/timeline.rs

@@ -24,6 +24,8 @@ use std::sync::atomic::{AtomicBool, AtomicU64, Ordering as AtomicOrdering};
 use std::sync::{Arc, Mutex, OnceLock, RwLock, Weak};
 use std::time::{Duration, Instant, SystemTime};
 
+use crate::PERF_TRACE_TARGET;
+use crate::walredo::RedoAttemptType;
 use anyhow::{Context, Result, anyhow, bail, ensure};
 use arc_swap::{ArcSwap, ArcSwapOption};
 use bytes::Bytes;
@@ -92,12 +94,10 @@ use super::storage_layer::{LayerFringe, LayerVisibilityHint, ReadableLayer};
 use super::tasks::log_compaction_error;
 use super::upload_queue::NotInitialized;
 use super::{
-    AttachedTenantConf, BasebackupPrepareSender, GcError, HeatMapTimeline, MaybeOffloaded,
+    AttachedTenantConf, GcError, HeatMapTimeline, MaybeOffloaded,
     debug_assert_current_span_has_tenant_and_timeline_id,
 };
-use crate::PERF_TRACE_TARGET;
 use crate::aux_file::AuxFileSizeEstimator;
-use crate::basebackup_cache::BasebackupPrepareRequest;
 use crate::config::PageServerConf;
 use crate::context::{
     DownloadBehavior, PerfInstrumentFutureExt, RequestContext, RequestContextBuilder,
@@ -131,7 +131,6 @@ use crate::tenant::tasks::BackgroundLoopKind;
 use crate::tenant::timeline::logical_size::CurrentLogicalSize;
 use crate::virtual_file::{MaybeFatalIo, VirtualFile};
 use crate::walingest::WalLagCooldown;
-use crate::walredo::RedoAttemptType;
 use crate::{ZERO_PAGE, task_mgr, walredo};
 
 #[derive(Debug, PartialEq, Eq, Clone, Copy)]
@@ -197,7 +196,6 @@ pub struct TimelineResources {
     pub pagestream_throttle_metrics: Arc<crate::metrics::tenant_throttling::Pagestream>,
     pub l0_compaction_trigger: Arc<Notify>,
     pub l0_flush_global_state: l0_flush::L0FlushGlobalState,
-    pub basebackup_prepare_sender: BasebackupPrepareSender,
 }
 
 pub struct Timeline {
@@ -441,9 +439,6 @@ pub struct Timeline {
     pub(crate) rel_size_v2_status: ArcSwapOption<RelSizeMigration>,
 
     wait_lsn_log_slow: tokio::sync::Semaphore,
-
-    /// A channel to send async requests to prepare a basebackup for the basebackup cache.
-    basebackup_prepare_sender: BasebackupPrepareSender,
 }
 
 pub(crate) enum PreviousHeatmap {
@@ -1033,7 +1028,6 @@ pub(crate) enum WaitLsnWaiter<'a> {
     Tenant,
     PageService,
     HttpEndpoint,
-    BaseBackupCache,
 }
 
 /// Argument to [`Timeline::shutdown`].
@@ -1560,8 +1554,7 @@ impl Timeline {
                         }
                         WaitLsnWaiter::Tenant
                         | WaitLsnWaiter::PageService
-                        | WaitLsnWaiter::HttpEndpoint
-                        | WaitLsnWaiter::BaseBackupCache => unreachable!(
+                        | WaitLsnWaiter::HttpEndpoint => unreachable!(
                             "tenant or page_service context are not expected to have task kind {:?}",
                             ctx.task_kind()
                         ),
@@ -2466,41 +2459,6 @@ impl Timeline {
             false
         }
     }
-
-    pub(crate) fn is_basebackup_cache_enabled(&self) -> bool {
-        let tenant_conf = self.tenant_conf.load();
-        tenant_conf
-            .tenant_conf
-            .basebackup_cache_enabled
-            .unwrap_or(self.conf.default_tenant_conf.basebackup_cache_enabled)
-    }
-
-    /// Prepare basebackup for the given LSN and store it in the basebackup cache.
-    /// The method is asynchronous and returns immediately.
-    /// The actual basebackup preparation is performed in the background
-    /// by the basebackup cache on a best-effort basis.
-    pub(crate) fn prepare_basebackup(&self, lsn: Lsn) {
-        if !self.is_basebackup_cache_enabled() {
-            return;
-        }
-        if !self.tenant_shard_id.is_shard_zero() {
-            // In theory we should never get here, but just in case check it.
-            // Preparing basebackup doesn't make sense for shards other than shard zero.
-            return;
-        }
-
-        let res = self
-            .basebackup_prepare_sender
-            .send(BasebackupPrepareRequest {
-                tenant_shard_id: self.tenant_shard_id,
-                timeline_id: self.timeline_id,
-                lsn,
-            });
-        if let Err(e) = res {
-            // May happen during shutdown, it's not critical.
-            info!("Failed to send shutdown checkpoint: {e:#}");
-        }
-    }
 }
 
 /// Number of times we will compute partition within a checkpoint distance.
@@ -2578,13 +2536,6 @@ impl Timeline {
             .unwrap_or(self.conf.default_tenant_conf.checkpoint_timeout)
     }
 
-    pub(crate) fn get_pitr_interval(&self) -> Duration {
-        let tenant_conf = &self.tenant_conf.load().tenant_conf;
-        tenant_conf
-            .pitr_interval
-            .unwrap_or(self.conf.default_tenant_conf.pitr_interval)
-    }
-
     fn get_compaction_period(&self) -> Duration {
         let tenant_conf = self.tenant_conf.load().tenant_conf.clone();
         tenant_conf
@@ -3070,8 +3021,6 @@ impl Timeline {
                 rel_size_v2_status: ArcSwapOption::from_pointee(rel_size_v2_status),
 
                 wait_lsn_log_slow: tokio::sync::Semaphore::new(1),
-
-                basebackup_prepare_sender: resources.basebackup_prepare_sender,
             };
 
             result.repartition_threshold =

pageserver/src/walingest.rs

@@ -1316,10 +1316,6 @@ impl WalIngest {
             }
         });
 
-        if info == pg_constants::XLOG_CHECKPOINT_SHUTDOWN {
-            modification.tline.prepare_basebackup(lsn);
-        }
-
         Ok(())
     }
 

pgxn/neon/communicator.c

@@ -717,7 +717,7 @@ prefetch_read(PrefetchRequest *slot)
 	Assert(slot->status == PRFS_REQUESTED);
 	Assert(slot->response == NULL);
 	Assert(slot->my_ring_index == MyPState->ring_receive);
-	Assert(readpage_reentrant_guard || AmPrewarmWorker);
+	Assert(readpage_reentrant_guard);
 
 	if (slot->status != PRFS_REQUESTED ||
 		slot->response != NULL ||
@@ -800,7 +800,7 @@ communicator_prefetch_receive(BufferTag tag)
 	PrfHashEntry *entry;
 	PrefetchRequest hashkey;
 
-	Assert(readpage_reentrant_guard || AmPrewarmWorker); /* do not pump prefetch state in prewarm worker */
+	Assert(readpage_reentrant_guard);
 	hashkey.buftag = tag;
 	entry = prfh_lookup(MyPState->prf_hash, &hashkey);
 	if (entry != NULL && prefetch_wait_for(entry->slot->my_ring_index))
@@ -2450,7 +2450,6 @@ void
 communicator_reconfigure_timeout_if_needed(void)
 {
 	bool	needs_set = MyPState->ring_receive != MyPState->ring_unused &&
-						!AmPrewarmWorker && /* do not pump prefetch state in prewarm worker */
 						readahead_getpage_pull_timeout_ms > 0;
 
 	if (needs_set != timeout_set)

pgxn/neon/file_cache.c

@@ -201,8 +201,6 @@ static shmem_request_hook_type prev_shmem_request_hook;
 bool lfc_store_prefetch_result;
 bool lfc_prewarm_update_ws_estimation;
 
-bool AmPrewarmWorker;
-
 #define LFC_ENABLED() (lfc_ctl->limit != 0)
 
 /*
@@ -847,8 +845,6 @@ lfc_prewarm_main(Datum main_arg)
 	PrewarmWorkerState* ws;
 	uint32 worker_id = DatumGetInt32(main_arg);
 
-	AmPrewarmWorker = true;
-
 	pqsignal(SIGTERM, die);
 	BackgroundWorkerUnblockSignals();
 

pgxn/neon/neon.h

@@ -23,8 +23,6 @@ extern int	wal_acceptor_connection_timeout;
 extern int	readahead_getpage_pull_timeout_ms;
 extern bool	disable_wal_prev_lsn_checks;
 
-extern bool AmPrewarmWorker;
-
 #if PG_MAJORVERSION_NUM >= 17
 extern uint32		WAIT_EVENT_NEON_LFC_MAINTENANCE;
 extern uint32		WAIT_EVENT_NEON_LFC_READ;

proxy/src/auth/backend/mod.rs

@@ -80,22 +80,10 @@ impl std::fmt::Display for Backend<'_, ()> {
                     .field(&endpoint.url())
                     .finish(),
                 #[cfg(any(test, feature = "testing"))]
-                ControlPlaneClient::PostgresMock(endpoint) => {
-                    let url = endpoint.url();
-                    match url::Url::parse(url) {
-                        Ok(mut url) => {
-                            let _ = url.set_password(Some("_redacted_"));
-                            let url = url.as_str();
-                            fmt.debug_tuple("ControlPlane::PostgresMock")
-                                .field(&url)
-                                .finish()
-                        }
-                        Err(_) => fmt
-                            .debug_tuple("ControlPlane::PostgresMock")
-                            .field(&url)
-                            .finish(),
-                    }
-                }
+                ControlPlaneClient::PostgresMock(endpoint) => fmt
+                    .debug_tuple("ControlPlane::PostgresMock")
+                    .field(&endpoint.url())
+                    .finish(),
                 #[cfg(test)]
                 ControlPlaneClient::Test(_) => fmt.debug_tuple("ControlPlane::Test").finish(),
             },

proxy/src/binary/proxy.rs

@@ -1,13 +1,9 @@
-#[cfg(any(test, feature = "testing"))]
-use std::env;
 use std::net::SocketAddr;
 use std::path::PathBuf;
 use std::pin::pin;
 use std::sync::Arc;
 use std::time::Duration;
 
-#[cfg(any(test, feature = "testing"))]
-use anyhow::Context;
 use anyhow::{bail, ensure};
 use arc_swap::ArcSwapOption;
 use futures::future::Either;
@@ -39,8 +35,6 @@ use crate::scram::threadpool::ThreadPool;
 use crate::serverless::GlobalConnPoolOptions;
 use crate::serverless::cancel_set::CancelSet;
 use crate::tls::client_config::compute_client_config_with_root_certs;
-#[cfg(any(test, feature = "testing"))]
-use crate::url::ApiUrl;
 use crate::{auth, control_plane, http, serverless, usage_metrics};
 
 project_git_version!(GIT_VERSION);
@@ -783,13 +777,7 @@ fn build_auth_backend(
 
         #[cfg(any(test, feature = "testing"))]
         AuthBackendType::Postgres => {
-            let mut url: ApiUrl = args.auth_endpoint.parse()?;
-            if url.password().is_none() {
-                let password = env::var("PGPASSWORD")
-                    .with_context(|| "auth-endpoint does not contain a password and environment variable `PGPASSWORD` is not set")?;
-                url.set_password(Some(&password))
-                    .expect("Failed to set password");
-            }
+            let url = args.auth_endpoint.parse()?;
             let api = control_plane::client::mock::MockControlPlane::new(
                 url,
                 !args.is_private_access_proxy,

proxy/src/serverless/json.rs

@@ -70,34 +70,6 @@ pub(crate) enum JsonConversionError {
     ParseJsonError(#[from] serde_json::Error),
     #[error("unbalanced array")]
     UnbalancedArray,
-    #[error("unbalanced quoted string")]
-    UnbalancedString,
-}
-
-enum OutputMode {
-    Array(Vec<Value>),
-    Object(Map<String, Value>),
-}
-
-impl OutputMode {
-    fn key(&mut self, key: &str) -> &mut Value {
-        match self {
-            OutputMode::Array(values) => push_entry(values, Value::Null),
-            OutputMode::Object(map) => map.entry(key.to_string()).or_insert(Value::Null),
-        }
-    }
-
-    fn finish(self) -> Value {
-        match self {
-            OutputMode::Array(values) => Value::Array(values),
-            OutputMode::Object(map) => Value::Object(map),
-        }
-    }
-}
-
-fn push_entry<T>(arr: &mut Vec<T>, t: T) -> &mut T {
-    arr.push(t);
-    arr.last_mut().expect("a value was just inserted")
 }
 
 //
@@ -105,276 +77,182 @@ fn push_entry<T>(arr: &mut Vec<T>, t: T) -> &mut T {
 //
 pub(crate) fn pg_text_row_to_json(
     row: &Row,
+    columns: &[Type],
     raw_output: bool,
     array_mode: bool,
 ) -> Result<Value, JsonConversionError> {
-    let mut entries = if array_mode {
-        OutputMode::Array(Vec::with_capacity(row.columns().len()))
+    let iter = row
+        .columns()
+        .iter()
+        .zip(columns)
+        .enumerate()
+        .map(|(i, (column, typ))| {
+            let name = column.name();
+            let pg_value = row.as_text(i).map_err(JsonConversionError::AsTextError)?;
+            let json_value = if raw_output {
+                match pg_value {
+                    Some(v) => Value::String(v.to_string()),
+                    None => Value::Null,
+                }
+            } else {
+                pg_text_to_json(pg_value, typ)?
+            };
+            Ok((name.to_string(), json_value))
+        });
+
+    if array_mode {
+        // drop keys and aggregate into array
+        let arr = iter
+            .map(|r| r.map(|(_key, val)| val))
+            .collect::<Result<Vec<Value>, JsonConversionError>>()?;
+        Ok(Value::Array(arr))
     } else {
-        OutputMode::Object(Map::with_capacity(row.columns().len()))
-    };
-
-    for (i, column) in row.columns().iter().enumerate() {
-        let pg_value = row.as_text(i).map_err(JsonConversionError::AsTextError)?;
-
-        let value = entries.key(column.name());
-
-        match pg_value {
-            Some(v) if raw_output => *value = Value::String(v.to_string()),
-            Some(v) => pg_text_to_json(value, v, column.type_())?,
-            None => *value = Value::Null,
-        }
+        let obj = iter.collect::<Result<Map<String, Value>, JsonConversionError>>()?;
+        Ok(Value::Object(obj))
     }
-
-    Ok(entries.finish())
 }
 
 //
 // Convert postgres text-encoded value to JSON value
 //
-fn pg_text_to_json(
-    output: &mut Value,
-    val: &str,
-    pg_type: &Type,
-) -> Result<(), JsonConversionError> {
-    if let Kind::Array(elem_type) = pg_type.kind() {
-        // todo: we should fetch this from postgres.
-        let delimiter = ',';
-
-        let mut array = vec![];
-        pg_array_parse(&mut array, val, elem_type, delimiter)?;
-        *output = Value::Array(array);
-        return Ok(());
-    }
-
-    match *pg_type {
-        Type::BOOL => *output = Value::Bool(val == "t"),
-        Type::INT2 | Type::INT4 => {
-            let val = val.parse::<i32>()?;
-            *output = Value::Number(serde_json::Number::from(val));
+fn pg_text_to_json(pg_value: Option<&str>, pg_type: &Type) -> Result<Value, JsonConversionError> {
+    if let Some(val) = pg_value {
+        if let Kind::Array(elem_type) = pg_type.kind() {
+            return pg_array_parse(val, elem_type);
         }
-        Type::FLOAT4 | Type::FLOAT8 => {
-            let fval = val.parse::<f64>()?;
-            let num = serde_json::Number::from_f64(fval);
-            if let Some(num) = num {
-                *output = 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
-                *output = Value::String(val.to_string());
+
+        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, JsonConversionError> {
+    pg_array_parse_inner(pg_array, elem_type, false).map(|(v, _)| v)
+}
+
+fn pg_array_parse_inner(
+    pg_array: &str,
+    elem_type: &Type,
+    nested: bool,
+) -> Result<(Value, usize), JsonConversionError> {
+    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;
             }
         }
-        Type::JSON | Type::JSONB => *output = serde_json::from_str(val)?,
-        _ => *output = Value::String(val.to_string()),
     }
 
-    Ok(())
-}
+    fn push_checked(
+        entry: &mut String,
+        entries: &mut Vec<Value>,
+        elem_type: &Type,
+    ) -> Result<(), JsonConversionError> {
+        if !entry.is_empty() {
+            // While in usual postgres response we get nulls as None and everything else
+            // as Some(&str), in arrays we get NULL as unquoted 'NULL' string (while
+            // string with value 'NULL' will be represented by '"NULL"'). So catch NULLs
+            // here while we have quotation info and convert them to None.
+            if entry == "NULL" {
+                entries.push(pg_text_to_json(None, elem_type)?);
+            } else {
+                entries.push(pg_text_to_json(Some(entry), elem_type)?);
+            }
+            entry.clear();
+        }
 
-/// 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.
-///
-/// <https://www.postgresql.org/docs/current/arrays.html#ARRAYS-IO>
-///
-/// The external text representation of an array value consists of items that are interpreted
-/// according to the I/O conversion rules for the array's element type, plus decoration that
-/// indicates the array structure. The decoration consists of curly braces (`{` and `}`) around
-/// the array value plus delimiter characters between adjacent items. The delimiter character
-/// is usually a comma (,) but can be something else: it is determined by the typdelim setting
-/// for the array's element type. Among the standard data types provided in the PostgreSQL
-/// distribution, all use a comma, except for type box, which uses a semicolon (;).
-///
-/// In a multidimensional array, each dimension (row, plane, cube, etc.)
-/// gets its own level of curly braces, and delimiters must be written between adjacent
-/// curly-braced entities of the same level.
-fn pg_array_parse(
-    elements: &mut Vec<Value>,
-    mut pg_array: &str,
-    elem: &Type,
-    delim: char,
-) -> Result<(), JsonConversionError> {
-    // skip bounds decoration, eg:
-    // `[1:1][-2:-1][3:5]={{{1,2,3},{4,5,6}}}`
-    // technically these are significant, but we have no way to represent them in json.
-    if let Some('[') = pg_array.chars().next() {
-        let Some((_bounds, array)) = pg_array.split_once('=') else {
-            return Err(JsonConversionError::UnbalancedArray);
-        };
-        pg_array = array;
+        Ok(())
     }
 
-    // whitespace might preceed a `{`.
-    let pg_array = pg_array.trim_start();
+    while let Some((mut i, mut c)) = pg_array_chr.next() {
+        let mut escaped = false;
 
-    let rest = pg_array_parse_inner(elements, pg_array, elem, delim)?;
-    if !rest.is_empty() {
-        return Err(JsonConversionError::UnbalancedArray);
-    }
+        if c == '\\' {
+            escaped = true;
+            let Some(x) = pg_array_chr.next() else {
+                return Err(JsonConversionError::UnbalancedArray);
+            };
+            (i, c) = x;
+        }
 
-    Ok(())
-}
-
-/// reads a single array from the `pg_array` string and pushes each values to `elements`.
-/// returns the rest of the `pg_array` string that was not read.
-fn pg_array_parse_inner<'a>(
-    elements: &mut Vec<Value>,
-    mut pg_array: &'a str,
-    elem: &Type,
-    delim: char,
-) -> Result<&'a str, JsonConversionError> {
-    // array should have a `{` prefix.
-    pg_array = pg_array
-        .strip_prefix('{')
-        .ok_or(JsonConversionError::UnbalancedArray)?;
-
-    let mut q = String::new();
-
-    loop {
-        let value = push_entry(elements, Value::Null);
-        pg_array = pg_array_parse_item(value, &mut q, pg_array, elem, delim)?;
-
-        // check for separator.
-        if let Some(next) = pg_array.strip_prefix(delim) {
-            // next item.
-            pg_array = next;
-        } else {
-            break;
+        match c {
+            '{' if !quote => {
+                level += 1;
+                if level > 1 {
+                    let (res, off) = pg_array_parse_inner(&pg_array[i..], elem_type, true)?;
+                    entries.push(res);
+                    for _ in 0..off - 1 {
+                        pg_array_chr.next();
+                    }
+                }
+            }
+            '}' if !quote => {
+                level -= 1;
+                if level == 0 {
+                    push_checked(&mut entry, &mut entries, elem_type)?;
+                    if nested {
+                        return Ok((Value::Array(entries), i));
+                    }
+                }
+            }
+            '"' if !escaped => {
+                if quote {
+                    // end of quoted string, so push it manually without any checks
+                    // for emptiness or nulls
+                    entries.push(pg_text_to_json(Some(&entry), elem_type)?);
+                    entry.clear();
+                }
+                quote = !quote;
+            }
+            ',' if !quote => {
+                push_checked(&mut entry, &mut entries, elem_type)?;
+            }
+            _ => {
+                entry.push(c);
+            }
         }
     }
 
-    let Some(next) = pg_array.strip_prefix('}') else {
-        // missing `}` terminator.
+    if level != 0 {
         return Err(JsonConversionError::UnbalancedArray);
-    };
-
-    // whitespace might follow a `}`.
-    Ok(next.trim_start())
-}
-
-/// reads a single item from the `pg_array` string.
-/// returns the rest of the `pg_array` string that was not read.
-///
-/// `quoted` is a scratch allocation that has no defined output.
-fn pg_array_parse_item<'a>(
-    output: &mut Value,
-    quoted: &mut String,
-    mut pg_array: &'a str,
-    elem: &Type,
-    delim: char,
-) -> Result<&'a str, JsonConversionError> {
-    // We are trying to parse an array item.
-    // This could be a new array, if this is a multi-dimentional array.
-    // This could be a quoted string representing `elem`.
-    // This could be an unquoted string representing `elem`.
-
-    // whitespace might preceed an item.
-    pg_array = pg_array.trim_start();
-
-    if pg_array.strip_prefix('{').is_some() {
-        // nested array.
-        let mut nested = vec![];
-        pg_array = pg_array_parse_inner(&mut nested, pg_array, elem, delim)?;
-        *output = Value::Array(nested);
-        return Ok(pg_array);
     }
 
-    if let Some(mut pg_array) = pg_array.strip_prefix('"') {
-        pg_array = pg_array_parse_quoted(quoted, pg_array)?;
-
-        // we have unquoted an item string:
-        pg_text_to_json(output, quoted, elem)?;
-
-        quoted.clear();
-
-        return Ok(pg_array);
-    }
-
-    // we need to parse an item. read until we find a delimiter or `}`.
-    let index = pg_array
-        .find([delim, '}'])
-        .ok_or(JsonConversionError::UnbalancedArray)?;
-
-    let item;
-    (item, pg_array) = pg_array.split_at(index);
-
-    // item might have trailing whitespace that we need to ignore.
-    let item = item.trim_end();
-
-    // we might have an item string:
-    // check for null
-    if item == "NULL" {
-        *output = Value::Null;
-    } else {
-        pg_text_to_json(output, item, elem)?;
-    }
-
-    Ok(pg_array)
-}
-
-/// reads a single quoted item from the `pg_array` string.
-///
-/// Returns the rest of the `pg_array` string that was not read.
-/// The output is written into `quoted`.
-///
-/// The pg_array string must have a `"` terminator, but the `"` initial value
-/// must have already been removed from the input. The terminator is removed.
-fn pg_array_parse_quoted<'a>(
-    quoted: &mut String,
-    mut pg_array: &'a str,
-) -> Result<&'a str, JsonConversionError> {
-    // The array output routine will put double quotes around element values if they are empty strings,
-    // contain curly braces, delimiter characters, double quotes, backslashes, or white space,
-    // or match the word `NULL`. Double quotes and backslashes embedded in element values will be backslash-escaped.
-    // For numeric data types it is safe to assume that double quotes will never appear,
-    // but for textual data types one should be prepared to cope with either the presence or absence of quotes.
-
-    // We write to quoted in chunks terminated by an escape character.
-    // Eg if we have the input `foo\"bar"`, then we write `foo`, then `"`, then finally `bar`.
-
-    loop {
-        // we need to parse an chunk. read until we find a '\\' or `"`.
-        let i = pg_array
-            .find(['\\', '"'])
-            .ok_or(JsonConversionError::UnbalancedString)?;
-
-        let chunk: &str;
-        (chunk, pg_array) = pg_array
-            .split_at_checked(i)
-            .expect("i is guaranteed to be in-bounds of pg_array");
-
-        // push the chunk.
-        quoted.push_str(chunk);
-
-        // consume the chunk_end character.
-        let chunk_end: char;
-        (chunk_end, pg_array) =
-            split_first_char(pg_array).expect("pg_array should start with either '\\\\' or '\"'");
-
-        // finished.
-        if chunk_end == '"' {
-            // whitespace might follow the '"'.
-            pg_array = pg_array.trim_start();
-
-            break Ok(pg_array);
-        }
-
-        // consume the escaped character.
-        let escaped: char;
-        (escaped, pg_array) =
-            split_first_char(pg_array).ok_or(JsonConversionError::UnbalancedString)?;
-
-        quoted.push(escaped);
-    }
-}
-
-fn split_first_char(s: &str) -> Option<(char, &str)> {
-    let mut chars = s.chars();
-    let c = chars.next()?;
-    Some((c, chars.as_str()))
+    Ok((Value::Array(entries), 0))
 }
 
 #[cfg(test)]
@@ -438,33 +316,37 @@ mod tests {
         );
     }
 
-    fn pg_text_to_json(val: &str, pg_type: &Type) -> Value {
-        let mut v = Value::Null;
-        super::pg_text_to_json(&mut v, val, pg_type).unwrap();
-        v
-    }
-
-    fn pg_array_parse(pg_array: &str, pg_type: &Type) -> Value {
-        let mut array = vec![];
-        super::pg_array_parse(&mut array, pg_array, pg_type, ',').unwrap();
-        Value::Array(array)
-    }
-
     #[test]
     fn test_atomic_types_parse() {
-        assert_eq!(pg_text_to_json("foo", &Type::TEXT), json!("foo"));
-        assert_eq!(pg_text_to_json("42", &Type::INT4), json!(42));
-        assert_eq!(pg_text_to_json("42", &Type::INT2), json!(42));
-        assert_eq!(pg_text_to_json("42", &Type::INT8), json!("42"));
-        assert_eq!(pg_text_to_json("42.42", &Type::FLOAT8), json!(42.42));
-        assert_eq!(pg_text_to_json("42.42", &Type::FLOAT4), json!(42.42));
-        assert_eq!(pg_text_to_json("NaN", &Type::FLOAT4), json!("NaN"));
         assert_eq!(
-            pg_text_to_json("Infinity", &Type::FLOAT4),
+            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("-Infinity", &Type::FLOAT4),
+            pg_text_to_json(Some("-Infinity"), &Type::FLOAT4).unwrap(),
             json!("-Infinity")
         );
 
@@ -473,9 +355,10 @@ mod tests {
                 .unwrap();
         assert_eq!(
             pg_text_to_json(
-                r#"{"s":"str","n":42,"f":4.2,"a":[null,3,"a"]}"#,
+                Some(r#"{"s":"str","n":42,"f":4.2,"a":[null,3,"a"]}"#),
                 &Type::JSONB
-            ),
+            )
+            .unwrap(),
             json
         );
     }
@@ -483,7 +366,7 @@ mod tests {
     #[test]
     fn test_pg_array_parse_text() {
         fn pt(pg_arr: &str) -> Value {
-            pg_array_parse(pg_arr, &Type::TEXT)
+            pg_array_parse(pg_arr, &Type::TEXT).unwrap()
         }
         assert_eq!(
             pt(r#"{"aa\"\\\,a",cha,"bbbb"}"#),
@@ -506,7 +389,7 @@ mod tests {
     #[test]
     fn test_pg_array_parse_bool() {
         fn pb(pg_arr: &str) -> Value {
-            pg_array_parse(pg_arr, &Type::BOOL)
+            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]]));
@@ -523,7 +406,7 @@ mod tests {
     #[test]
     fn test_pg_array_parse_numbers() {
         fn pn(pg_arr: &str, ty: &Type) -> Value {
-            pg_array_parse(pg_arr, ty)
+            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]));
@@ -551,7 +434,7 @@ mod tests {
     #[test]
     fn test_pg_array_with_decoration() {
         fn p(pg_arr: &str) -> Value {
-            pg_array_parse(pg_arr, &Type::INT2)
+            pg_array_parse(pg_arr, &Type::INT2).unwrap()
         }
         assert_eq!(
             p(r#"[1:1][-2:-1][3:5]={{{1,2,3},{4,5,6}}}"#),
@@ -562,7 +445,7 @@ mod tests {
     #[test]
     fn test_pg_array_parse_json() {
         fn pt(pg_arr: &str) -> Value {
-            pg_array_parse(pg_arr, &Type::JSONB)
+            pg_array_parse(pg_arr, &Type::JSONB).unwrap()
         }
         assert_eq!(pt(r#"{"{}"}"#), json!([{}]));
         assert_eq!(

proxy/src/serverless/sql_over_http.rs

@@ -14,9 +14,7 @@ use hyper::http::{HeaderName, HeaderValue};
 use hyper::{HeaderMap, Request, Response, StatusCode, header};
 use indexmap::IndexMap;
 use postgres_client::error::{DbError, ErrorPosition, SqlState};
-use postgres_client::{
-    GenericClient, IsolationLevel, NoTls, ReadyForQueryStatus, RowStream, Transaction,
-};
+use postgres_client::{GenericClient, IsolationLevel, NoTls, ReadyForQueryStatus, Transaction};
 use pq_proto::StartupMessageParamsBuilder;
 use serde::Serialize;
 use serde_json::Value;
@@ -1094,38 +1092,22 @@ async fn query_to_json<T: GenericClient>(
     let query_start = Instant::now();
 
     let query_params = data.params;
-    let mut row_stream = client
-        .query_raw_txt(&data.query, query_params)
-        .await
-        .map_err(SqlOverHttpError::Postgres)?;
+    let mut row_stream = std::pin::pin!(
+        client
+            .query_raw_txt(&data.query, query_params)
+            .await
+            .map_err(SqlOverHttpError::Postgres)?
+    );
     let query_acknowledged = Instant::now();
 
-    let columns_len = row_stream.statement.columns().len();
-    let mut fields = Vec::with_capacity(columns_len);
-
-    for c in row_stream.statement.columns() {
-        fields.push(json!({
-            "name": c.name().to_owned(),
-            "dataTypeID": c.type_().oid(),
-            "tableID": c.table_oid(),
-            "columnID": c.column_id(),
-            "dataTypeSize": c.type_size(),
-            "dataTypeModifier": c.type_modifier(),
-            "format": "text",
-        }));
-    }
-
-    let raw_output = parsed_headers.raw_output;
-    let array_mode = data.array_mode.unwrap_or(parsed_headers.default_array_mode);
-
     // 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.
-    let mut rows = Vec::new();
+    let mut rows: Vec<postgres_client::Row> = Vec::new();
     while let Some(row) = row_stream.next().await {
         let row = row.map_err(SqlOverHttpError::Postgres)?;
         *current_size += row.body_len();
-
+        rows.push(row);
         // we don't have a streaming response support yet so this is to prevent OOM
         // from a malicious query (eg a cross join)
         if *current_size > config.max_response_size_bytes {
@@ -1133,26 +1115,13 @@ async fn query_to_json<T: GenericClient>(
                 config.max_response_size_bytes,
             ));
         }
-
-        let row = pg_text_row_to_json(&row, raw_output, array_mode)?;
-        rows.push(row);
-
-        // assumption: parsing pg text and converting to json takes CPU time.
-        // let's assume it is slightly expensive, so we should consume some cooperative budget.
-        // Especially considering that `RowStream::next` might be pulling from a batch
-        // of rows and never hit the tokio mpsc for a long time (although unlikely).
-        tokio::task::consume_budget().await;
     }
 
     let query_resp_end = Instant::now();
-    let RowStream {
-        command_tag,
-        status: ready,
-        ..
-    } = row_stream;
+    let ready = row_stream.ready_status();
 
     // grab the command tag and number of rows affected
-    let command_tag = command_tag.unwrap_or_default();
+    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" {
@@ -1173,6 +1142,38 @@ async fn query_to_json<T: GenericClient>(
         "finished executing query"
     );
 
+    let columns_len = row_stream.columns().len();
+    let mut fields = Vec::with_capacity(columns_len);
+    let mut columns = Vec::with_capacity(columns_len);
+
+    for c in row_stream.columns() {
+        fields.push(json!({
+            "name": c.name().to_owned(),
+            "dataTypeID": c.type_().oid(),
+            "tableID": c.table_oid(),
+            "columnID": c.column_id(),
+            "dataTypeSize": c.type_size(),
+            "dataTypeModifier": c.type_modifier(),
+            "format": "text",
+        }));
+
+        match client.get_type(c.type_oid()).await {
+            Ok(t) => columns.push(t),
+            Err(err) => {
+                tracing::warn!(?err, "unable to query type information");
+                return Err(SqlOverHttpError::InternalPostgres(err));
+            }
+        }
+    }
+
+    let array_mode = data.array_mode.unwrap_or(parsed_headers.default_array_mode);
+
+    // convert rows to JSON
+    let rows = rows
+        .iter()
+        .map(|row| pg_text_row_to_json(row, &columns, parsed_headers.raw_output, array_mode))
+        .collect::<Result<Vec<_>, _>>()?;
+
     // Resulting JSON format is based on the format of node-postgres result.
     let results = json!({
         "command": command_tag_name.to_string(),

proxy/src/url.rs

@@ -43,12 +43,6 @@ impl std::ops::Deref for ApiUrl {
     }
 }
 
-impl std::ops::DerefMut for ApiUrl {
-    fn deref_mut(&mut self) -> &mut Self::Target {
-        &mut self.0
-    }
-}
-
 impl std::fmt::Display for ApiUrl {
     fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
         self.0.fmt(f)

scripts/benchmark_durations.py

@@ -32,6 +32,12 @@ BENCHMARKS_DURATION_QUERY = """
 # the total duration varies from 8 to 40 minutes.
 # We use some pre-collected durations as a fallback to have a better distribution.
 FALLBACK_DURATION = {
+    "test_runner/performance/pageserver/pagebench/test_pageserver_max_throughput_getpage_at_latest_lsn.py::test_pageserver_max_throughput_getpage_at_latest_lsn[1-13-30]": 400.15,
+    "test_runner/performance/pageserver/pagebench/test_pageserver_max_throughput_getpage_at_latest_lsn.py::test_pageserver_max_throughput_getpage_at_latest_lsn[1-6-30]": 372.521,
+    "test_runner/performance/pageserver/pagebench/test_pageserver_max_throughput_getpage_at_latest_lsn.py::test_pageserver_max_throughput_getpage_at_latest_lsn[10-13-30]": 420.017,
+    "test_runner/performance/pageserver/pagebench/test_pageserver_max_throughput_getpage_at_latest_lsn.py::test_pageserver_max_throughput_getpage_at_latest_lsn[10-6-30]": 373.769,
+    "test_runner/performance/pageserver/pagebench/test_pageserver_max_throughput_getpage_at_latest_lsn.py::test_pageserver_max_throughput_getpage_at_latest_lsn[100-13-30]": 678.742,
+    "test_runner/performance/pageserver/pagebench/test_pageserver_max_throughput_getpage_at_latest_lsn.py::test_pageserver_max_throughput_getpage_at_latest_lsn[100-6-30]": 512.135,
     "test_runner/performance/test_branch_creation.py::test_branch_creation_heavy_write[20]": 58.036,
     "test_runner/performance/test_branch_creation.py::test_branch_creation_many_relations": 22.104,
     "test_runner/performance/test_branch_creation.py::test_branch_creation_many[1024]": 126.073,

storage_broker/src/bin/storage_broker.rs

@@ -17,14 +17,12 @@ use std::pin::Pin;
 use std::sync::Arc;
 use std::time::Duration;
 
-use bytes::Bytes;
 use camino::Utf8PathBuf;
 use clap::{Parser, command};
 use futures::future::OptionFuture;
 use futures_core::Stream;
 use futures_util::StreamExt;
-use http_body_util::combinators::BoxBody;
-use http_body_util::{Empty, Full};
+use http_body_util::Full;
 use http_utils::tls_certs::ReloadingCertificateResolver;
 use hyper::body::Incoming;
 use hyper::header::CONTENT_TYPE;
@@ -48,6 +46,7 @@ use tokio::net::TcpListener;
 use tokio::sync::broadcast;
 use tokio::sync::broadcast::error::RecvError;
 use tokio::time;
+use tonic::body::{self, BoxBody, empty_body};
 use tonic::codegen::Service;
 use tonic::{Code, Request, Response, Status};
 use tracing::*;
@@ -635,7 +634,7 @@ impl BrokerService for Broker {
 // We serve only metrics and healthcheck through http1.
 async fn http1_handler(
     req: hyper::Request<Incoming>,
-) -> Result<hyper::Response<BoxBody<Bytes, Infallible>>, Infallible> {
+) -> Result<hyper::Response<BoxBody>, Infallible> {
     let resp = match (req.method(), req.uri().path()) {
         (&Method::GET, "/metrics") => {
             let mut buffer = vec![];
@@ -646,16 +645,16 @@ async fn http1_handler(
             hyper::Response::builder()
                 .status(StatusCode::OK)
                 .header(CONTENT_TYPE, encoder.format_type())
-                .body(BoxBody::new(Full::new(Bytes::from(buffer))))
+                .body(body::boxed(Full::new(bytes::Bytes::from(buffer))))
                 .unwrap()
         }
         (&Method::GET, "/status") => hyper::Response::builder()
             .status(StatusCode::OK)
-            .body(BoxBody::new(Empty::new()))
+            .body(empty_body())
             .unwrap(),
         _ => hyper::Response::builder()
             .status(StatusCode::NOT_FOUND)
-            .body(BoxBody::new(Empty::new()))
+            .body(empty_body())
             .unwrap(),
     };
     Ok(resp)

test_runner/fixtures/neon_fixtures.py

@@ -682,7 +682,7 @@ class NeonEnvBuilder:
                 log.info(
                     f"Copying pageserver tenants directory {tenants_from_dir} to {tenants_to_dir}"
                 )
-                subprocess.run(["cp", "-a", tenants_from_dir, tenants_to_dir], check=True)
+                shutil.copytree(tenants_from_dir, tenants_to_dir)
             else:
                 log.info(
                     f"Creating overlayfs mount of pageserver tenants directory {tenants_from_dir} to {tenants_to_dir}"
@@ -698,9 +698,8 @@ class NeonEnvBuilder:
         shutil.rmtree(self.repo_dir / "local_fs_remote_storage", ignore_errors=True)
         if self.test_overlay_dir is None:
             log.info("Copying local_fs_remote_storage directory from snapshot")
-            subprocess.run(
-                ["cp", "-a", f"{repo_dir / 'local_fs_remote_storage'}", f"{self.repo_dir}"],
-                check=True,
+            shutil.copytree(
+                repo_dir / "local_fs_remote_storage", self.repo_dir / "local_fs_remote_storage"
             )
         else:
             log.info("Creating overlayfs mount of local_fs_remote_storage directory from snapshot")
@@ -1224,7 +1223,6 @@ class NeonEnv:
         # Create config for pageserver
         http_auth_type = "NeonJWT" if config.auth_enabled else "Trust"
         pg_auth_type = "NeonJWT" if config.auth_enabled else "Trust"
-        grpc_auth_type = "NeonJWT" if config.auth_enabled else "Trust"
         for ps_id in range(
             self.BASE_PAGESERVER_ID, self.BASE_PAGESERVER_ID + config.num_pageservers
         ):
@@ -1251,7 +1249,6 @@ class NeonEnv:
                 else None,
                 "pg_auth_type": pg_auth_type,
                 "http_auth_type": http_auth_type,
-                "grpc_auth_type": grpc_auth_type,
                 "availability_zone": availability_zone,
                 # Disable pageserver disk syncs in tests: when running tests concurrently, this avoids
                 # the pageserver taking a long time to start up due to syncfs flushing other tests' data

test_runner/performance/pageserver/pagebench/test_pageserver_max_throughput_getpage_at_latest_lsn.py

@@ -14,7 +14,7 @@ from fixtures.neon_fixtures import (
     PgBin,
     wait_for_last_flush_lsn,
 )
-from fixtures.utils import get_scale_for_db, humantime_to_ms
+from fixtures.utils import get_scale_for_db, humantime_to_ms, skip_on_ci
 
 from performance.pageserver.util import setup_pageserver_with_tenants
 
@@ -36,6 +36,9 @@ if TYPE_CHECKING:
 @pytest.mark.parametrize("pgbench_scale", [get_scale_for_db(200)])
 @pytest.mark.parametrize("n_tenants", [500])
 @pytest.mark.timeout(10000)
+@skip_on_ci(
+    "This test needs lot of resources and should run on dedicated HW, not in github action runners as part of CI"
+)
 def test_pageserver_characterize_throughput_with_n_tenants(
     neon_env_builder: NeonEnvBuilder,
     zenbenchmark: NeonBenchmarker,
@@ -60,6 +63,9 @@ def test_pageserver_characterize_throughput_with_n_tenants(
 @pytest.mark.parametrize("n_clients", [1, 64])
 @pytest.mark.parametrize("n_tenants", [1])
 @pytest.mark.timeout(2400)
+@skip_on_ci(
+    "This test needs lot of resources and should run on dedicated HW, not in github action runners as part of CI"
+)
 def test_pageserver_characterize_latencies_with_1_client_and_throughput_with_many_clients_one_tenant(
     neon_env_builder: NeonEnvBuilder,
     zenbenchmark: NeonBenchmarker,

test_runner/regress/test_basebackup.py

@@ -1,77 +0,0 @@
-from __future__ import annotations
-
-from typing import TYPE_CHECKING
-
-from fixtures.utils import wait_until
-
-if TYPE_CHECKING:
-    from fixtures.neon_fixtures import NeonEnvBuilder
-
-
-def test_basebackup_cache(neon_env_builder: NeonEnvBuilder):
-    """
-    Simple test for basebackup cache.
-    1. Check that we always hit the cache after compute restart.
-    2. Check that we eventually delete old basebackup files, but not the latest one.
-    3. Check that we delete basebackup file for timeline with active compute.
-    """
-
-    neon_env_builder.pageserver_config_override = """
-        tenant_config = { basebackup_cache_enabled = true }
-        basebackup_cache_config = { cleanup_period = '1s' }
-    """
-
-    env = neon_env_builder.init_start()
-    ep = env.endpoints.create("main")
-    ps = env.pageserver
-    ps_http = ps.http_client()
-
-    # 1. Check that we always hit the cache after compute restart.
-    for i in range(3):
-        ep.start()
-        ep.stop()
-
-        def check_metrics(i=i):
-            metrics = ps_http.get_metrics()
-            # Never miss.
-            # The first time compute_ctl sends `get_basebackup` with lsn=None, we do not cache such requests.
-            # All other requests should be a hit
-            assert (
-                metrics.query_one(
-                    "pageserver_basebackup_cache_read_total", {"result": "miss"}
-                ).value
-                == 0
-            )
-            # All but the first requests are hits.
-            assert (
-                metrics.query_one("pageserver_basebackup_cache_read_total", {"result": "hit"}).value
-                == i
-            )
-            # Every compute shut down should trigger a prepare reuest.
-            assert (
-                metrics.query_one(
-                    "pageserver_basebackup_cache_prepare_total", {"result": "ok"}
-                ).value
-                == i + 1
-            )
-
-        wait_until(check_metrics)
-
-    # 2. Check that we eventually delete old basebackup files, but not the latest one.
-    def check_bb_file_count():
-        bb_files = list(ps.workdir.joinpath("basebackup_cache").iterdir())
-        # tmp dir + 1 basebackup file.
-        assert len(bb_files) == 2
-
-    wait_until(check_bb_file_count)
-
-    # 3. Check that we delete basebackup file for timeline with active compute.
-    ep.start()
-    ep.safe_psql("create table t1 as select generate_series(1, 10) as n")
-
-    def check_bb_dir_empty():
-        bb_files = list(ps.workdir.joinpath("basebackup_cache").iterdir())
-        # only tmp dir.
-        assert len(bb_files) == 1
-
-    wait_until(check_bb_dir_empty)

test_runner/regress/test_compute_metrics.py

@@ -217,11 +217,11 @@ if SQL_EXPORTER is None:
             self, logs_dir: Path, config_file: Path, collector_file: Path, port: int
         ) -> None:
             # NOTE: Keep the version the same as in
-            # compute/compute-node.Dockerfile and build-tools.Dockerfile.
+            # compute/Dockerfile.compute-node and Dockerfile.build-tools.
             #
             # The "host" network mode allows sql_exporter to talk to the
             # endpoint which is running on the host.
-            super().__init__("docker.io/burningalchemist/sql_exporter:0.17.3", network_mode="host")
+            super().__init__("docker.io/burningalchemist/sql_exporter:0.17.0", network_mode="host")
 
             self.__logs_dir = logs_dir
             self.__port = port

test_runner/regress/test_pageserver_metric_collection.py

@@ -508,9 +508,6 @@ PER_METRIC_VERIFIERS = {
     "remote_storage_size": CannotVerifyAnything,
     "written_size": WrittenDataVerifier,
     "written_data_bytes_delta": WrittenDataDeltaVerifier,
-    "written_size_since_parent": WrittenDataVerifier,  # same as written_size on root
-    "pitr_cutoff": CannotVerifyAnything,
-    "pitr_history_size_since_parent": WrittenDataVerifier,  # same as written_size on root w/o GC
     "timeline_logical_size": CannotVerifyAnything,
     "synthetic_storage_size": SyntheticSizeVerifier,
 }

workspace_hack/Cargo.toml

@@ -18,8 +18,6 @@ license.workspace = true
 ahash = { version = "0.8" }
 anstream = { version = "0.6" }
 anyhow = { version = "1", features = ["backtrace"] }
-axum = { version = "0.8", features = ["ws"] }
-axum-core = { version = "0.5", default-features = false, features = ["tracing"] }
 base64-594e8ee84c453af0 = { package = "base64", version = "0.13", features = ["alloc"] }
 base64-647d43efb71741da = { package = "base64", version = "0.21" }
 base64ct = { version = "1", default-features = false, features = ["std"] }
@@ -54,8 +52,9 @@ hex = { version = "0.4", features = ["serde"] }
 hmac = { version = "0.12", default-features = false, features = ["reset"] }
 hyper-582f2526e08bb6a0 = { package = "hyper", version = "0.14", features = ["client", "http1", "http2", "runtime", "server", "stream"] }
 hyper-dff4ba8e3ae991db = { package = "hyper", version = "1", features = ["full"] }
-hyper-util = { version = "0.1", features = ["client-legacy", "server-auto", "service"] }
-indexmap = { version = "2", features = ["serde"] }
+hyper-util = { version = "0.1", features = ["client-legacy", "http1", "http2", "server", "service"] }
+indexmap-dff4ba8e3ae991db = { package = "indexmap", version = "1", default-features = false, features = ["std"] }
+indexmap-f595c2ba2a3f28df = { package = "indexmap", version = "2", features = ["serde"] }
 itertools = { version = "0.12" }
 lazy_static = { version = "1", default-features = false, features = ["spin_no_std"] }
 libc = { version = "0.2", features = ["extra_traits", "use_std"] }
@@ -83,7 +82,7 @@ regex-syntax = { version = "0.8" }
 reqwest = { version = "0.12", default-features = false, features = ["blocking", "json", "rustls-tls", "rustls-tls-native-roots", "stream"] }
 rustls = { version = "0.23", default-features = false, features = ["logging", "ring", "std", "tls12"] }
 rustls-pki-types = { version = "1", features = ["std"] }
-rustls-webpki = { version = "0.103", default-features = false, features = ["ring", "std"] }
+rustls-webpki = { version = "0.102", default-features = false, features = ["ring", "std"] }
 scopeguard = { version = "1" }
 sec1 = { version = "0.7", features = ["pem", "serde", "std", "subtle"] }
 serde = { version = "1", features = ["alloc", "derive"] }
@@ -100,10 +99,11 @@ tikv-jemalloc-sys = { version = "0.6", features = ["profiling", "stats", "unpref
 time = { version = "0.3", features = ["macros", "serde-well-known"] }
 tokio = { version = "1", features = ["full", "test-util"] }
 tokio-rustls = { version = "0.26", default-features = false, features = ["logging", "ring", "tls12"] }
-tokio-stream = { version = "0.1", features = ["net"] }
+tokio-stream = { version = "0.1" }
 tokio-util = { version = "0.7", features = ["codec", "compat", "io", "rt"] }
 toml_edit = { version = "0.22", features = ["serde"] }
-tower = { version = "0.5", default-features = false, features = ["balance", "buffer", "limit", "log"] }
+tonic = { version = "0.12", default-features = false, features = ["codegen", "prost", "tls-roots"] }
+tower = { version = "0.4", default-features = false, features = ["balance", "buffer", "limit", "util"] }
 tracing = { version = "0.1", features = ["log"] }
 tracing-core = { version = "0.1" }
 tracing-log = { version = "0.2" }
@@ -125,7 +125,8 @@ either = { version = "1" }
 getrandom = { version = "0.2", default-features = false, features = ["std"] }
 half = { version = "2", default-features = false, features = ["num-traits"] }
 hashbrown = { version = "0.14", features = ["raw"] }
-indexmap = { version = "2", features = ["serde"] }
+indexmap-dff4ba8e3ae991db = { package = "indexmap", version = "1", default-features = false, features = ["std"] }
+indexmap-f595c2ba2a3f28df = { package = "indexmap", version = "2", features = ["serde"] }
 itertools = { version = "0.12" }
 libc = { version = "0.2", features = ["extra_traits", "use_std"] }
 log = { version = "0.4", default-features = false, features = ["std"] }