DO NOT MERGE [proxy]: Add geo-based routing for replicated projects

2026-06-01 04:20:39 +00:00 · 2025-05-24 12:15:16 +03:00
51 changed files with 1189 additions and 1110 deletions
--- a/.github/workflows/build_and_test.yml
+++ b/.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 }}
--- a/.github/workflows/periodic_pagebench.yml
+++ b/.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
--- a/Cargo.lock
+++ b/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]]
@@ -4339,7 +4339,7 @@ dependencies = [
 "reqwest",
 "rpds",
 "rstest",
- "rustls 0.23.27",
+ "rustls 0.23.18",
 "scopeguard",
 "send-future",
 "serde",
@@ -4358,7 +4358,7 @@ 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",
@@ -4455,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",
 ]
@@ -4837,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",
 ]
@@ -4951,7 +4951,7 @@ dependencies = [
 "inferno 0.12.0",
 "num",
 "paste",
- "prost 0.13.5",
+ "prost 0.13.3",
 ]

 [[package]]
@@ -5056,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]]
@@ -5099,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",
@@ -5121,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",
@@ -5147,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]]
@@ -5195,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",
@@ -5229,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",
@@ -5248,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",
@@ -5472,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",
 ]
@@ -5926,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",
 ]
@@ -6023,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"
@@ -6085,7 +6074,7 @@ dependencies = [
 "regex",
 "remote_storage",
 "reqwest",
- "rustls 0.23.27",
+ "rustls 0.23.18",
 "safekeeper_api",
 "safekeeper_client",
 "scopeguard",
@@ -6102,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",
@@ -6274,7 +6263,7 @@ checksum = "255914a8e53822abd946e2ce8baa41d4cded6b8e938913b7f7b9da5b7ab44335"
 dependencies = [
 "httpdate",
 "reqwest",
- "rustls 0.23.27",
+ "rustls 0.23.18",
 "sentry-backtrace",
 "sentry-contexts",
 "sentry-core",
@@ -6703,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",
@@ -6749,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",
@@ -6764,7 +6753,7 @@ dependencies = [
 "tokio",
 "tokio-postgres",
 "tokio-postgres-rustls",
- "tokio-rustls 0.26.2",
+ "tokio-rustls 0.26.0",
 "tokio-util",
 "tracing",
 "utils",
@@ -6802,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",
@@ -7336,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",
 ]

@@ -7383,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",
 ]

@@ -7485,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",
@@ -7497,27 +7487,6 @@ name = "tonic"
 version = "0.12.3"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "877c5b330756d856ffcc4553ab34a5684481ade925ecc54bcd1bf02b1d0d4d52"
-dependencies = [
- "async-trait",
- "base64 0.22.1",
- "bytes",
- "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",
 "base64 0.22.1",
@@ -7530,12 +7499,13 @@ dependencies = [
 "hyper-util",
 "percent-encoding",
 "pin-project",
- "prost 0.13.5",
+ "prost 0.13.3",
 "rustls-native-certs 0.8.0",
+ "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",
@@ -7543,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",
@@ -7563,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]]
@@ -7578,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",
@@ -7911,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",
@@ -8106,7 +8078,7 @@ dependencies = [
 "pageserver_api",
 "postgres_ffi",
 "pprof",
- "prost 0.13.5",
+ "prost 0.13.3",
 "remote_storage",
 "serde",
 "serde_json",
@@ -8562,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",
@@ -8584,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",
@@ -8611,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",
--- a/Cargo.toml
+++ b/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,7 +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", "tls-ring", "tls-native-roots"] }
+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"] }

@@ -280,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]

--- a/build-tools.Dockerfile
+++ b/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 \
--- a/compute/compute-node.Dockerfile
+++ b/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/
--- a/compute_tools/src/rsyslog.rs
+++ b/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(());
    }

--- a/control_plane/src/endpoint.rs
+++ b/control_plane/src/endpoint.rs
@@ -616,16 +616,19 @@ impl Endpoint {

    /// Map safekeepers ids to the actual connection strings.
    fn build_safekeepers_connstrs(&self, sk_ids: Vec<NodeId>) -> Result<Vec<String>> {
-        sk_ids.into_iter()
-            .map(|node_id| {
-                self.env
+        let mut safekeeper_connstrings = Vec::new();
+        if self.mode == ComputeMode::Primary {
+            for sk_id in sk_ids {
+                let sk = self
+                    .env
                    .safekeepers
                    .iter()
-                    .find(|node| node.id == node_id)
-                    .map(|node| format!("127.0.0.1:{}", node.get_compute_port()))
-                    .ok_or_else(|| anyhow!("safekeeer {node_id} does not exist"))
-            })
-            .collect::<Result<Vec<String>>>()
+                    .find(|node| node.id == sk_id)
+                    .ok_or_else(|| anyhow!("safekeeper {sk_id} does not exist"))?;
+                safekeeper_connstrings.push(format!("127.0.0.1:{}", sk.get_compute_port()));
+            }
+        }
+        Ok(safekeeper_connstrings)
    }

    /// Generate a JWT with the correct claims.
--- a/libs/proxy/postgres-protocol2/src/message/frontend.rs
+++ b/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');
--- a/libs/proxy/postgres-types2/src/lib.rs
+++ b/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);
--- a/libs/proxy/postgres-types2/src/type_gen.rs
+++ b/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)),
--- a/libs/proxy/tokio-postgres2/src/client.rs
+++ b/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.
--- a/libs/proxy/tokio-postgres2/src/codec.rs
+++ b/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,
            }))
        }
    }
--- a/libs/proxy/tokio-postgres2/src/connect.rs
+++ b/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))
 }
--- a/libs/proxy/tokio-postgres2/src/connection.rs
+++ b/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;
            }

--- a/libs/proxy/tokio-postgres2/src/generic_client.rs
+++ b/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
+    }
 }
--- a/libs/proxy/tokio-postgres2/src/lib.rs
+++ b/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 _)
+}
--- a/libs/proxy/tokio-postgres2/src/prepare.rs
+++ b/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)
 }
--- a/libs/proxy/tokio-postgres2/src/query.rs
+++ b/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
+    }
+}
--- a/libs/proxy/tokio-postgres2/src/simple_query.rs
+++ b/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>> {
--- a/libs/proxy/tokio-postgres2/src/statement.rs
+++ b/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,
--- a/libs/proxy/tokio-postgres2/src/transaction.rs
+++ b/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>>,
--- a/libs/walproposer/src/walproposer.rs
+++ b/libs/walproposer/src/walproposer.rs
@@ -213,7 +213,6 @@ impl Wrapper {
            safekeeper_connection_timeout: config.safekeeper_connection_timeout,
            wal_segment_size: WAL_SEGMENT_SIZE as i32, // default 16MB
            syncSafekeepers: config.sync_safekeepers,
-            replicaPromote: false,
            systemId: 0,
            pgTimeline: 1,
            proto_version: 3,
--- a/pgxn/neon/communicator.c
+++ b/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)
--- a/pgxn/neon/file_cache.c
+++ b/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();

--- a/pgxn/neon/neon.h
+++ b/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;
--- a/pgxn/neon/neon_walreader.c
+++ b/pgxn/neon/neon_walreader.c
@@ -69,7 +69,6 @@ struct NeonWALReader
 	WALSegmentContext segcxt;
 	WALOpenSegment seg;
 	int			wre_errno;
-	TimeLineID	local_active_tlid;
 	/* Explains failure to read, static for simplicity. */
 	char		err_msg[NEON_WALREADER_ERR_MSG_LEN];

@@ -107,8 +106,7 @@ struct NeonWALReader

 /* palloc and initialize NeonWALReader */
 NeonWALReader *
-NeonWALReaderAllocate(int wal_segment_size, XLogRecPtr available_lsn,
-					  char *log_prefix, TimeLineID tlid)
+NeonWALReaderAllocate(int wal_segment_size, XLogRecPtr available_lsn, char *log_prefix)
 {
 	NeonWALReader *reader;

@@ -120,7 +118,6 @@ NeonWALReaderAllocate(int wal_segment_size, XLogRecPtr available_lsn,
 		MemoryContextAllocZero(TopMemoryContext, sizeof(NeonWALReader));

 	reader->available_lsn = available_lsn;
-	reader->local_active_tlid = tlid;
 	reader->seg.ws_file = -1;
 	reader->seg.ws_segno = 0;
 	reader->seg.ws_tli = 0;
@@ -580,17 +577,6 @@ NeonWALReaderIsRemConnEstablished(NeonWALReader *state)
 	return state->rem_state == RS_ESTABLISHED;
 }

-/*
- * Whether remote connection is established. Once this is done, until successful
- * local read or error socket is stable and user can update socket events
- * instead of readding it each time.
- */
-TimeLineID
-NeonWALReaderLocalActiveTimeLineID(NeonWALReader *state)
-{
-	return state->local_active_tlid;
-}
-
 /*
 * Returns events user should wait on connection socket or 0 if remote
 * connection is not active.
--- a/pgxn/neon/neon_walreader.h
+++ b/pgxn/neon/neon_walreader.h
@@ -19,12 +19,9 @@ typedef enum
 	NEON_WALREAD_ERROR,
 } NeonWALReadResult;

-extern NeonWALReader *NeonWALReaderAllocate(int wal_segment_size,
-											XLogRecPtr available_lsn,
-											char *log_prefix, TimeLineID tlid);
+extern NeonWALReader *NeonWALReaderAllocate(int wal_segment_size, XLogRecPtr available_lsn, char *log_prefix);
 extern void NeonWALReaderFree(NeonWALReader *state);
 extern void NeonWALReaderResetRemote(NeonWALReader *state);
-extern TimeLineID NeonWALReaderLocalActiveTimeLineID(NeonWALReader *state);
 extern NeonWALReadResult NeonWALRead(NeonWALReader *state, char *buf, XLogRecPtr startptr, Size count, TimeLineID tli);
 extern pgsocket NeonWALReaderSocket(NeonWALReader *state);
 extern uint32 NeonWALReaderEvents(NeonWALReader *state);
--- a/pgxn/neon/walproposer.c
+++ b/pgxn/neon/walproposer.c
@@ -98,7 +98,6 @@ WalProposerCreate(WalProposerConfig *config, walproposer_api api)
 	wp = palloc0(sizeof(WalProposer));
 	wp->config = config;
 	wp->api = api;
-	wp->localTimeLineID = config->pgTimeline;
 	wp->state = WPS_COLLECTING_TERMS;
 	wp->mconf.generation = INVALID_GENERATION;
 	wp->mconf.members.len = 0;
@@ -1380,7 +1379,7 @@ ProcessPropStartPos(WalProposer *wp)
 	 * we must bail out, as clog and other non rel data is inconsistent.
 	 */
 	walprop_shared = wp->api.get_shmem_state(wp);
-	if (!wp->config->syncSafekeepers && !wp->config->replicaPromote)
+	if (!wp->config->syncSafekeepers)
 	{
 		/*
 		 * Basebackup LSN always points to the beginning of the record (not
--- a/pgxn/neon/walproposer.h
+++ b/pgxn/neon/walproposer.h
@@ -739,11 +739,6 @@ typedef struct WalProposerConfig
 	 */
 	bool		syncSafekeepers;

-	/*
-	 * Replica is promoted to primary
-	 */
-	bool		replicaPromote;
-
 	/* Will be passed to safekeepers in greet request. */
 	uint64		systemId;

@@ -810,8 +805,6 @@ typedef struct WalProposer

 	/* WAL has been generated up to this point */
 	XLogRecPtr	availableLsn;
-	/* Current local TimeLineId in use */
-	TimeLineID	localTimeLineID;

 	/* cached GetAcknowledgedByQuorumWALPosition result */
 	XLogRecPtr	commitLsn;
--- a/pgxn/neon/walproposer_pg.c
+++ b/pgxn/neon/walproposer_pg.c
@@ -73,7 +73,6 @@ static XLogRecPtr sentPtr = InvalidXLogRecPtr;
 static const walproposer_api walprop_pg;
 static volatile sig_atomic_t got_SIGUSR2 = false;
 static bool reported_sigusr2 = false;
-static bool start_as_replica = false;

 static XLogRecPtr standby_flush_lsn = InvalidXLogRecPtr;
 static XLogRecPtr standby_apply_lsn = InvalidXLogRecPtr;
@@ -124,7 +123,6 @@ init_walprop_config(bool syncSafekeepers)
 	walprop_config.safekeeper_connection_timeout = wal_acceptor_connection_timeout;
 	walprop_config.wal_segment_size = wal_segment_size;
 	walprop_config.syncSafekeepers = syncSafekeepers;
-	walprop_config.replicaPromote = start_as_replica;
 	if (!syncSafekeepers)
 		walprop_config.systemId = GetSystemIdentifier();
 	else
@@ -151,8 +149,6 @@ WalProposerSync(int argc, char *argv[])
 	WalProposerStart(wp);
 }

-#define GUC_POLL_DELAY 100000L // 0.1 sec
-
 /*
 * WAL proposer bgworker entry point.
 */
@@ -161,18 +157,12 @@ WalProposerMain(Datum main_arg)
 {
 	WalProposer *wp;

-	if (*wal_acceptors_list == '\0')
-	{
-		elog(PANIC, "Safekeepers list is empty");
-	}
-
 	init_walprop_config(false);
 	walprop_pg_init_bgworker();
 	am_walproposer = true;
 	walprop_pg_load_libpqwalreceiver();

 	wp = WalProposerCreate(&walprop_config, walprop_pg);
-	wp->localTimeLineID = GetWALInsertionTimeLine();
 	wp->last_reconnect_attempt = walprop_pg_get_current_timestamp(wp);

 	walprop_pg_init_walsender();
@@ -304,15 +294,16 @@ safekeepers_cmp(char *old, char *new)
 	return true;
 }

+/*
+ * GUC assign_hook for neon.safekeepers. Restarts walproposer through FATAL if
+ * the list changed.
+ */
 static void
 assign_neon_safekeepers(const char *newval, void *extra)
 {
 	char	   *newval_copy;
 	char	   *oldval;

-	if (newval && *newval == '\0')
-		start_as_replica = true;
-
 	if (!am_walproposer)
 		return;

@@ -509,6 +500,10 @@ walprop_register_bgworker(void)
 {
 	BackgroundWorker bgw;

+	/* If no wal acceptors are specified, don't start the background worker. */
+	if (*wal_acceptors_list == '\0')
+		return;
+
 	memset(&bgw, 0, sizeof(bgw));
 	bgw.bgw_flags = BGWORKER_SHMEM_ACCESS;
 	bgw.bgw_start_time = BgWorkerStart_RecoveryFinished;
@@ -519,6 +514,7 @@ walprop_register_bgworker(void)
 	bgw.bgw_restart_time = 1;
 	bgw.bgw_notify_pid = 0;
 	bgw.bgw_main_arg = (Datum) 0;
+
 	RegisterBackgroundWorker(&bgw);
 }

@@ -1500,10 +1496,7 @@ walprop_pg_wal_reader_allocate(Safekeeper *sk)

 	snprintf(log_prefix, sizeof(log_prefix), WP_LOG_PREFIX "sk %s:%s nwr: ", sk->host, sk->port);
 	Assert(!sk->xlogreader);
-	/* note that WalProposer shouldn't access safekeepers when active */
-	sk->xlogreader = NeonWALReaderAllocate(wal_segment_size,
-										   sk->wp->propTermStartLsn, log_prefix,
-										   sk->wp->localTimeLineID);
+	sk->xlogreader = NeonWALReaderAllocate(wal_segment_size, sk->wp->propTermStartLsn, log_prefix);
 	if (sk->xlogreader == NULL)
 		wpg_log(FATAL, "failed to allocate xlog reader");
 }
@@ -1517,7 +1510,7 @@ walprop_pg_wal_read(Safekeeper *sk, char *buf, XLogRecPtr startptr, Size count,
 					  buf,
 					  startptr,
 					  count,
-					  sk->wp->localTimeLineID);
+					  walprop_pg_get_timeline_id());

 	if (res == NEON_WALREAD_SUCCESS)
 	{
@@ -2016,7 +2009,6 @@ walprop_pg_get_redo_start_lsn(WalProposer *wp)
 	return GetRedoStartLsn();
 }

-
 static bool
 walprop_pg_strong_random(WalProposer *wp, void *buf, size_t len)
 {
--- a/pgxn/neon/walsender_hooks.c
+++ b/pgxn/neon/walsender_hooks.c
@@ -68,7 +68,8 @@ NeonWALReadWaitForWAL(XLogRecPtr loc)
 }

 static int
-NeonWALPageRead(XLogReaderState *xlogreader,
+NeonWALPageRead(
+				XLogReaderState *xlogreader,
 				XLogRecPtr targetPagePtr,
 				int reqLen,
 				XLogRecPtr targetRecPtr,
@@ -105,11 +106,12 @@ NeonWALPageRead(XLogReaderState *xlogreader,

 	for (;;)
 	{
-		NeonWALReadResult res = NeonWALRead(wal_reader,
+		NeonWALReadResult res = NeonWALRead(
+											wal_reader,
 											readBuf,
 											targetPagePtr,
 											count,
-											NeonWALReaderLocalActiveTimeLineID(wal_reader));
+											walprop_pg_get_timeline_id());

 		if (res == NEON_WALREAD_SUCCESS)
 		{
@@ -200,8 +202,7 @@ NeonOnDemandXLogReaderRoutines(XLogReaderRoutine *xlr)
 		{
 			elog(ERROR, "unable to start walsender when basebackupLsn is 0");
 		}
-		wal_reader = NeonWALReaderAllocate(wal_segment_size, basebackupLsn,
-										   "[walsender] ", 1);
+		wal_reader = NeonWALReaderAllocate(wal_segment_size, basebackupLsn, "[walsender] ");
 	}
 	xlr->page_read = NeonWALPageRead;
 	xlr->segment_open = NeonWALReadSegmentOpen;
--- a/proxy/src/binary/proxy.rs
+++ b/proxy/src/binary/proxy.rs
@@ -27,6 +27,7 @@ use crate::config::{
    ProxyConfig, ProxyProtocolV2, remote_storage_from_toml,
 };
 use crate::context::parquet::ParquetUploadArgs;
+use crate::control_plane::client::cplane_proxy_v1::{GeoProximity, RegionProximityMap};
 use crate::http::health_server::AppMetrics;
 use crate::metrics::Metrics;
 use crate::rate_limiter::{
@@ -766,12 +767,21 @@ fn build_auth_backend(
            let wake_compute_endpoint_rate_limiter =
                Arc::new(WakeComputeRateLimiter::new(wake_compute_rps_limit));

+            let geo_map = Box::leak(Box::new(RegionProximityMap::from([(
+                args.region.clone(),
+                GeoProximity {
+                    _weight: 1,
+                    _distance: 0,
+                },
+            )])));
+
            let api = control_plane::client::cplane_proxy_v1::NeonControlPlaneClient::new(
                endpoint,
                args.control_plane_token.clone(),
                caches,
                locks,
                wake_compute_endpoint_rate_limiter,
+                geo_map,
            );

            let api = control_plane::client::ControlPlaneClient::ProxyV1(api);
@@ -845,6 +855,14 @@ fn build_auth_backend(
            let wake_compute_endpoint_rate_limiter =
                Arc::new(WakeComputeRateLimiter::new(wake_compute_rps_limit));

+            let geo_map = Box::leak(Box::new(RegionProximityMap::from([(
+                args.region.clone(),
+                GeoProximity {
+                    _weight: 1,
+                    _distance: 0,
+                },
+            )])));
+
            // Since we use only get_allowed_ips_and_secret() wake_compute_endpoint_rate_limiter
            // and locks are not used in ConsoleRedirectBackend,
            // but they are required by the NeonControlPlaneClient
@@ -854,6 +872,7 @@ fn build_auth_backend(
                caches,
                locks,
                wake_compute_endpoint_rate_limiter,
+                geo_map,
            );

            let backend = ConsoleRedirectBackend::new(url, api);
--- a/proxy/src/context/mod.rs
+++ b/proxy/src/context/mod.rs
@@ -296,6 +296,10 @@ impl RequestContext {
            .has_private_peer_addr()
    }

+    pub fn is_global(&self) -> bool {
+        self.0.try_lock().expect("should not deadlock").region == "global"
+    }
+
    pub(crate) fn set_error_kind(&self, kind: ErrorKind) {
        let mut this = self.0.try_lock().expect("should not deadlock");
        // Do not record errors from the private address to metrics.
--- a/proxy/src/control_plane/client/cplane_proxy_v1.rs
+++ b/proxy/src/control_plane/client/cplane_proxy_v1.rs
@@ -1,5 +1,6 @@
 //! Production console backend.

+use std::collections::HashMap;
 use std::net::IpAddr;
 use std::str::FromStr;
 use std::sync::Arc;
@@ -12,7 +13,10 @@ use postgres_client::config::SslMode;
 use tokio::time::Instant;
 use tracing::{Instrument, debug, info, info_span, warn};

-use super::super::messages::{ControlPlaneErrorMessage, GetEndpointAccessControl, WakeCompute};
+use super::super::messages::{
+    ControlPlaneErrorMessage, GetEndpointAccessControl, GetEndpointAccessControlReplicated,
+    WakeCompute,
+};
 use crate::auth::backend::ComputeUserInfo;
 use crate::auth::backend::jwt::AuthRule;
 use crate::cache::Cached;
@@ -34,6 +38,15 @@ use crate::{compute, http, scram};

 pub(crate) const X_REQUEST_ID: HeaderName = HeaderName::from_static("x-request-id");

+pub type RegionId = String;
+pub type RegionProximityMap = HashMap<RegionId, GeoProximity>;
+
+#[derive(Clone)]
+pub struct GeoProximity {
+    pub _weight: u64,   // load or preference-based parameter
+    pub _distance: u64, // approximate distance from the region to the current proxy
+}
+
 #[derive(Clone)]
 pub struct NeonControlPlaneClient {
    endpoint: http::Endpoint,
@@ -42,6 +55,7 @@ pub struct NeonControlPlaneClient {
    pub(crate) wake_compute_endpoint_rate_limiter: Arc<WakeComputeRateLimiter>,
    // put in a shared ref so we don't copy secrets all over in memory
    jwt: Arc<str>,
+    geo_map: &'static RegionProximityMap,
 }

 impl NeonControlPlaneClient {
@@ -52,6 +66,7 @@ impl NeonControlPlaneClient {
        caches: &'static ApiCaches,
        locks: &'static ApiLocks<EndpointCacheKey>,
        wake_compute_endpoint_rate_limiter: Arc<WakeComputeRateLimiter>,
+        geo_map: &'static RegionProximityMap,
    ) -> Self {
        Self {
            endpoint,
@@ -59,6 +74,7 @@ impl NeonControlPlaneClient {
            locks,
            wake_compute_endpoint_rate_limiter,
            jwt,
+            geo_map,
        }
    }

@@ -81,10 +97,126 @@ impl NeonControlPlaneClient {
            info!("endpoint is not valid, skipping the request");
            return Ok(AuthInfo::default());
        }
+
+        if ctx.is_global() {
+            return self
+                .do_get_auth_req_replicated(user_info, &ctx.session_id(), Some(ctx))
+                .await;
+        }
+
        self.do_get_auth_req(user_info, &ctx.session_id(), Some(ctx))
            .await
    }

+    async fn do_get_auth_req_replicated(
+        &self,
+        user_info: &ComputeUserInfo,
+        session_id: &uuid::Uuid,
+        ctx: Option<&RequestContext>,
+    ) -> Result<AuthInfo, GetAuthInfoError> {
+        let request_id: String = session_id.to_string();
+        let application_name = if let Some(ctx) = ctx {
+            ctx.console_application_name()
+        } else {
+            "auth_cancellation".to_string()
+        };
+
+        async {
+            let request = self
+                .endpoint
+                .get_path("get_endpoint_access_control_replicated")
+                .header(X_REQUEST_ID, &request_id)
+                .header(AUTHORIZATION, format!("Bearer {}", &self.jwt))
+                .query(&[("session_id", session_id)])
+                .query(&[
+                    ("application_name", application_name.as_str()),
+                    ("endpointish", user_info.endpoint.as_str()),
+                    ("role", user_info.user.as_str()),
+                ])
+                .build()?;
+
+            debug!(url = request.url().as_str(), "sending http request");
+            let start = Instant::now();
+            let response = match ctx {
+                Some(ctx) => {
+                    let pause = ctx.latency_timer_pause(crate::metrics::Waiting::Cplane);
+                    let rsp = self.endpoint.execute(request).await;
+                    drop(pause);
+                    rsp?
+                }
+                None => self.endpoint.execute(request).await?,
+            };
+
+            info!(duration = ?start.elapsed(), "received http response");
+            let body = match parse_body::<GetEndpointAccessControlReplicated>(response).await {
+                Ok(body) => body,
+                // Error 404 is special: it's ok not to have a secret.
+                // TODO(anna): retry
+                Err(e) => {
+                    return if e.get_reason().is_not_found() {
+                        // TODO: refactor this because it's weird
+                        // this is a failure to authenticate but we return Ok.
+                        Ok(AuthInfo::default())
+                    } else {
+                        Err(e.into())
+                    };
+                }
+            };
+
+            for endpoint in body.endpoints {
+                if let Some(region_id) = &endpoint.region_id {
+                    if let Some(_proximity) = self.geo_map.get(region_id) {
+                        // TODO:: calculate proximity and reroute
+
+                        let secret = if endpoint.role_secret.is_empty() {
+                            None
+                        } else {
+                            let secret = scram::ServerSecret::parse(&endpoint.role_secret)
+                                .map(AuthSecret::Scram)
+                                .ok_or(GetAuthInfoError::BadSecret)?;
+                            Some(secret)
+                        };
+                        let allowed_ips = endpoint.allowed_ips.unwrap_or_default();
+                        Metrics::get()
+                            .proxy
+                            .allowed_ips_number
+                            .observe(allowed_ips.len() as f64);
+                        let allowed_vpc_endpoint_ids =
+                            endpoint.allowed_vpc_endpoint_ids.unwrap_or_default();
+                        Metrics::get()
+                            .proxy
+                            .allowed_vpc_endpoint_ids
+                            .observe(allowed_vpc_endpoint_ids.len() as f64);
+                        let block_public_connections =
+                            endpoint.block_public_connections.unwrap_or_default();
+                        let block_vpc_connections =
+                            endpoint.block_vpc_connections.unwrap_or_default();
+
+                        // return the closest replica
+                        return Ok(AuthInfo {
+                            secret,
+                            allowed_ips,
+                            allowed_vpc_endpoint_ids,
+                            project_id: endpoint.project_id,
+                            account_id: endpoint.account_id,
+                            access_blocker_flags: AccessBlockerFlags {
+                                public_access_blocked: block_public_connections,
+                                vpc_access_blocked: block_vpc_connections,
+                            },
+                        });
+                    }
+
+                    return Err(GetAuthInfoError::RegionNotFound);
+                }
+            }
+
+            Err(GetAuthInfoError::RegionNotFound)
+        }
+        .inspect_err(|e| tracing::debug!(error = ?e))
+        .instrument(info_span!("do_get_auth_info"))
+        .await
+    }
+
    async fn do_get_auth_req(
        &self,
        user_info: &ComputeUserInfo,
--- a/proxy/src/control_plane/errors.rs
+++ b/proxy/src/control_plane/errors.rs
@@ -99,6 +99,9 @@ pub(crate) enum GetAuthInfoError {

    #[error(transparent)]
    ApiError(ControlPlaneError),
+
+    #[error("No endpoint found in this region")]
+    RegionNotFound,
 }

 // This allows more useful interactions than `#[from]`.
@@ -115,6 +118,7 @@ impl UserFacingError for GetAuthInfoError {
            Self::BadSecret => REQUEST_FAILED.to_owned(),
            // However, API might return a meaningful error.
            Self::ApiError(e) => e.to_string_client(),
+            Self::RegionNotFound => "No endpoint found in this region".to_owned(),
        }
    }
 }
@@ -124,6 +128,7 @@ impl ReportableError for GetAuthInfoError {
        match self {
            Self::BadSecret => crate::error::ErrorKind::ControlPlane,
            Self::ApiError(_) => crate::error::ErrorKind::ControlPlane,
+            Self::RegionNotFound => crate::error::ErrorKind::User,
        }
    }
 }
--- a/proxy/src/control_plane/messages.rs
+++ b/proxy/src/control_plane/messages.rs
@@ -222,6 +222,25 @@ pub(crate) struct UserFacingMessage {
    pub(crate) message: Box<str>,
 }

+/// Response which holds client's auth secret, e.g. [`crate::scram::ServerSecret`].
+/// Returned by the `/get_endpoint_access_control_replicated` API method.
+#[derive(Deserialize)]
+pub(crate) struct GetEndpointAccessControlReplicated {
+    pub(crate) endpoints: Vec<EndpointAccessControlReplicated>,
+}
+
+#[derive(Deserialize)]
+pub(crate) struct EndpointAccessControlReplicated {
+    pub(crate) role_secret: Box<str>,
+    pub(crate) allowed_ips: Option<Vec<IpPattern>>,
+    pub(crate) allowed_vpc_endpoint_ids: Option<Vec<String>>,
+    pub(crate) project_id: Option<ProjectIdInt>,
+    pub(crate) account_id: Option<AccountIdInt>,
+    pub(crate) block_public_connections: Option<bool>,
+    pub(crate) block_vpc_connections: Option<bool>,
+    pub(crate) region_id: Option<String>,
+}
+
 /// Response which holds client's auth secret, e.g. [`crate::scram::ServerSecret`].
 /// Returned by the `/get_endpoint_access_control` API method.
 #[derive(Deserialize)]
--- a/proxy/src/serverless/sql_over_http.rs
+++ b/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,41 +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);
-    let mut types = 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",
-        }));
-
-        types.push(c.type_().clone());
-    }
-
-    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 {
@@ -1136,26 +1115,13 @@ async fn query_to_json<T: GenericClient>(
                config.max_response_size_bytes,
            ));
        }
-
-        let row = pg_text_row_to_json(&row, &types, 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" {
@@ -1176,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(),
--- a/scripts/benchmark_durations.py
+++ b/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,
--- a/storage_broker/src/bin/storage_broker.rs
+++ b/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)
--- a/test_runner/fixtures/neon_fixtures.py
+++ b/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")
@@ -4658,7 +4657,7 @@ class EndpointFactory:
        origin: Endpoint,
        endpoint_id: str | None = None,
        config_lines: list[str] | None = None,
-    ) -> Endpoint:
+    ):
        branch_name = origin.branch_name
        assert origin in self.endpoints
        assert branch_name is not None
@@ -4677,7 +4676,7 @@ class EndpointFactory:
        origin: Endpoint,
        endpoint_id: str | None = None,
        config_lines: list[str] | None = None,
-    ) -> Endpoint:
+    ):
        branch_name = origin.branch_name
        assert origin in self.endpoints
        assert branch_name is not None
--- a/test_runner/performance/pageserver/pagebench/test_pageserver_max_throughput_getpage_at_latest_lsn.py
+++ b/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,
--- a/test_runner/regress/test_compute_metrics.py
+++ b/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
--- a/test_runner/regress/test_hot_standby.py
+++ b/test_runner/regress/test_hot_standby.py
@@ -74,9 +74,8 @@ def test_hot_standby(neon_simple_env: NeonEnv):
                for query in queries:
                    with s_con.cursor() as secondary_cursor:
                        secondary_cursor.execute(query)
-                        res = secondary_cursor.fetchone()
-                        assert res is not None
-                        response = res
+                        response = secondary_cursor.fetchone()
+                        assert response is not None
                        assert response == responses[query]

            # Check for corrupted WAL messages which might otherwise go unnoticed if
@@ -165,7 +164,7 @@ def test_hot_standby_gc(neon_env_builder: NeonEnvBuilder, pause_apply: bool):

            s_cur.execute("SELECT COUNT(*) FROM test")
            res = s_cur.fetchone()
-            assert res == (10000,)
+            assert res[0] == 10000

            # Clear the cache in the standby, so that when we
            # re-execute the query, it will make GetPage
@@ -196,7 +195,7 @@ def test_hot_standby_gc(neon_env_builder: NeonEnvBuilder, pause_apply: bool):
            s_cur.execute("SELECT COUNT(*) FROM test")
            log_replica_lag(primary, secondary)
            res = s_cur.fetchone()
-            assert res == (10000,)
+            assert res[0] == 10000


 def run_pgbench(connstr: str, pg_bin: PgBin):
--- a/test_runner/regress/test_replica_promotes.py
+++ b/test_runner/regress/test_replica_promotes.py
@@ -1,130 +0,0 @@
-"""
-File with secondary->primary promotion testing.
-
-This far, only contains a test that we don't break and that the data is persisted.
-"""
-
-import psycopg2
-
-from fixtures.log_helper import log
-from fixtures.neon_fixtures import Endpoint, NeonEnv, wait_replica_caughtup, wait_for_last_flush_lsn
-from fixtures.pg_version import PgVersion
-from pytest import raises
-
-
-def test_replica_promotes(neon_simple_env: NeonEnv, pg_version: PgVersion):
-    """
-    Test that a replica safely promotes, and can commit data updates which
-    show up when the primary boots up after the promoted secondary endpoint
-    shut down.
-    """
-
-    # Initialize the primary, a test table, and a helper function to create lots
-    # of subtransactions.
-    env: NeonEnv = neon_simple_env
-    primary: Endpoint = env.endpoints.create_start(branch_name="main", endpoint_id="primary")
-    secondary: Endpoint = env.endpoints.new_replica_start(origin=primary, endpoint_id="secondary")
-
-    with primary.connect() as primary_conn:
-        primary_cur = primary_conn.cursor()
-        primary_cur.execute(
-            "create table t(pk bigint GENERATED ALWAYS AS IDENTITY, payload integer)"
-        )
-        primary_cur.execute("INSERT INTO t(payload) SELECT generate_series(1, 100)")
-        primary_cur.execute(
-            """
-            SELECT pg_current_wal_insert_lsn(),
-                   pg_current_wal_lsn(),
-                   pg_current_wal_flush_lsn()
-            """
-        )
-        log.info(f"Primary: Current LSN after workload is {primary_cur.fetchone()}")
-
-    wait_replica_caughtup(primary, secondary)
-
-    with secondary.connect() as secondary_conn:
-        secondary_cur = secondary_conn.cursor()
-        secondary_cur.execute("select count(*) from t")
-
-        assert secondary_cur.fetchone() == (100,)
-
-        with raises(psycopg2.Error):
-            secondary_cur.execute("INSERT INTO t (payload) SELECT generate_series(101, 200)")
-            secondary_conn.commit()
-
-        secondary_conn.rollback()
-        secondary_cur.execute("select count(*) from t")
-        assert secondary_cur.fetchone() == (100,)
-
-    primary.stop_and_destroy(mode="immediate")
-    primary = None
-
-    # Reconnect to the secondary to make sure we get a read-write connection
-    with secondary.connect() as promo_conn:
-        promo_cur = promo_conn.cursor()
-
-        promo_cur.execute("SELECT * FROM pg_promote()")
-        assert promo_cur.fetchone() == (True,)
-        promo_cur.execute(
-            """
-            SELECT pg_current_wal_insert_lsn(),
-                   pg_current_wal_lsn(),
-                   pg_current_wal_flush_lsn()
-            """
-        )
-        log.info(f"Secondary: LSN after promotion is {promo_cur.fetchone()}")
-
-    # Reconnect to the secondary to make sure we get a read-write connection
-    with secondary.connect() as new_primary_conn:
-        new_primary_cur = new_primary_conn.cursor()
-        new_primary_cur.execute("select count(*) from t")
-        assert new_primary_cur.fetchone() == (100,)
-
-        new_primary_cur.execute("INSERT INTO t (payload) SELECT generate_series(101, 200) RETURNING payload")
-        assert new_primary_cur.fetchall() == [(it,) for it in range(101, 201)]
-
-        new_primary_cur = new_primary_conn.cursor()
-        new_primary_cur.execute("select payload from t")
-        assert new_primary_cur.fetchall() == [(it,) for it in range(1, 201)]
-
-        new_primary_cur.execute("select count(*) from t")
-        assert new_primary_cur.fetchone() == (200,)
-        new_primary_cur.execute(
-            """
-            SELECT pg_current_wal_insert_lsn(),
-                   pg_current_wal_lsn(),
-                   pg_current_wal_flush_lsn()
-            """
-        )
-        log.info(f"Secondary: LSN after workload is {new_primary_cur.fetchone()}")
-
-
-    with secondary.connect() as second_viewpoint_conn:
-        new_primary_cur = second_viewpoint_conn.cursor()
-        new_primary_cur.execute("select payload from t")
-        assert new_primary_cur.fetchall() == [(it,) for it in range(1, 201)]
-
-    wait_for_last_flush_lsn(env, secondary, env.initial_tenant, env.initial_timeline)
-
-    secondary.stop_and_destroy()
-
-    primary = env.endpoints.create_start(branch_name="main", endpoint_id="primary")
-
-    with primary.connect() as new_primary:
-        new_primary_cur = new_primary.cursor()
-        new_primary_cur.execute(
-            """
-            SELECT pg_current_wal_insert_lsn(),
-                   pg_current_wal_lsn(),
-                   pg_current_wal_flush_lsn()
-            """
-        )
-        log.info(f"New primary: Boot LSN is {new_primary_cur.fetchone()}")
-
-        new_primary_cur.execute("select count(*) from t")
-        assert new_primary_cur.fetchone() == (200,)
-        new_primary_cur.execute("INSERT INTO t (payload) SELECT generate_series(201, 300)")
-        new_primary_cur.execute("select count(*) from t")
-        assert new_primary_cur.fetchone() == (300,)
-
-    primary.stop(mode="immediate")
--- a/vendor/postgres-v14
+++ b/vendor/postgres-v14
--- a/vendor/postgres-v15
+++ b/vendor/postgres-v15
--- a/vendor/postgres-v16
+++ b/vendor/postgres-v16
--- a/vendor/postgres-v17
+++ b/vendor/postgres-v17
--- a/vendor/revisions.json
+++ b/vendor/revisions.json
@@ -1,18 +1,18 @@
 {
  "v17": [
    "17.5",
-    "28b88cfedf9a6028cf7ea32a80ea15a9bf971803"
+    "8be779fd3ab9e87206da96a7e4842ef1abf04f44"
  ],
  "v16": [
    "16.9",
-    "3d6896afaa89c05b693e9a29275001b801b8b479"
+    "0bf96bd6d70301a0b43b0b3457bb3cf8fb43c198"
  ],
  "v15": [
    "15.13",
-    "f2cd31737536218c39c66f920ff62f72c51d0d61"
+    "de7640f55da07512834d5cc40c4b3fb376b5f04f"
  ],
  "v14": [
    "14.18",
-    "78c1568afbbacc1604ac5e5cc2ebc9b7ed0cfde9"
+    "55c0d45abe6467c02084c2192bca117eda6ce1e7"
  ]
 }
--- a/workspace_hack/Cargo.toml
+++ b/workspace_hack/Cargo.toml
@@ -53,7 +53,8 @@ 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", "http1", "http2", "server", "service"] }
-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" }
 lazy_static = { version = "1", default-features = false, features = ["spin_no_std"] }
 libc = { version = "0.2", features = ["extra_traits", "use_std"] }
@@ -81,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"] }
@@ -101,7 +102,8 @@ tokio-rustls = { version = "0.26", default-features = false, features = ["loggin
 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" }
@@ -123,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"] }