cancellation does not need tls

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

@@ -38,11 +38,6 @@ on:
         required: false
         default: 1
         type: number
-      rerun-failed:
-        description: 'rerun failed tests to ignore flaky tests'
-        required: false
-        default: true
-        type: boolean
 
 defaults:
   run:
@@ -104,10 +99,11 @@ jobs:
 
       # Set some environment variables used by all the steps.
       #
-      # CARGO_FLAGS is extra options to pass to all "cargo" subcommands.
+      # CARGO_FLAGS is extra options to pass to "cargo build", "cargo test" etc.
+      #   It also includes --features, if any
       #
-      # CARGO_PROFILE is passed to "cargo build", "cargo test" etc, but not to
-      #   "cargo metadata", because it doesn't accept --release or --debug options.
+      # CARGO_FEATURES is passed to "cargo metadata". It is separate from CARGO_FLAGS,
+      #   because "cargo metadata" doesn't accept --release or --debug options
       #
       # We run tests with addtional features, that are turned off by default (e.g. in release builds), see
       # corresponding Cargo.toml files for their descriptions.
@@ -116,16 +112,16 @@ jobs:
           ARCH: ${{ inputs.arch }}
           SANITIZERS: ${{ inputs.sanitizers }}
         run: |
-          CARGO_FLAGS="--locked --features testing"
+          CARGO_FEATURES="--features testing"
           if [[ $BUILD_TYPE == "debug" && $ARCH == 'x64' ]]; then
             cov_prefix="scripts/coverage --profraw-prefix=$GITHUB_JOB --dir=/tmp/coverage run"
-            CARGO_PROFILE=""
+            CARGO_FLAGS="--locked"
           elif [[ $BUILD_TYPE == "debug" ]]; then
             cov_prefix=""
-            CARGO_PROFILE=""
+            CARGO_FLAGS="--locked"
           elif [[ $BUILD_TYPE == "release" ]]; then
             cov_prefix=""
-            CARGO_PROFILE="--release"
+            CARGO_FLAGS="--locked --release"
           fi
           if [[ $SANITIZERS == 'enabled' ]]; then
             make_vars="WITH_SANITIZERS=yes"
@@ -135,8 +131,8 @@ jobs:
           {
             echo "cov_prefix=${cov_prefix}"
             echo "make_vars=${make_vars}"
+            echo "CARGO_FEATURES=${CARGO_FEATURES}"
             echo "CARGO_FLAGS=${CARGO_FLAGS}"
-            echo "CARGO_PROFILE=${CARGO_PROFILE}"
             echo "CARGO_HOME=${GITHUB_WORKSPACE}/.cargo"
           } >> $GITHUB_ENV
 
@@ -188,18 +184,34 @@ jobs:
           path: pg_install/v17
           key: v1-${{ runner.os }}-${{ runner.arch }}-${{ inputs.build-type }}-pg-${{ steps.pg_v17_rev.outputs.pg_rev }}-bookworm-${{ hashFiles('Makefile', 'build-tools.Dockerfile') }}
 
-      - name: Build all
-        # Note: the Makefile picks up BUILD_TYPE and CARGO_PROFILE from the env variables
-        run: mold -run make ${make_vars} all -j$(nproc) CARGO_BUILD_FLAGS="$CARGO_FLAGS"
+      - name: Build postgres v14
+        if: steps.cache_pg_14.outputs.cache-hit != 'true'
+        run: mold -run make ${make_vars} postgres-v14 -j$(nproc)
+
+      - name: Build postgres v15
+        if: steps.cache_pg_15.outputs.cache-hit != 'true'
+        run: mold -run make ${make_vars} postgres-v15 -j$(nproc)
+
+      - name: Build postgres v16
+        if: steps.cache_pg_16.outputs.cache-hit != 'true'
+        run: mold -run make ${make_vars} postgres-v16 -j$(nproc)
+
+      - name: Build postgres v17
+        if: steps.cache_pg_17.outputs.cache-hit != 'true'
+        run: mold -run make ${make_vars} postgres-v17 -j$(nproc)
+
+      - name: Build neon extensions
+        run: mold -run make ${make_vars} neon-pg-ext -j$(nproc)
 
       - name: Build walproposer-lib
         run: mold -run make ${make_vars} walproposer-lib -j$(nproc)
 
-      - name: Build unit tests
-        if: inputs.sanitizers != 'enabled'
+      - name: Run cargo build
+        env:
+          WITH_TESTS: ${{ inputs.sanitizers != 'enabled' && '--tests' || '' }}
         run: |
           export ASAN_OPTIONS=detect_leaks=0
-          ${cov_prefix} mold -run cargo build $CARGO_FLAGS $CARGO_PROFILE --tests
+          ${cov_prefix} mold -run cargo build $CARGO_FLAGS $CARGO_FEATURES --bins ${WITH_TESTS}
 
       # Do install *before* running rust tests because they might recompile the
       # binaries with different features/flags.
@@ -211,7 +223,7 @@ jobs:
           # Install target binaries
           mkdir -p /tmp/neon/bin/
           binaries=$(
-            ${cov_prefix} cargo metadata $CARGO_FLAGS --format-version=1 --no-deps |
+            ${cov_prefix} cargo metadata $CARGO_FEATURES --format-version=1 --no-deps |
             jq -r '.packages[].targets[] | select(.kind | index("bin")) | .name'
           )
           for bin in $binaries; do
@@ -228,7 +240,7 @@ jobs:
             mkdir -p /tmp/neon/test_bin/
 
             test_exe_paths=$(
-              ${cov_prefix} cargo test $CARGO_FLAGS $CARGO_PROFILE --message-format=json --no-run |
+              ${cov_prefix} cargo test $CARGO_FLAGS $CARGO_FEATURES --message-format=json --no-run |
               jq -r '.executable | select(. != null)'
             )
             for bin in $test_exe_paths; do
@@ -262,10 +274,10 @@ jobs:
           export LD_LIBRARY_PATH
 
           #nextest does not yet support running doctests
-          ${cov_prefix} cargo test --doc $CARGO_FLAGS $CARGO_PROFILE
+          ${cov_prefix} cargo test --doc $CARGO_FLAGS $CARGO_FEATURES
 
           # run all non-pageserver tests
-          ${cov_prefix} cargo nextest run $CARGO_FLAGS $CARGO_PROFILE -E '!package(pageserver)'
+          ${cov_prefix} cargo nextest run $CARGO_FLAGS $CARGO_FEATURES -E '!package(pageserver)'
 
           # run pageserver tests
           # (When developing new pageserver features gated by config fields, we commonly make the rust
@@ -274,13 +286,13 @@ jobs:
           # pageserver tests from non-pageserver tests cuts down the time it takes for this CI step.)
           NEON_PAGESERVER_UNIT_TEST_VIRTUAL_FILE_IOENGINE=tokio-epoll-uring  \
           ${cov_prefix} \
-          cargo nextest run $CARGO_FLAGS $CARGO_PROFILE  -E 'package(pageserver)'
+          cargo nextest run $CARGO_FLAGS $CARGO_FEATURES  -E 'package(pageserver)'
 
           # Run separate tests for real S3
           export ENABLE_REAL_S3_REMOTE_STORAGE=nonempty
           export REMOTE_STORAGE_S3_BUCKET=neon-github-ci-tests
           export REMOTE_STORAGE_S3_REGION=eu-central-1
-          ${cov_prefix} cargo nextest run $CARGO_FLAGS $CARGO_PROFILE -E 'package(remote_storage)' -E 'test(test_real_s3)'
+          ${cov_prefix} cargo nextest run $CARGO_FLAGS $CARGO_FEATURES -E 'package(remote_storage)' -E 'test(test_real_s3)'
 
           # Run separate tests for real Azure Blob Storage
           # XXX: replace region with `eu-central-1`-like region
@@ -289,17 +301,17 @@ jobs:
           export AZURE_STORAGE_ACCESS_KEY="${{ secrets.AZURE_STORAGE_ACCESS_KEY_DEV }}"
           export REMOTE_STORAGE_AZURE_CONTAINER="${{ vars.REMOTE_STORAGE_AZURE_CONTAINER }}"
           export REMOTE_STORAGE_AZURE_REGION="${{ vars.REMOTE_STORAGE_AZURE_REGION }}"
-          ${cov_prefix} cargo nextest run $CARGO_FLAGS $CARGO_PROFILE -E 'package(remote_storage)' -E 'test(test_real_azure)'
+          ${cov_prefix} cargo nextest run $CARGO_FLAGS $CARGO_FEATURES -E 'package(remote_storage)' -E 'test(test_real_azure)'
 
       - name: Install postgres binaries
         run: |
           # Use tar to copy files matching the pattern, preserving the paths in the destionation
           tar c \
             pg_install/v* \
-            build/*/src/test/regress/*.so \
-            build/*/src/test/regress/pg_regress \
-            build/*/src/test/isolation/isolationtester \
-            build/*/src/test/isolation/pg_isolation_regress \
+            pg_install/build/*/src/test/regress/*.so \
+            pg_install/build/*/src/test/regress/pg_regress \
+            pg_install/build/*/src/test/isolation/isolationtester \
+            pg_install/build/*/src/test/isolation/pg_isolation_regress \
             | tar  x -C /tmp/neon
 
       - name: Upload Neon artifact
@@ -367,7 +379,7 @@ jobs:
       - name: Pytest regression tests
         continue-on-error: ${{ matrix.lfc_state == 'with-lfc' && inputs.build-type == 'debug' }}
         uses: ./.github/actions/run-python-test-set
-        timeout-minutes: ${{ (inputs.build-type == 'release' && inputs.sanitizers != 'enabled') && 75 || 180 }}
+        timeout-minutes: ${{ inputs.sanitizers != 'enabled' && 75 || 180 }}
         with:
           build_type: ${{ inputs.build-type }}
           test_selection: regress
@@ -375,14 +387,14 @@ jobs:
           run_with_real_s3: true
           real_s3_bucket: neon-github-ci-tests
           real_s3_region: eu-central-1
-          rerun_failed: ${{ inputs.rerun-failed }}
+          rerun_failed: ${{ inputs.test-run-count == 1 }}
           pg_version: ${{ matrix.pg_version }}
           sanitizers: ${{ inputs.sanitizers }}
           aws-oidc-role-arn: ${{ vars.DEV_AWS_OIDC_ROLE_ARN }}
           # `--session-timeout` is equal to (timeout-minutes - 10 minutes) * 60 seconds.
           # Attempt to stop tests gracefully to generate test reports
           # until they are forcibly stopped by the stricter `timeout-minutes` limit.
-          extra_params: --session-timeout=${{ (inputs.build-type == 'release' && inputs.sanitizers != 'enabled') && 3000 || 10200 }} --count=${{ inputs.test-run-count }}
+          extra_params: --session-timeout=${{ inputs.sanitizers != 'enabled' && 3000 || 10200 }} --count=${{ inputs.test-run-count }}
                         ${{ inputs.test-selection != '' && format('-k "{0}"', inputs.test-selection) || '' }}
         env:
           TEST_RESULT_CONNSTR: ${{ secrets.REGRESS_TEST_RESULT_CONNSTR_NEW }}

.github/workflows/build-macos.yml

@@ -110,7 +110,7 @@ jobs:
 
   build-walproposer-lib:
     if: |
-      contains(inputs.pg_versions, 'v17') || inputs.rebuild_everything ||
+      inputs.pg_versions != '[]' || inputs.rebuild_everything ||
       contains(github.event.pull_request.labels.*.name, 'run-extra-build-macos')  ||
       contains(github.event.pull_request.labels.*.name, 'run-extra-build-*') ||
       github.ref_name == 'main'
@@ -144,7 +144,7 @@ jobs:
         id: cache_walproposer_lib
         uses: actions/cache@5a3ec84eff668545956fd18022155c47e93e2684 # v4.2.3
         with:
-          path: build/walproposer-lib
+          path: pg_install/build/walproposer-lib
           key: v1-${{ runner.os }}-${{ runner.arch }}-${{ env.BUILD_TYPE }}-walproposer_lib-v17-${{ steps.pg_rev.outputs.pg_rev }}-${{ hashFiles('Makefile') }}
 
       - name: Checkout submodule vendor/postgres-v17
@@ -169,11 +169,11 @@ jobs:
         run:
           make walproposer-lib -j$(sysctl -n hw.ncpu)
 
-      - name: Upload "build/walproposer-lib" artifact
+      - name: Upload "pg_install/build/walproposer-lib" artifact
         uses: actions/upload-artifact@ea165f8d65b6e75b540449e92b4886f43607fa02 # v4.6.2
         with:
-          name: build--walproposer-lib
-          path: build/walproposer-lib
+          name: pg_install--build--walproposer-lib
+          path: pg_install/build/walproposer-lib
           # The artifact is supposed to be used by the next job in the same workflow,
           # so there’s no need to store it for too long.
           retention-days: 1
@@ -226,11 +226,11 @@ jobs:
           name: pg_install--v17
           path: pg_install/v17
 
-      - name: Download "build/walproposer-lib" artifact
+      - name: Download "pg_install/build/walproposer-lib" artifact
         uses: actions/download-artifact@d3f86a106a0bac45b974a628896c90dbdf5c8093 # v4.3.0
         with:
-          name: build--walproposer-lib
-          path: build/walproposer-lib
+          name: pg_install--build--walproposer-lib
+          path: pg_install/build/walproposer-lib
 
       # `actions/download-artifact` doesn't preserve permissions:
       # https://github.com/actions/download-artifact?tab=readme-ov-file#permission-loss

.github/workflows/build_and_run_selected_test.yml

@@ -58,7 +58,6 @@ jobs:
       test-cfg: ${{ inputs.pg-versions }}
       test-selection: ${{ inputs.test-selection }}
       test-run-count: ${{ fromJson(inputs.run-count) }}
-      rerun-failed: false
     secrets: inherit
 
   create-test-report:

.github/workflows/build_and_test.yml

@@ -199,28 +199,6 @@ jobs:
       build-tools-image: ${{ needs.build-build-tools-image.outputs.image }}-bookworm
     secrets: inherit
 
-  validate-compute-manifest:
-    runs-on: ubuntu-22.04
-    needs: [ meta, check-permissions ]
-    # We do need to run this in `.*-rc-pr` because of hotfixes.
-    if: ${{ contains(fromJSON('["pr", "push-main", "storage-rc-pr", "proxy-rc-pr", "compute-rc-pr"]'), needs.meta.outputs.run-kind) }}
-    steps:
-      - name: Harden the runner (Audit all outbound calls)
-        uses: step-security/harden-runner@4d991eb9b905ef189e4c376166672c3f2f230481 # v2.11.0
-        with:
-          egress-policy: audit
-
-      - uses: actions/checkout@11bd71901bbe5b1630ceea73d27597364c9af683 # v4.2.2
-
-      - name: Set up Node.js
-        uses: actions/setup-node@49933ea5288caeca8642d1e84afbd3f7d6820020 # v4.4.0
-        with:
-          node-version: '24'
-
-      - name: Validate manifest against schema
-        run: |
-          make -C compute manifest-schema-validation
-
   build-and-test-locally:
     needs: [ meta, build-build-tools-image ]
     # We do need to run this in `.*-rc-pr` because of hotfixes.
@@ -670,7 +648,7 @@ jobs:
                                              ghcr.io/neondatabase/neon:${{ needs.meta.outputs.build-tag }}-bookworm-arm64
 
   compute-node-image-arch:
-    needs: [ check-permissions, meta ]
+    needs: [ check-permissions, build-build-tools-image, meta ]
     if: ${{ contains(fromJSON('["push-main", "pr", "compute-rc-pr"]'), needs.meta.outputs.run-kind) }}
     permissions:
       id-token: write # aws-actions/configure-aws-credentials
@@ -743,6 +721,7 @@ jobs:
             GIT_VERSION=${{ github.event.pull_request.head.sha || github.sha }}
             PG_VERSION=${{ matrix.version.pg }}
             BUILD_TAG=${{ needs.meta.outputs.release-tag || needs.meta.outputs.build-tag }}
+            TAG=${{ needs.build-build-tools-image.outputs.image-tag }}-${{ matrix.version.debian }}
             DEBIAN_VERSION=${{ matrix.version.debian }}
           provenance: false
           push: true
@@ -762,6 +741,7 @@ jobs:
             GIT_VERSION=${{ github.event.pull_request.head.sha || github.sha }}
             PG_VERSION=${{ matrix.version.pg }}
             BUILD_TAG=${{ needs.meta.outputs.release-tag || needs.meta.outputs.build-tag }}
+            TAG=${{ needs.build-build-tools-image.outputs.image-tag }}-${{ matrix.version.debian }}
             DEBIAN_VERSION=${{ matrix.version.debian }}
           provenance: false
           push: true

.github/workflows/build_and_test_fully.yml

@@ -1,151 +0,0 @@
-name: Build and Test Fully
-
-on:
-  schedule:
-    # * is a special character in YAML so you have to quote this string
-    #          ┌───────────── minute (0 - 59)
-    #          │ ┌───────────── hour (0 - 23)
-    #          │ │ ┌───────────── day of the month (1 - 31)
-    #          │ │ │ ┌───────────── month (1 - 12 or JAN-DEC)
-    #          │ │ │ │ ┌───────────── day of the week (0 - 6 or SUN-SAT)
-    - cron:   '0 3 * * *' # run once a day, timezone is utc
-  workflow_dispatch:
-
-defaults:
-  run:
-    shell: bash -euxo pipefail {0}
-
-concurrency:
-  # Allow only one workflow per any non-`main` branch.
-  group: ${{ github.workflow }}-${{ github.ref_name }}-${{ github.ref_name == 'main' && github.sha || 'anysha' }}
-  cancel-in-progress: true
-
-env:
-  RUST_BACKTRACE: 1
-  COPT: '-Werror'
-
-jobs:
-  tag:
-    runs-on: [ self-hosted, small ]
-    container: ${{ vars.NEON_DEV_AWS_ACCOUNT_ID }}.dkr.ecr.${{ vars.AWS_ECR_REGION }}.amazonaws.com/base:pinned
-    outputs:
-      build-tag: ${{steps.build-tag.outputs.tag}}
-
-    steps:
-      # Need `fetch-depth: 0` to count the number of commits in the branch
-      - name: Harden the runner (Audit all outbound calls)
-        uses: step-security/harden-runner@4d991eb9b905ef189e4c376166672c3f2f230481 # v2.11.0
-        with:
-          egress-policy: audit
-
-      - uses: actions/checkout@11bd71901bbe5b1630ceea73d27597364c9af683 # v4.2.2
-        with:
-          fetch-depth: 0
-
-      - name: Get build tag
-        run: |
-          echo run:$GITHUB_RUN_ID
-          echo ref:$GITHUB_REF_NAME
-          echo rev:$(git rev-list --count HEAD)
-          if [[ "$GITHUB_REF_NAME" == "main" ]]; then
-            echo "tag=$(git rev-list --count HEAD)" >> $GITHUB_OUTPUT
-          elif [[ "$GITHUB_REF_NAME" == "release" ]]; then
-            echo "tag=release-$(git rev-list --count HEAD)" >> $GITHUB_OUTPUT
-          elif [[ "$GITHUB_REF_NAME" == "release-proxy" ]]; then
-            echo "tag=release-proxy-$(git rev-list --count HEAD)" >> $GITHUB_OUTPUT
-          elif [[ "$GITHUB_REF_NAME" == "release-compute" ]]; then
-            echo "tag=release-compute-$(git rev-list --count HEAD)" >> $GITHUB_OUTPUT
-          else
-            echo "GITHUB_REF_NAME (value '$GITHUB_REF_NAME') is not set to either 'main' or 'release', 'release-proxy', 'release-compute'"
-            echo "tag=$GITHUB_RUN_ID" >> $GITHUB_OUTPUT
-          fi
-        shell: bash
-        id: build-tag
-
-  build-build-tools-image:
-    uses: ./.github/workflows/build-build-tools-image.yml
-    secrets: inherit
-
-  build-and-test-locally:
-    needs: [ tag, build-build-tools-image ]
-    strategy:
-      fail-fast: false
-      matrix:
-        arch: [ x64, arm64 ]
-        build-type: [ debug, release ]
-    uses: ./.github/workflows/_build-and-test-locally.yml
-    with:
-      arch: ${{ matrix.arch }}
-      build-tools-image: ${{ needs.build-build-tools-image.outputs.image }}-bookworm
-      build-tag: ${{ needs.tag.outputs.build-tag }}
-      build-type: ${{ matrix.build-type }}
-      rerun-failed: false
-      test-cfg: '[{"pg_version":"v14", "lfc_state": "with-lfc"},
-                  {"pg_version":"v15", "lfc_state": "with-lfc"},
-                  {"pg_version":"v16", "lfc_state": "with-lfc"},
-                  {"pg_version":"v17", "lfc_state": "with-lfc"},
-                  {"pg_version":"v14", "lfc_state": "without-lfc"},
-                  {"pg_version":"v15", "lfc_state": "without-lfc"},
-                  {"pg_version":"v16", "lfc_state": "without-lfc"},
-                  {"pg_version":"v17", "lfc_state": "withouts-lfc"}]'
-    secrets: inherit
-
-
-  create-test-report:
-    needs: [ build-and-test-locally, build-build-tools-image ]
-    if: ${{ !cancelled() }}
-    permissions:
-      id-token: write # aws-actions/configure-aws-credentials
-      statuses: write
-      contents: write
-      pull-requests: write
-    outputs:
-      report-url: ${{ steps.create-allure-report.outputs.report-url }}
-
-    runs-on: [ self-hosted, small ]
-    container:
-      image: ${{ needs.build-build-tools-image.outputs.image }}-bookworm
-      credentials:
-        username: ${{ github.actor }}
-        password: ${{ secrets.GITHUB_TOKEN }}
-      options: --init
-
-    steps:
-      - name: Harden the runner (Audit all outbound calls)
-        uses: step-security/harden-runner@4d991eb9b905ef189e4c376166672c3f2f230481 # v2.11.0
-        with:
-          egress-policy: audit
-
-      - uses: actions/checkout@11bd71901bbe5b1630ceea73d27597364c9af683 # v4.2.2
-
-      - name: Create Allure report
-        if: ${{ !cancelled() }}
-        id: create-allure-report
-        uses: ./.github/actions/allure-report-generate
-        with:
-          store-test-results-into-db: true
-          aws-oidc-role-arn: ${{ vars.DEV_AWS_OIDC_ROLE_ARN }}
-        env:
-          REGRESS_TEST_RESULT_CONNSTR_NEW: ${{ secrets.REGRESS_TEST_RESULT_CONNSTR_NEW }}
-
-      - uses: actions/github-script@60a0d83039c74a4aee543508d2ffcb1c3799cdea # v7.0.1
-        if: ${{ !cancelled() }}
-        with:
-          # Retry script for 5XX server errors: https://github.com/actions/github-script#retries
-          retries: 5
-          script: |
-            const report = {
-              reportUrl:     "${{ steps.create-allure-report.outputs.report-url }}",
-              reportJsonUrl: "${{ steps.create-allure-report.outputs.report-json-url }}",
-            }
-
-            const coverage = {}
-
-            const script = require("./scripts/comment-test-report.js")
-            await script({
-              github,
-              context,
-              fetch,
-              report,
-              coverage,
-            })

.github/workflows/build_and_test_with_sanitizers.yml

@@ -79,7 +79,6 @@ jobs:
       build-tools-image: ${{ needs.build-build-tools-image.outputs.image }}-bookworm
       build-tag: ${{ needs.tag.outputs.build-tag }}
       build-type: ${{ matrix.build-type }}
-      rerun-failed: false
       test-cfg: '[{"pg_version":"v17"}]'
       sanitizers: enabled
     secrets: inherit

.github/workflows/large_oltp_benchmark.yml

@@ -33,19 +33,11 @@ jobs:
       fail-fast: false # allow other variants to continue even if one fails
       matrix:
         include:
-          # test only read-only custom scripts in new branch without database maintenance
-          - target: new_branch
-            custom_scripts: select_any_webhook_with_skew.sql@300 select_recent_webhook.sql@397 select_prefetch_webhook.sql@3
-            test_maintenance: false
-          # test all custom scripts in new branch with database maintenance
           - target: new_branch
             custom_scripts: insert_webhooks.sql@200 select_any_webhook_with_skew.sql@300 select_recent_webhook.sql@397 select_prefetch_webhook.sql@3 IUD_one_transaction.sql@100
-            test_maintenance: true
-          # test all custom scripts in reuse branch with database maintenance
           - target: reuse_branch
             custom_scripts: insert_webhooks.sql@200 select_any_webhook_with_skew.sql@300 select_recent_webhook.sql@397 select_prefetch_webhook.sql@3 IUD_one_transaction.sql@100
-            test_maintenance: true
-      max-parallel: 1 # we want to run each benchmark sequentially to not have noisy neighbors on shared storage (PS, SK)
+      max-parallel: 1 # we want to run each stripe size sequentially to be able to compare the results
     permissions:
       contents: write
       statuses: write
@@ -153,7 +145,6 @@ jobs:
         PERF_TEST_RESULT_CONNSTR: "${{ secrets.PERF_TEST_RESULT_CONNSTR }}"
 
     - name: Benchmark database maintenance
-      if: ${{ matrix.test_maintenance == 'true' }}
       uses: ./.github/actions/run-python-test-set
       with:
         build_type: ${{ env.BUILD_TYPE }}

.github/workflows/large_oltp_growth.yml

@@ -1,175 +0,0 @@
-name: large oltp growth
-# workflow to grow the reuse branch of large oltp benchmark continuously (about 16 GB per run)
-
-on:
-  # uncomment to run on push for debugging your PR
-  # push:
-  #  branches: [ bodobolero/increase_large_oltp_workload ]
-
-  schedule:
-    # * is a special character in YAML so you have to quote this string
-    #        ┌───────────── minute (0 - 59)
-    #        │ ┌───────────── hour (0 - 23)
-    #        │ │  ┌───────────── day of the month (1 - 31)
-    #        │ │  │ ┌───────────── month (1 - 12 or JAN-DEC)
-    #        │ │  │ │ ┌───────────── day of the week (0 - 6 or SUN-SAT)
-    - cron: '0 6 * * *'   # 06:00 UTC
-    - cron: '0 8 * * *'   # 08:00 UTC
-    - cron: '0 10 * * *'  # 10:00 UTC
-    - cron: '0 12 * * *'  # 12:00 UTC
-    - cron: '0 14 * * *'  # 14:00 UTC
-    - cron: '0 16 * * *'  # 16:00 UTC
-  workflow_dispatch: # adds ability to run this manually
-
-defaults:
-  run:
-    shell: bash -euxo pipefail {0}
-
-concurrency:
-  # Allow only one workflow globally because we need dedicated resources which only exist once
-  group: large-oltp-growth
-  cancel-in-progress: true
-
-permissions:
-  contents: read
-
-jobs:
-  oltp:
-    strategy:
-      fail-fast: false # allow other variants to continue even if one fails
-      matrix:
-        include:
-          # for now only grow the reuse branch, not the other branches.
-          - target: reuse_branch
-            custom_scripts:
-            - grow_action_blocks.sql
-            - grow_action_kwargs.sql
-            - grow_device_fingerprint_event.sql
-            - grow_edges.sql
-            - grow_hotel_rate_mapping.sql
-            - grow_ocr_pipeline_results_version.sql
-            - grow_priceline_raw_response.sql
-            - grow_relabled_transactions.sql
-            - grow_state_values.sql
-            - grow_values.sql
-            - grow_vertices.sql
-            - update_accounting_coding_body_tracking_category_selection.sql
-            - update_action_blocks.sql
-            - update_action_kwargs.sql
-            - update_denormalized_approval_workflow.sql
-            - update_device_fingerprint_event.sql
-            - update_edges.sql
-            - update_heron_transaction_enriched_log.sql
-            - update_heron_transaction_enrichment_requests.sql
-            - update_hotel_rate_mapping.sql
-            - update_incoming_webhooks.sql
-            - update_manual_transaction.sql
-            - update_ml_receipt_matching_log.sql
-            - update_ocr_pipeine_results_version.sql
-            - update_orc_pipeline_step_results.sql
-            - update_orc_pipeline_step_results_version.sql
-            - update_priceline_raw_response.sql
-            - update_quickbooks_transactions.sql
-            - update_raw_finicity_transaction.sql
-            - update_relabeled_transactions.sql
-            - update_state_values.sql
-            - update_stripe_authorization_event_log.sql
-            - update_transaction.sql
-            - update_values.sql
-            - update_vertices.sql
-      max-parallel: 1 # we want to run each growth workload sequentially (for now there is just one)
-    permissions:
-      contents: write
-      statuses: write
-      id-token: write # aws-actions/configure-aws-credentials
-    env:
-      TEST_PG_BENCH_DURATIONS_MATRIX: "1h"
-      TEST_PGBENCH_CUSTOM_SCRIPTS: ${{ join(matrix.custom_scripts, ' ') }}
-      POSTGRES_DISTRIB_DIR: /tmp/neon/pg_install
-      PG_VERSION: 16 # pre-determined by pre-determined project
-      TEST_OUTPUT: /tmp/test_output
-      BUILD_TYPE: remote
-      PLATFORM: ${{ matrix.target }}
-
-    runs-on: [ self-hosted, us-east-2, x64 ]
-    container:
-      image: ghcr.io/neondatabase/build-tools:pinned-bookworm
-      credentials:
-        username: ${{ github.actor }}
-        password: ${{ secrets.GITHUB_TOKEN }}
-      options: --init
-
-    steps:
-    - name: Harden the runner (Audit all outbound calls)
-      uses: step-security/harden-runner@4d991eb9b905ef189e4c376166672c3f2f230481 # v2.11.0
-      with:
-        egress-policy: audit
-
-    - uses: actions/checkout@11bd71901bbe5b1630ceea73d27597364c9af683 # v4.2.2
-
-    - name: Configure AWS credentials # necessary to download artefacts
-      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 # 5 hours is currently max associated with IAM role
-
-    - name: Download Neon artifact
-      uses: ./.github/actions/download
-      with:
-        name: neon-${{ runner.os }}-${{ runner.arch }}-release-artifact
-        path: /tmp/neon/
-        prefix: latest
-        aws-oidc-role-arn: ${{ vars.DEV_AWS_OIDC_ROLE_ARN }}
-
-    - name: Set up Connection String
-      id: set-up-connstr
-      run: |
-        case "${{ matrix.target }}" in
-          reuse_branch)
-          CONNSTR=${{ secrets.BENCHMARK_LARGE_OLTP_REUSE_CONNSTR }}
-          ;;
-          *)
-          echo >&2 "Unknown target=${{ matrix.target }}"
-          exit 1
-          ;;
-        esac
-
-        CONNSTR_WITHOUT_POOLER="${CONNSTR//-pooler/}"
-
-        echo "connstr=${CONNSTR}" >> $GITHUB_OUTPUT
-        echo "connstr_without_pooler=${CONNSTR_WITHOUT_POOLER}" >> $GITHUB_OUTPUT
-
-    - name: pgbench with custom-scripts
-      uses: ./.github/actions/run-python-test-set
-      with:
-        build_type: ${{ env.BUILD_TYPE }}
-        test_selection: performance
-        run_in_parallel: false
-        save_perf_report: true
-        extra_params: -m remote_cluster --timeout 7200 -k test_perf_oltp_large_tenant_growth
-        pg_version: ${{ env.PG_VERSION }}
-        aws-oidc-role-arn: ${{ vars.DEV_AWS_OIDC_ROLE_ARN }}
-      env:
-        BENCHMARK_CONNSTR: ${{ steps.set-up-connstr.outputs.connstr }}
-        VIP_VAP_ACCESS_TOKEN: "${{ secrets.VIP_VAP_ACCESS_TOKEN }}"
-        PERF_TEST_RESULT_CONNSTR: "${{ secrets.PERF_TEST_RESULT_CONNSTR }}"
-
-    - name: 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: Post to a Slack channel
-      if: ${{ github.event.schedule && failure() }}
-      uses: slackapi/slack-github-action@fcfb566f8b0aab22203f066d80ca1d7e4b5d05b3 # v1.27.1
-      with:
-        channel-id: "C06KHQVQ7U3" # on-call-qa-staging-stream
-        slack-message: |
-          Periodic large oltp tenant growth increase: ${{ job.status }}
-          <${{ github.server_url }}/${{ github.repository }}/actions/runs/${{ github.run_id }}|GitHub Run>
-          <${{ steps.create-allure-report.outputs.report-url }}|Allure report>
-      env:
-        SLACK_BOT_TOKEN: ${{ secrets.SLACK_BOT_TOKEN }}

.github/workflows/proxy-benchmark.yml

@@ -1,83 +0,0 @@
-name: Periodic proxy performance test on unit-perf hetzner runner
-
-on:
-  push: # TODO: remove after testing
-    branches:
-      - test-proxy-bench # Runs on pushes to branches starting with test-proxy-bench
-  # schedule:
-    # * is a special character in YAML so you have to quote this string
-    #        ┌───────────── minute (0 - 59)
-    #        │ ┌───────────── hour (0 - 23)
-    #        │ │ ┌───────────── day of the month (1 - 31)
-    #        │ │ │ ┌───────────── month (1 - 12 or JAN-DEC)
-    #        │ │ │ │ ┌───────────── day of the week (0 - 6 or SUN-SAT)
-    # - cron: '0 5 * * *' # Runs at 5 UTC once a day
-  workflow_dispatch: # adds an ability to run this manually
-
-defaults:
-  run:
-    shell: bash -euo pipefail {0}
-
-concurrency:
-  group: ${{ github.workflow }}
-  cancel-in-progress: false
-
-permissions:
-  contents: read
-
-jobs:
-  run_periodic_proxybench_test:
-    permissions:
-      id-token: write # aws-actions/configure-aws-credentials
-      statuses: write
-      contents: write
-      pull-requests: write
-    runs-on: [self-hosted, unit-perf]
-    timeout-minutes: 60  # 1h timeout
-    container:
-      image: ghcr.io/neondatabase/build-tools:pinned-bookworm
-      credentials:
-        username: ${{ github.actor }}
-        password: ${{ secrets.GITHUB_TOKEN }}
-      options: --init
-    steps:
-    - name: Checkout proxy-bench Repo
-      uses: actions/checkout@v4
-      with:
-        repository: neondatabase/proxy-bench
-        path: proxy-bench
-
-    - name: Set up the environment which depends on $RUNNER_TEMP on nvme drive
-      id: set-env
-      shell: bash -euxo pipefail {0}
-      run: |
-        PROXY_BENCH_PATH=$(realpath ./proxy-bench)
-        {
-          echo "PROXY_BENCH_PATH=$PROXY_BENCH_PATH"
-          echo "NEON_DIR=${RUNNER_TEMP}/neon"
-          echo "TEST_OUTPUT=${PROXY_BENCH_PATH}/test_output"
-          echo ""
-        } >> "$GITHUB_ENV"
-
-    - name: Run proxy-bench
-      run: ./${PROXY_BENCH_PATH}/run.sh
-
-    - name: Ingest Bench Results # neon repo script
-      if: success()
-      run: |
-        mkdir -p $TEST_OUTPUT
-        python $NEON_DIR/scripts/proxy_bench_results_ingest.py --out $TEST_OUTPUT
-
-    - name: Push Metrics to Proxy perf database
-      if: success()
-      env:
-        PERF_TEST_RESULT_CONNSTR: "${{ secrets.PROXY_TEST_RESULT_CONNSTR }}"
-        REPORT_FROM: $TEST_OUTPUT
-      run: $NEON_DIR/scripts/generate_and_push_perf_report.sh
-
-    - name: Docker cleanup
-      run: docker compose down
-
-    - name: Notify Failure
-      if: failure()
-      run: echo "Proxy bench job failed" && exit 1

.gitignore

@@ -1,5 +1,4 @@
 /artifact_cache
-/build
 /pg_install
 /target
 /tmp_check

Cargo.lock

@@ -903,6 +903,12 @@ version = "0.13.1"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "9e1b586273c5702936fe7b7d6896644d8be71e6314cfe09d3167c95f712589e8"
 
+[[package]]
+name = "base64"
+version = "0.20.0"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "0ea22880d78093b0cbe17c89f64a7d457941e65759157ec6cb31a31d652b05e5"
+
 [[package]]
 name = "base64"
 version = "0.21.7"
@@ -1294,7 +1300,7 @@ dependencies = [
  "aws-smithy-types",
  "axum",
  "axum-extra",
- "base64 0.22.1",
+ "base64 0.13.1",
  "bytes",
  "camino",
  "cfg-if",
@@ -1420,7 +1426,7 @@ name = "control_plane"
 version = "0.1.0"
 dependencies = [
  "anyhow",
- "base64 0.22.1",
+ "base64 0.13.1",
  "camino",
  "clap",
  "comfy-table",
@@ -4255,7 +4261,6 @@ dependencies = [
  "tokio-util",
  "tonic 0.13.1",
  "tracing",
- "url",
  "utils",
  "workspace_hack",
 ]
@@ -4335,7 +4340,6 @@ dependencies = [
  "postgres_backend",
  "postgres_connection",
  "postgres_ffi",
- "postgres_ffi_types",
  "postgres_initdb",
  "posthog_client_lite",
  "pprof",
@@ -4405,7 +4409,7 @@ dependencies = [
  "nix 0.30.1",
  "once_cell",
  "postgres_backend",
- "postgres_ffi_types",
+ "postgres_ffi",
  "rand 0.8.5",
  "remote_storage",
  "reqwest",
@@ -4467,16 +4471,11 @@ dependencies = [
 name = "pageserver_page_api"
 version = "0.1.0"
 dependencies = [
- "anyhow",
  "bytes",
- "futures",
  "pageserver_api",
  "postgres_ffi",
  "prost 0.13.5",
- "strum",
- "strum_macros",
  "thiserror 1.0.69",
- "tokio",
  "tonic 0.13.1",
  "tonic-build",
  "utils",
@@ -4819,7 +4818,7 @@ dependencies = [
 name = "postgres-protocol2"
 version = "0.1.0"
 dependencies = [
- "base64 0.22.1",
+ "base64 0.20.0",
  "byteorder",
  "bytes",
  "fallible-iterator",
@@ -4896,7 +4895,6 @@ dependencies = [
  "memoffset 0.9.0",
  "once_cell",
  "postgres",
- "postgres_ffi_types",
  "pprof",
  "regex",
  "serde",
@@ -4905,14 +4903,6 @@ dependencies = [
  "utils",
 ]
 
-[[package]]
-name = "postgres_ffi_types"
-version = "0.1.0"
-dependencies = [
- "thiserror 1.0.69",
- "workspace_hack",
-]
-
 [[package]]
 name = "postgres_initdb"
 version = "0.1.0"
@@ -5200,7 +5190,7 @@ dependencies = [
  "aws-config",
  "aws-sdk-iam",
  "aws-sigv4",
- "base64 0.22.1",
+ "base64 0.13.1",
  "bstr",
  "bytes",
  "camino",
@@ -6504,17 +6494,15 @@ dependencies = [
 
 [[package]]
 name = "serde_with"
-version = "3.12.0"
+version = "2.3.3"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "d6b6f7f2fcb69f747921f79f3926bd1e203fce4fef62c268dd3abfb6d86029aa"
+checksum = "07ff71d2c147a7b57362cead5e22f772cd52f6ab31cfcd9edcd7f6aeb2a0afbe"
 dependencies = [
- "base64 0.22.1",
+ "base64 0.13.1",
  "chrono",
  "hex",
  "indexmap 1.9.3",
- "indexmap 2.9.0",
  "serde",
- "serde_derive",
  "serde_json",
  "serde_with_macros",
  "time",
@@ -6522,9 +6510,9 @@ dependencies = [
 
 [[package]]
 name = "serde_with_macros"
-version = "3.12.0"
+version = "2.3.3"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "8d00caa5193a3c8362ac2b73be6b9e768aa5a4b2f721d8f4b339600c3cb51f8e"
+checksum = "881b6f881b17d13214e5d494c939ebab463d01264ce1811e9d4ac3a882e7695f"
 dependencies = [
  "darling",
  "proc-macro2",
@@ -7574,7 +7562,6 @@ dependencies = [
  "axum",
  "base64 0.22.1",
  "bytes",
- "flate2",
  "h2 0.4.4",
  "http 1.1.0",
  "http-body 1.0.0",
@@ -7594,7 +7581,6 @@ dependencies = [
  "tower-layer",
  "tower-service",
  "tracing",
- "zstd",
 ]
 
 [[package]]
@@ -8174,7 +8160,6 @@ dependencies = [
  "futures",
  "pageserver_api",
  "postgres_ffi",
- "postgres_ffi_types",
  "pprof",
  "prost 0.13.5",
  "remote_storage",
@@ -8598,6 +8583,7 @@ dependencies = [
  "anyhow",
  "axum",
  "axum-core",
+ "base64 0.13.1",
  "base64 0.21.7",
  "base64ct",
  "bytes",

Cargo.toml

@@ -22,7 +22,6 @@ members = [
     "libs/http-utils",
     "libs/pageserver_api",
     "libs/postgres_ffi",
-    "libs/postgres_ffi_types",
     "libs/safekeeper_api",
     "libs/desim",
     "libs/neon-shmem",
@@ -73,7 +72,7 @@ aws-sigv4 = { version = "1.2", features = ["sign-http"] }
 aws-types = "1.3"
 axum = { version = "0.8.1", features = ["ws"] }
 axum-extra = { version = "0.10.0", features = ["typed-header", "query"] }
-base64 = "0.22"
+base64 = "0.13.0"
 bincode = "1.3"
 bindgen = "0.71"
 bit_field = "0.10.2"
@@ -172,7 +171,7 @@ sentry = { version = "0.37", default-features = false, features = ["backtrace",
 serde = { version = "1.0", features = ["derive"] }
 serde_json = "1"
 serde_path_to_error = "0.1"
-serde_with = { version = "3", features = [ "base64" ] }
+serde_with = { version = "2.0", features = [ "base64" ] }
 serde_assert = "0.5.0"
 sha2 = "0.10.2"
 signal-hook = "0.3"
@@ -200,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", "gzip", "prost", "router", "server", "tls-ring", "tls-native-roots", "zstd"] }
+tonic = { version = "0.13.1", default-features = false, features = ["channel", "codegen", "prost", "router", "server", "tls-ring", "tls-native-roots"] }
 tonic-reflection = { version = "0.13.1", features = ["server"] }
 tower = { version = "0.5.2", default-features = false }
 tower-http = { version = "0.6.2", features = ["auth", "request-id", "trace"] }
@@ -260,7 +259,6 @@ pageserver_page_api = { path = "./pageserver/page_api" }
 postgres_backend = { version = "0.1", path = "./libs/postgres_backend/" }
 postgres_connection = { version = "0.1", path = "./libs/postgres_connection/" }
 postgres_ffi = { version = "0.1", path = "./libs/postgres_ffi/" }
-postgres_ffi_types = { version = "0.1", path = "./libs/postgres_ffi_types/" }
 postgres_initdb = { path = "./libs/postgres_initdb" }
 posthog_client_lite = { version = "0.1", path = "./libs/posthog_client_lite" }
 pq_proto = { version = "0.1", path = "./libs/pq_proto/" }

Dockerfile

@@ -5,6 +5,8 @@
 ARG REPOSITORY=ghcr.io/neondatabase
 ARG IMAGE=build-tools
 ARG TAG=pinned
+ARG DEFAULT_PG_VERSION=17
+ARG STABLE_PG_VERSION=16
 ARG DEBIAN_VERSION=bookworm
 ARG DEBIAN_FLAVOR=${DEBIAN_VERSION}-slim
 
@@ -45,6 +47,7 @@ COPY --chown=nonroot scripts/ninstall.sh scripts/ninstall.sh
 ENV BUILD_TYPE=release
 RUN set -e \
     && mold -run make -j $(nproc) -s neon-pg-ext \
+    && rm -rf pg_install/build \
     && tar -C pg_install -czf /home/nonroot/postgres_install.tar.gz .
 
 # Prepare cargo-chef recipe
@@ -60,11 +63,14 @@ FROM $REPOSITORY/$IMAGE:$TAG AS build
 WORKDIR /home/nonroot
 ARG GIT_VERSION=local
 ARG BUILD_TAG
+ARG STABLE_PG_VERSION
 
 COPY --from=pg-build /home/nonroot/pg_install/v14/include/postgresql/server pg_install/v14/include/postgresql/server
 COPY --from=pg-build /home/nonroot/pg_install/v15/include/postgresql/server pg_install/v15/include/postgresql/server
 COPY --from=pg-build /home/nonroot/pg_install/v16/include/postgresql/server pg_install/v16/include/postgresql/server
 COPY --from=pg-build /home/nonroot/pg_install/v17/include/postgresql/server pg_install/v17/include/postgresql/server
+COPY --from=pg-build /home/nonroot/pg_install/v16/lib                       pg_install/v16/lib
+COPY --from=pg-build /home/nonroot/pg_install/v17/lib                       pg_install/v17/lib
 COPY --from=plan     /home/nonroot/recipe.json                              recipe.json
 
 ARG ADDITIONAL_RUSTFLAGS=""
@@ -91,6 +97,7 @@ RUN set -e \
 # Build final image
 #
 FROM $BASE_IMAGE_SHA
+ARG DEFAULT_PG_VERSION
 WORKDIR /data
 
 RUN set -e \
@@ -100,6 +107,8 @@ RUN set -e \
         libreadline-dev \
         libseccomp-dev \
         ca-certificates \
+	# System postgres for use with client libraries (e.g. in storage controller)
+        postgresql-15 \
         openssl \
         unzip \
         curl \

Makefile

@@ -1,18 +1,8 @@
 ROOT_PROJECT_DIR := $(dir $(abspath $(lastword $(MAKEFILE_LIST))))
 
-# Where to install Postgres, default is ./pg_install, maybe useful for package
-# managers.
+# Where to install Postgres, default is ./pg_install, maybe useful for package managers
 POSTGRES_INSTALL_DIR ?= $(ROOT_PROJECT_DIR)/pg_install/
 
-# CARGO_BUILD_FLAGS: Extra flags to pass to `cargo build`. `--locked`
-# and `--features testing` are popular examples.
-#
-# CARGO_PROFILE: You can also set to override the cargo profile to
-# use. By default, it is derived from BUILD_TYPE.
-
-# All intermediate build artifacts are stored here.
-BUILD_DIR := build
-
 ICU_PREFIX_DIR := /usr/local/icu
 
 #
@@ -26,12 +16,12 @@ ifeq ($(BUILD_TYPE),release)
 	PG_CONFIGURE_OPTS = --enable-debug --with-openssl
 	PG_CFLAGS += -O2 -g3 $(CFLAGS)
 	PG_LDFLAGS = $(LDFLAGS)
-	CARGO_PROFILE ?= --profile=release
+	# Unfortunately, `--profile=...` is a nightly feature
+	CARGO_BUILD_FLAGS += --release
 else ifeq ($(BUILD_TYPE),debug)
 	PG_CONFIGURE_OPTS = --enable-debug --with-openssl --enable-cassert --enable-depend
 	PG_CFLAGS += -O0 -g3 $(CFLAGS)
 	PG_LDFLAGS = $(LDFLAGS)
-	CARGO_PROFILE ?= --profile=dev
 else
 	$(error Bad build type '$(BUILD_TYPE)', see Makefile for options)
 endif
@@ -103,7 +93,7 @@ all: neon postgres neon-pg-ext
 .PHONY: neon
 neon: postgres-headers walproposer-lib cargo-target-dir
 	+@echo "Compiling Neon"
-	$(CARGO_CMD_PREFIX) cargo build $(CARGO_BUILD_FLAGS) $(CARGO_PROFILE)
+	$(CARGO_CMD_PREFIX) cargo build $(CARGO_BUILD_FLAGS)
 .PHONY: cargo-target-dir
 cargo-target-dir:
 	# https://github.com/rust-lang/cargo/issues/14281
@@ -114,20 +104,21 @@ cargo-target-dir:
 # Some rules are duplicated for Postgres v14 and 15. We may want to refactor
 # to avoid the duplication in the future, but it's tolerable for now.
 #
-$(BUILD_DIR)/%/config.status:
-	mkdir -p $(BUILD_DIR)
-	test -e $(BUILD_DIR)/CACHEDIR.TAG || echo "$(CACHEDIR_TAG_CONTENTS)" > $(BUILD_DIR)/CACHEDIR.TAG
+$(POSTGRES_INSTALL_DIR)/build/%/config.status:
+
+	mkdir -p $(POSTGRES_INSTALL_DIR)
+	test -e $(POSTGRES_INSTALL_DIR)/CACHEDIR.TAG || echo "$(CACHEDIR_TAG_CONTENTS)" > $(POSTGRES_INSTALL_DIR)/CACHEDIR.TAG
 
 	+@echo "Configuring Postgres $* build"
 	@test -s $(ROOT_PROJECT_DIR)/vendor/postgres-$*/configure || { \
 		echo "\nPostgres submodule not found in $(ROOT_PROJECT_DIR)/vendor/postgres-$*/, execute "; \
 		echo "'git submodule update --init --recursive --depth 2 --progress .' in project root.\n"; \
 		exit 1; }
-	mkdir -p $(BUILD_DIR)/$*
+	mkdir -p $(POSTGRES_INSTALL_DIR)/build/$*
 
 	VERSION=$*; \
 	EXTRA_VERSION=$$(cd $(ROOT_PROJECT_DIR)/vendor/postgres-$$VERSION && git rev-parse HEAD); \
-	(cd $(BUILD_DIR)/$$VERSION && \
+	(cd $(POSTGRES_INSTALL_DIR)/build/$$VERSION && \
 	env PATH="$(EXTRA_PATH_OVERRIDES):$$PATH" $(ROOT_PROJECT_DIR)/vendor/postgres-$$VERSION/configure \
 		CFLAGS='$(PG_CFLAGS)' LDFLAGS='$(PG_LDFLAGS)' \
 		$(PG_CONFIGURE_OPTS) --with-extra-version=" ($$EXTRA_VERSION)" \
@@ -139,54 +130,96 @@ $(BUILD_DIR)/%/config.status:
 # the "build-all-versions" entry points) where direct mention of PostgreSQL
 # versions is used.
 .PHONY: postgres-configure-v17
-postgres-configure-v17: $(BUILD_DIR)/v17/config.status
+postgres-configure-v17: $(POSTGRES_INSTALL_DIR)/build/v17/config.status
 .PHONY: postgres-configure-v16
-postgres-configure-v16: $(BUILD_DIR)/v16/config.status
+postgres-configure-v16: $(POSTGRES_INSTALL_DIR)/build/v16/config.status
 .PHONY: postgres-configure-v15
-postgres-configure-v15: $(BUILD_DIR)/v15/config.status
+postgres-configure-v15: $(POSTGRES_INSTALL_DIR)/build/v15/config.status
 .PHONY: postgres-configure-v14
-postgres-configure-v14: $(BUILD_DIR)/v14/config.status
+postgres-configure-v14: $(POSTGRES_INSTALL_DIR)/build/v14/config.status
 
 # Install the PostgreSQL header files into $(POSTGRES_INSTALL_DIR)/<version>/include
 .PHONY: postgres-headers-%
 postgres-headers-%: postgres-configure-%
 	+@echo "Installing PostgreSQL $* headers"
-	$(MAKE) -C $(BUILD_DIR)/$*/src/include MAKELEVEL=0 install
+	$(MAKE) -C $(POSTGRES_INSTALL_DIR)/build/$*/src/include MAKELEVEL=0 install
 
 # Compile and install PostgreSQL
 .PHONY: postgres-%
 postgres-%: postgres-configure-% \
 		  postgres-headers-% # to prevent `make install` conflicts with neon's `postgres-headers`
 	+@echo "Compiling PostgreSQL $*"
-	$(MAKE) -C $(BUILD_DIR)/$* MAKELEVEL=0 install
+	$(MAKE) -C $(POSTGRES_INSTALL_DIR)/build/$* MAKELEVEL=0 install
 	+@echo "Compiling libpq $*"
-	$(MAKE) -C $(BUILD_DIR)/$*/src/interfaces/libpq install
+	$(MAKE) -C $(POSTGRES_INSTALL_DIR)/build/$*/src/interfaces/libpq install
 	+@echo "Compiling pg_prewarm $*"
-	$(MAKE) -C $(BUILD_DIR)/$*/contrib/pg_prewarm install
+	$(MAKE) -C $(POSTGRES_INSTALL_DIR)/build/$*/contrib/pg_prewarm install
 	+@echo "Compiling pg_buffercache $*"
-	$(MAKE) -C $(BUILD_DIR)/$*/contrib/pg_buffercache install
+	$(MAKE) -C $(POSTGRES_INSTALL_DIR)/build/$*/contrib/pg_buffercache install
 	+@echo "Compiling pg_visibility $*"
-	$(MAKE) -C $(BUILD_DIR)/$*/contrib/pg_visibility install
+	$(MAKE) -C $(POSTGRES_INSTALL_DIR)/build/$*/contrib/pg_visibility install
 	+@echo "Compiling pageinspect $*"
-	$(MAKE) -C $(BUILD_DIR)/$*/contrib/pageinspect install
+	$(MAKE) -C $(POSTGRES_INSTALL_DIR)/build/$*/contrib/pageinspect install
 	+@echo "Compiling pg_trgm $*"
-	$(MAKE) -C $(BUILD_DIR)/$*/contrib/pg_trgm install
+	$(MAKE) -C $(POSTGRES_INSTALL_DIR)/build/$*/contrib/pg_trgm install
 	+@echo "Compiling amcheck $*"
-	$(MAKE) -C $(BUILD_DIR)/$*/contrib/amcheck install
+	$(MAKE) -C $(POSTGRES_INSTALL_DIR)/build/$*/contrib/amcheck install
 	+@echo "Compiling test_decoding $*"
-	$(MAKE) -C $(BUILD_DIR)/$*/contrib/test_decoding install
+	$(MAKE) -C $(POSTGRES_INSTALL_DIR)/build/$*/contrib/test_decoding install
+
+.PHONY: postgres-clean-%
+postgres-clean-%:
+	$(MAKE) -C $(POSTGRES_INSTALL_DIR)/build/$* MAKELEVEL=0 clean
+	$(MAKE) -C $(POSTGRES_INSTALL_DIR)/build/$*/contrib/pg_buffercache clean
+	$(MAKE) -C $(POSTGRES_INSTALL_DIR)/build/$*/contrib/pageinspect clean
+	$(MAKE) -C $(POSTGRES_INSTALL_DIR)/build/$*/src/interfaces/libpq clean
 
 .PHONY: postgres-check-%
 postgres-check-%: postgres-%
-	$(MAKE) -C $(BUILD_DIR)/$* MAKELEVEL=0 check
+	$(MAKE) -C $(POSTGRES_INSTALL_DIR)/build/$* MAKELEVEL=0 check
 
 .PHONY: neon-pg-ext-%
 neon-pg-ext-%: postgres-%
-	+@echo "Compiling neon-specific Postgres extensions for $*"
-	mkdir -p $(BUILD_DIR)/pgxn-$*
+	+@echo "Compiling neon $*"
+	mkdir -p $(POSTGRES_INSTALL_DIR)/build/neon-$*
 	$(MAKE) PG_CONFIG=$(POSTGRES_INSTALL_DIR)/$*/bin/pg_config COPT='$(COPT)' \
-		-C $(BUILD_DIR)/pgxn-$*\
-		-f $(ROOT_PROJECT_DIR)/pgxn/Makefile  install
+		-C $(POSTGRES_INSTALL_DIR)/build/neon-$* \
+		-f $(ROOT_PROJECT_DIR)/pgxn/neon/Makefile install
+	+@echo "Compiling neon_walredo $*"
+	mkdir -p $(POSTGRES_INSTALL_DIR)/build/neon-walredo-$*
+	$(MAKE) PG_CONFIG=$(POSTGRES_INSTALL_DIR)/$*/bin/pg_config COPT='$(COPT)' \
+		-C $(POSTGRES_INSTALL_DIR)/build/neon-walredo-$* \
+		-f $(ROOT_PROJECT_DIR)/pgxn/neon_walredo/Makefile install
+	+@echo "Compiling neon_rmgr $*"
+	mkdir -p $(POSTGRES_INSTALL_DIR)/build/neon-rmgr-$*
+	$(MAKE) PG_CONFIG=$(POSTGRES_INSTALL_DIR)/$*/bin/pg_config COPT='$(COPT)' \
+		-C $(POSTGRES_INSTALL_DIR)/build/neon-rmgr-$* \
+		-f $(ROOT_PROJECT_DIR)/pgxn/neon_rmgr/Makefile install
+	+@echo "Compiling neon_test_utils $*"
+	mkdir -p $(POSTGRES_INSTALL_DIR)/build/neon-test-utils-$*
+	$(MAKE) PG_CONFIG=$(POSTGRES_INSTALL_DIR)/$*/bin/pg_config COPT='$(COPT)' \
+		-C $(POSTGRES_INSTALL_DIR)/build/neon-test-utils-$* \
+		-f $(ROOT_PROJECT_DIR)/pgxn/neon_test_utils/Makefile install
+	+@echo "Compiling neon_utils $*"
+	mkdir -p $(POSTGRES_INSTALL_DIR)/build/neon-utils-$*
+	$(MAKE) PG_CONFIG=$(POSTGRES_INSTALL_DIR)/$*/bin/pg_config COPT='$(COPT)' \
+		-C $(POSTGRES_INSTALL_DIR)/build/neon-utils-$* \
+		-f $(ROOT_PROJECT_DIR)/pgxn/neon_utils/Makefile install
+
+.PHONY: neon-pg-clean-ext-%
+neon-pg-clean-ext-%:
+	$(MAKE) PG_CONFIG=$(POSTGRES_INSTALL_DIR)/$*/bin/pg_config \
+	-C $(POSTGRES_INSTALL_DIR)/build/neon-$* \
+	-f $(ROOT_PROJECT_DIR)/pgxn/neon/Makefile clean
+	$(MAKE) PG_CONFIG=$(POSTGRES_INSTALL_DIR)/$*/bin/pg_config \
+	-C $(POSTGRES_INSTALL_DIR)/build/neon-walredo-$* \
+	-f $(ROOT_PROJECT_DIR)/pgxn/neon_walredo/Makefile clean
+	$(MAKE) PG_CONFIG=$(POSTGRES_INSTALL_DIR)/$*/bin/pg_config \
+	-C $(POSTGRES_INSTALL_DIR)/build/neon-test-utils-$* \
+	-f $(ROOT_PROJECT_DIR)/pgxn/neon_test_utils/Makefile clean
+	$(MAKE) PG_CONFIG=$(POSTGRES_INSTALL_DIR)/$*/bin/pg_config \
+	-C $(POSTGRES_INSTALL_DIR)/build/neon-utils-$* \
+	-f $(ROOT_PROJECT_DIR)/pgxn/neon_utils/Makefile clean
 
 # Build walproposer as a static library. walproposer source code is located
 # in the pgxn/neon directory.
@@ -200,15 +233,15 @@ neon-pg-ext-%: postgres-%
 .PHONY: walproposer-lib
 walproposer-lib: neon-pg-ext-v17
 	+@echo "Compiling walproposer-lib"
-	mkdir -p $(BUILD_DIR)/walproposer-lib
+	mkdir -p $(POSTGRES_INSTALL_DIR)/build/walproposer-lib
 	$(MAKE) PG_CONFIG=$(POSTGRES_INSTALL_DIR)/v17/bin/pg_config COPT='$(COPT)' \
-		-C $(BUILD_DIR)/walproposer-lib \
+		-C $(POSTGRES_INSTALL_DIR)/build/walproposer-lib \
 		-f $(ROOT_PROJECT_DIR)/pgxn/neon/Makefile walproposer-lib
-	cp $(POSTGRES_INSTALL_DIR)/v17/lib/libpgport.a $(BUILD_DIR)/walproposer-lib
-	cp $(POSTGRES_INSTALL_DIR)/v17/lib/libpgcommon.a $(BUILD_DIR)/walproposer-lib
-	$(AR) d $(BUILD_DIR)/walproposer-lib/libpgport.a \
+	cp $(POSTGRES_INSTALL_DIR)/v17/lib/libpgport.a $(POSTGRES_INSTALL_DIR)/build/walproposer-lib
+	cp $(POSTGRES_INSTALL_DIR)/v17/lib/libpgcommon.a $(POSTGRES_INSTALL_DIR)/build/walproposer-lib
+	$(AR) d $(POSTGRES_INSTALL_DIR)/build/walproposer-lib/libpgport.a \
 		pg_strong_random.o
-	$(AR) d $(BUILD_DIR)/walproposer-lib/libpgcommon.a \
+	$(AR) d $(POSTGRES_INSTALL_DIR)/build/walproposer-lib/libpgcommon.a \
 		checksum_helper.o \
 		cryptohash_openssl.o \
 		hmac_openssl.o \
@@ -216,10 +249,16 @@ walproposer-lib: neon-pg-ext-v17
 		parse_manifest.o \
 		scram-common.o
 ifeq ($(UNAME_S),Linux)
-	$(AR) d $(BUILD_DIR)/walproposer-lib/libpgcommon.a \
+	$(AR) d $(POSTGRES_INSTALL_DIR)/build/walproposer-lib/libpgcommon.a \
 		pg_crc32c.o
 endif
 
+.PHONY: walproposer-lib-clean
+walproposer-lib-clean:
+	$(MAKE) PG_CONFIG=$(POSTGRES_INSTALL_DIR)/v17/bin/pg_config \
+		-C $(POSTGRES_INSTALL_DIR)/build/walproposer-lib \
+		-f $(ROOT_PROJECT_DIR)/pgxn/neon/Makefile clean
+
 .PHONY: neon-pg-ext
 neon-pg-ext: \
 	neon-pg-ext-v14 \
@@ -227,6 +266,13 @@ neon-pg-ext: \
 	neon-pg-ext-v16 \
 	neon-pg-ext-v17
 
+.PHONY: neon-pg-clean-ext
+neon-pg-clean-ext: \
+	neon-pg-clean-ext-v14 \
+	neon-pg-clean-ext-v15 \
+	neon-pg-clean-ext-v16 \
+	neon-pg-clean-ext-v17
+
 # shorthand to build all Postgres versions
 .PHONY: postgres
 postgres: \
@@ -242,6 +288,13 @@ postgres-headers: \
 	postgres-headers-v16 \
 	postgres-headers-v17
 
+.PHONY: postgres-clean
+postgres-clean: \
+	postgres-clean-v14 \
+	postgres-clean-v15 \
+	postgres-clean-v16 \
+	postgres-clean-v17
+
 .PHONY: postgres-check
 postgres-check: \
 	postgres-check-v14 \
@@ -249,6 +302,12 @@ postgres-check: \
 	postgres-check-v16 \
 	postgres-check-v17
 
+# This doesn't remove the effects of 'configure'.
+.PHONY: clean
+clean: postgres-clean neon-pg-clean-ext
+	$(MAKE) -C compute clean
+	$(CARGO_CMD_PREFIX) cargo clean
+
 # This removes everything
 .PHONY: distclean
 distclean:
@@ -261,7 +320,7 @@ fmt:
 
 postgres-%-pg-bsd-indent: postgres-%
 	+@echo "Compiling pg_bsd_indent"
-	$(MAKE) -C $(BUILD_DIR)/$*/src/tools/pg_bsd_indent/
+	$(MAKE) -C $(POSTGRES_INSTALL_DIR)/build/$*/src/tools/pg_bsd_indent/
 
 # Create typedef list for the core. Note that generally it should be combined with
 # buildfarm one to cover platform specific stuff.
@@ -280,7 +339,7 @@ postgres-%-pgindent: postgres-%-pg-bsd-indent postgres-%-typedefs.list
 	cat $(ROOT_PROJECT_DIR)/vendor/postgres-$*/src/tools/pgindent/typedefs.list |\
 		cat - postgres-$*-typedefs.list | sort | uniq > postgres-$*-typedefs-full.list
 	+@echo note: you might want to run it on selected files/dirs instead.
-	INDENT=$(BUILD_DIR)/$*/src/tools/pg_bsd_indent/pg_bsd_indent \
+	INDENT=$(POSTGRES_INSTALL_DIR)/build/$*/src/tools/pg_bsd_indent/pg_bsd_indent \
 		$(ROOT_PROJECT_DIR)/vendor/postgres-$*/src/tools/pgindent/pgindent --typedefs postgres-$*-typedefs-full.list \
 		$(ROOT_PROJECT_DIR)/vendor/postgres-$*/src/ \
 		--excludes $(ROOT_PROJECT_DIR)/vendor/postgres-$*/src/tools/pgindent/exclude_file_patterns
@@ -291,9 +350,9 @@ postgres-%-pgindent: postgres-%-pg-bsd-indent postgres-%-typedefs.list
 neon-pgindent: postgres-v17-pg-bsd-indent neon-pg-ext-v17
 	$(MAKE) PG_CONFIG=$(POSTGRES_INSTALL_DIR)/v17/bin/pg_config COPT='$(COPT)' \
 		FIND_TYPEDEF=$(ROOT_PROJECT_DIR)/vendor/postgres-v17/src/tools/find_typedef \
-		INDENT=$(BUILD_DIR)/v17/src/tools/pg_bsd_indent/pg_bsd_indent \
+		INDENT=$(POSTGRES_INSTALL_DIR)/build/v17/src/tools/pg_bsd_indent/pg_bsd_indent \
 		PGINDENT_SCRIPT=$(ROOT_PROJECT_DIR)/vendor/postgres-v17/src/tools/pgindent/pgindent \
-		-C $(BUILD_DIR)/neon-v17 \
+		-C $(POSTGRES_INSTALL_DIR)/build/neon-v17 \
 		-f $(ROOT_PROJECT_DIR)/pgxn/neon/Makefile pgindent
 
 

compute/.gitignore

@@ -3,6 +3,3 @@ etc/neon_collector.yml
 etc/neon_collector_autoscaling.yml
 etc/sql_exporter.yml
 etc/sql_exporter_autoscaling.yml
-
-# Node.js dependencies
-node_modules/

compute/Makefile

@@ -48,11 +48,3 @@ jsonnetfmt-test:
 .PHONY: jsonnetfmt-format
 jsonnetfmt-format:
 	jsonnetfmt --in-place $(jsonnet_files)
-
-.PHONY: manifest-schema-validation
-manifest-schema-validation: node_modules
-	node_modules/.bin/jsonschema validate -d https://json-schema.org/draft/2020-12/schema manifest.schema.json manifest.yaml
-
-node_modules: package.json
-	npm install
-	touch node_modules

compute/compute-node.Dockerfile

@@ -77,6 +77,9 @@
 # build_and_test.yml github workflow for how that's done.
 
 ARG PG_VERSION
+ARG REPOSITORY=ghcr.io/neondatabase
+ARG IMAGE=build-tools
+ARG TAG=pinned
 ARG BUILD_TAG
 ARG DEBIAN_VERSION=bookworm
 ARG DEBIAN_FLAVOR=${DEBIAN_VERSION}-slim
@@ -146,11 +149,8 @@ RUN case $DEBIAN_VERSION in \
     ninja-build git autoconf automake libtool build-essential bison flex libreadline-dev \
     zlib1g-dev libxml2-dev libcurl4-openssl-dev libossp-uuid-dev wget ca-certificates pkg-config libssl-dev \
     libicu-dev libxslt1-dev liblz4-dev libzstd-dev zstd curl unzip g++ \
-    libclang-dev \
-    jsonnet \
     $VERSION_INSTALLS \
-    && apt clean && rm -rf /var/lib/apt/lists/* && \
-    useradd -ms /bin/bash nonroot -b /home
+    && apt clean && rm -rf /var/lib/apt/lists/*
 
 #########################################################################################
 #
@@ -1057,10 +1057,17 @@ RUN make -j $(getconf _NPROCESSORS_ONLN) && \
 
 #########################################################################################
 #
-# Layer "build-deps with Rust toolchain installed"
+# Layer "pg build with nonroot user and cargo installed"
+# This layer is base and common for layers with `pgrx`
 #
 #########################################################################################
-FROM build-deps AS build-deps-with-cargo
+FROM pg-build AS pg-build-nonroot-with-cargo
+ARG PG_VERSION
+
+RUN apt update && \
+    apt install --no-install-recommends --no-install-suggests -y curl libclang-dev && \
+    apt clean && rm -rf /var/lib/apt/lists/* && \
+    useradd -ms /bin/bash nonroot -b /home
 
 ENV HOME=/home/nonroot
 ENV PATH="/home/nonroot/.cargo/bin:$PATH"
@@ -1075,29 +1082,13 @@ RUN curl -sSO https://static.rust-lang.org/rustup/dist/$(uname -m)-unknown-linux
     ./rustup-init -y --no-modify-path --profile minimal --default-toolchain stable && \
     rm rustup-init
 
-#########################################################################################
-#
-# Layer "pg-build with Rust toolchain installed"
-# This layer is base and common for layers with `pgrx`
-#
-#########################################################################################
-FROM pg-build AS pg-build-with-cargo
-ARG PG_VERSION
-
-ENV HOME=/home/nonroot
-ENV PATH="/home/nonroot/.cargo/bin:$PATH"
-USER nonroot
-WORKDIR /home/nonroot
-
-COPY --from=build-deps-with-cargo /home/nonroot /home/nonroot
-
 #########################################################################################
 #
 # Layer "rust extensions"
 # This layer is used to build `pgrx` deps
 #
 #########################################################################################
-FROM pg-build-with-cargo AS rust-extensions-build
+FROM pg-build-nonroot-with-cargo AS rust-extensions-build
 ARG PG_VERSION
 
 RUN case "${PG_VERSION:?}" in \
@@ -1119,7 +1110,7 @@ USER root
 # and eventually get merged with `rust-extensions-build`
 #
 #########################################################################################
-FROM pg-build-with-cargo AS rust-extensions-build-pgrx12
+FROM pg-build-nonroot-with-cargo AS rust-extensions-build-pgrx12
 ARG PG_VERSION
 
 RUN cargo install --locked --version 0.12.9 cargo-pgrx && \
@@ -1136,7 +1127,7 @@ USER root
 # and eventually get merged with `rust-extensions-build`
 #
 #########################################################################################
-FROM pg-build-with-cargo AS rust-extensions-build-pgrx14
+FROM pg-build-nonroot-with-cargo AS rust-extensions-build-pgrx14
 ARG PG_VERSION
 
 RUN cargo install --locked --version 0.14.1 cargo-pgrx && \
@@ -1153,12 +1144,10 @@ USER root
 
 FROM build-deps AS pgrag-src
 ARG PG_VERSION
-WORKDIR /ext-src
-COPY compute/patches/onnxruntime.patch .
 
+WORKDIR /ext-src
 RUN wget https://github.com/microsoft/onnxruntime/archive/refs/tags/v1.18.1.tar.gz -O onnxruntime.tar.gz && \
     mkdir onnxruntime-src && cd onnxruntime-src && tar xzf ../onnxruntime.tar.gz --strip-components=1 -C . && \
-    patch -p1 < /ext-src/onnxruntime.patch && \
     echo "#nothing to test here" > neon-test.sh
 
 RUN wget https://github.com/neondatabase-labs/pgrag/archive/refs/tags/v0.1.2.tar.gz -O pgrag.tar.gz &&  \
@@ -1632,7 +1621,18 @@ FROM pg-build AS neon-ext-build
 ARG PG_VERSION
 
 COPY pgxn/ pgxn/
-RUN make -j $(getconf _NPROCESSORS_ONLN) -C pgxn -s install-compute
+RUN make -j $(getconf _NPROCESSORS_ONLN) \
+        -C pgxn/neon \
+        -s install && \
+    make -j $(getconf _NPROCESSORS_ONLN) \
+        -C pgxn/neon_utils \
+        -s install && \
+    make -j $(getconf _NPROCESSORS_ONLN) \
+        -C pgxn/neon_test_utils \
+        -s install && \
+    make -j $(getconf _NPROCESSORS_ONLN) \
+        -C pgxn/neon_rmgr \
+        -s install
 
 #########################################################################################
 #
@@ -1722,7 +1722,7 @@ FROM extensions-${EXTENSIONS} AS neon-pg-ext-build
 # Compile the Neon-specific `compute_ctl`, `fast_import`, and `local_proxy` binaries
 #
 #########################################################################################
-FROM build-deps-with-cargo AS compute-tools
+FROM $REPOSITORY/$IMAGE:$TAG AS compute-tools
 ARG BUILD_TAG
 ENV BUILD_TAG=$BUILD_TAG
 
@@ -1732,7 +1732,7 @@ COPY --chown=nonroot . .
 RUN --mount=type=cache,uid=1000,target=/home/nonroot/.cargo/registry \
     --mount=type=cache,uid=1000,target=/home/nonroot/.cargo/git \
     --mount=type=cache,uid=1000,target=/home/nonroot/target \
-    cargo build --locked --profile release-line-debug-size-lto --bin compute_ctl --bin fast_import --bin local_proxy && \
+    mold -run cargo build --locked --profile release-line-debug-size-lto --bin compute_ctl --bin fast_import --bin local_proxy && \
     mkdir target-bin && \
     cp target/release-line-debug-size-lto/compute_ctl \
        target/release-line-debug-size-lto/fast_import \
@@ -1826,11 +1826,10 @@ RUN rm /usr/local/pgsql/lib/lib*.a
 # Preprocess the sql_exporter configuration files
 #
 #########################################################################################
-FROM build-deps AS sql_exporter_preprocessor
+FROM $REPOSITORY/$IMAGE:$TAG AS sql_exporter_preprocessor
 ARG PG_VERSION
 
 USER nonroot
-WORKDIR /home/nonroot
 
 COPY --chown=nonroot compute compute
 

compute/etc/pgbouncer.ini

@@ -21,8 +21,6 @@ unix_socket_dir=/tmp/
 unix_socket_mode=0777
 ; required for pgbouncer_exporter
 ignore_startup_parameters=extra_float_digits
-; pidfile for graceful termination
-pidfile=/tmp/pgbouncer.pid
 
 ;; Disable connection logging. It produces a lot of logs that no one looks at,
 ;; and we can get similar log entries from the proxy too. We had incidents in

compute/manifest.schema.json

@@ -1,209 +0,0 @@
-{
-  "$schema": "https://json-schema.org/draft/2020-12/schema",
-  "title": "Neon Compute Manifest Schema",
-  "description": "Schema for Neon compute node configuration manifest",
-  "type": "object",
-  "properties": {
-    "pg_settings": {
-      "type": "object",
-      "properties": {
-        "common": {
-          "type": "object",
-          "properties": {
-            "client_connection_check_interval": {
-              "type": "string",
-              "description": "Check for client disconnection interval in milliseconds"
-            },
-            "effective_io_concurrency": {
-              "type": "string",
-              "description": "Effective IO concurrency setting"
-            },
-            "fsync": {
-              "type": "string",
-              "enum": ["on", "off"],
-              "description": "Whether to force fsync to disk"
-            },
-            "hot_standby": {
-              "type": "string",
-              "enum": ["on", "off"],
-              "description": "Whether hot standby is enabled"
-            },
-            "idle_in_transaction_session_timeout": {
-              "type": "string",
-              "description": "Timeout for idle transactions in milliseconds"
-            },
-            "listen_addresses": {
-              "type": "string",
-              "description": "Addresses to listen on"
-            },
-            "log_connections": {
-              "type": "string",
-              "enum": ["on", "off"],
-              "description": "Whether to log connections"
-            },
-            "log_disconnections": {
-              "type": "string",
-              "enum": ["on", "off"],
-              "description": "Whether to log disconnections"
-            },
-            "log_temp_files": {
-              "type": "string",
-              "description": "Size threshold for logging temporary files in KB"
-            },
-            "log_error_verbosity": {
-              "type": "string",
-              "enum": ["terse", "verbose", "default"],
-              "description": "Error logging verbosity level"
-            },
-            "log_min_error_statement": {
-              "type": "string",
-              "description": "Minimum error level for statement logging"
-            },
-            "maintenance_io_concurrency": {
-              "type": "string",
-              "description": "Maintenance IO concurrency setting"
-            },
-            "max_connections": {
-              "type": "string",
-              "description": "Maximum number of connections"
-            },
-            "max_replication_flush_lag": {
-              "type": "string",
-              "description": "Maximum replication flush lag"
-            },
-            "max_replication_slots": {
-              "type": "string",
-              "description": "Maximum number of replication slots"
-            },
-            "max_replication_write_lag": {
-              "type": "string",
-              "description": "Maximum replication write lag"
-            },
-            "max_wal_senders": {
-              "type": "string",
-              "description": "Maximum number of WAL senders"
-            },
-            "max_wal_size": {
-              "type": "string",
-              "description": "Maximum WAL size"
-            },
-            "neon.unstable_extensions": {
-              "type": "string",
-              "description": "List of unstable extensions"
-            },
-            "neon.protocol_version": {
-              "type": "string",
-              "description": "Neon protocol version"
-            },
-            "password_encryption": {
-              "type": "string",
-              "description": "Password encryption method"
-            },
-            "restart_after_crash": {
-              "type": "string",
-              "enum": ["on", "off"],
-              "description": "Whether to restart after crash"
-            },
-            "superuser_reserved_connections": {
-              "type": "string",
-              "description": "Number of reserved connections for superuser"
-            },
-            "synchronous_standby_names": {
-              "type": "string",
-              "description": "Names of synchronous standby servers"
-            },
-            "wal_keep_size": {
-              "type": "string",
-              "description": "WAL keep size"
-            },
-            "wal_level": {
-              "type": "string",
-              "description": "WAL level"
-            },
-            "wal_log_hints": {
-              "type": "string",
-              "enum": ["on", "off"],
-              "description": "Whether to log hints in WAL"
-            },
-            "wal_sender_timeout": {
-              "type": "string",
-              "description": "WAL sender timeout in milliseconds"
-            }
-          },
-          "required": [
-            "client_connection_check_interval",
-            "effective_io_concurrency",
-            "fsync",
-            "hot_standby",
-            "idle_in_transaction_session_timeout",
-            "listen_addresses",
-            "log_connections",
-            "log_disconnections",
-            "log_temp_files",
-            "log_error_verbosity",
-            "log_min_error_statement",
-            "maintenance_io_concurrency",
-            "max_connections",
-            "max_replication_flush_lag",
-            "max_replication_slots",
-            "max_replication_write_lag",
-            "max_wal_senders",
-            "max_wal_size",
-            "neon.unstable_extensions",
-            "neon.protocol_version",
-            "password_encryption",
-            "restart_after_crash",
-            "superuser_reserved_connections",
-            "synchronous_standby_names",
-            "wal_keep_size",
-            "wal_level",
-            "wal_log_hints",
-            "wal_sender_timeout"
-          ]
-        },
-        "replica": {
-          "type": "object",
-          "properties": {
-            "hot_standby": {
-              "type": "string",
-              "enum": ["on", "off"],
-              "description": "Whether hot standby is enabled for replicas"
-            }
-          },
-          "required": ["hot_standby"]
-        },
-        "per_version": {
-          "type": "object",
-          "patternProperties": {
-            "^1[4-7]$": {
-              "type": "object",
-              "properties": {
-                "common": {
-                  "type": "object",
-                  "properties": {
-                    "io_combine_limit": {
-                      "type": "string",
-                      "description": "IO combine limit"
-                    }
-                  }
-                },
-                "replica": {
-                  "type": "object",
-                  "properties": {
-                    "recovery_prefetch": {
-                      "type": "string",
-                      "enum": ["on", "off"],
-                      "description": "Whether to enable recovery prefetch for PostgreSQL replicas"
-                    }
-                  }
-                }
-              }
-            }
-          }
-        }
-      },
-      "required": ["common", "replica", "per_version"]
-    }
-  },
-  "required": ["pg_settings"]
-} 

compute/manifest.yaml

@@ -105,17 +105,17 @@ pg_settings:
         # Neon hot standby ignores pages that are not in the shared_buffers
         recovery_prefetch: "off"
     16:
-      common: {}
+      common:
       replica:
         # prefetching of blocks referenced in WAL doesn't make sense for us
         # Neon hot standby ignores pages that are not in the shared_buffers
         recovery_prefetch: "off"
     15:
-      common: {}
+      common:
       replica:
         # prefetching of blocks referenced in WAL doesn't make sense for us
         # Neon hot standby ignores pages that are not in the shared_buffers
         recovery_prefetch: "off"
     14:
-      common: {}
-      replica: {}
+      common:
+      replica:

compute/package-lock.json

@@ -1,37 +0,0 @@
-{
-  "name": "neon-compute",
-  "lockfileVersion": 3,
-  "requires": true,
-  "packages": {
-    "": {
-      "name": "neon-compute",
-      "dependencies": {
-        "@sourcemeta/jsonschema": "9.3.4"
-      }
-    },
-    "node_modules/@sourcemeta/jsonschema": {
-      "version": "9.3.4",
-      "resolved": "https://registry.npmjs.org/@sourcemeta/jsonschema/-/jsonschema-9.3.4.tgz",
-      "integrity": "sha512-hkujfkZAIGXUs4U//We9faZW8LZ4/H9LqagRYsFSulH/VLcKPNhZyCTGg7AhORuzm27zqENvKpnX4g2FzudYFw==",
-      "cpu": [
-        "x64",
-        "arm64"
-      ],
-      "license": "AGPL-3.0",
-      "os": [
-        "darwin",
-        "linux",
-        "win32"
-      ],
-      "bin": {
-        "jsonschema": "cli.js"
-      },
-      "engines": {
-        "node": ">=16"
-      },
-      "funding": {
-        "url": "https://github.com/sponsors/sourcemeta"
-      }
-    }
-  }
-}

compute/package.json

@@ -1,7 +0,0 @@
-{
-  "name": "neon-compute",
-  "private": true,
-  "dependencies": {
-    "@sourcemeta/jsonschema": "9.3.4"
-  }
-} 

compute/patches/onnxruntime.patch

@@ -1,15 +0,0 @@
-diff --git a/cmake/deps.txt b/cmake/deps.txt
-index d213b09034..229de2ebf0 100644
---- a/cmake/deps.txt
-+++ b/cmake/deps.txt
-@@ -22,7 +22,9 @@ dlpack;https://github.com/dmlc/dlpack/archive/refs/tags/v0.6.zip;4d565dd2e5b3132
- # it contains changes on top of 3.4.0 which are required to fix build issues.
- # Until the 3.4.1 release this is the best option we have.
- # Issue link: https://gitlab.com/libeigen/eigen/-/issues/2744
--eigen;https://gitlab.com/libeigen/eigen/-/archive/e7248b26a1ed53fa030c5c459f7ea095dfd276ac/eigen-e7248b26a1ed53fa030c5c459f7ea095dfd276ac.zip;be8be39fdbc6e60e94fa7870b280707069b5b81a
-+# Moved to github mirror to avoid gitlab issues.Add commentMore actions
-+# Issue link: https://github.com/bazelbuild/bazel-central-registry/issues/4355
-+eigen;https://github.com/eigen-mirror/eigen/archive/e7248b26a1ed53fa030c5c459f7ea095dfd276ac/eigen-e7248b26a1ed53fa030c5c459f7ea095dfd276ac.zip;61418a349000ba7744a3ad03cf5071f22ebf860a
- flatbuffers;https://github.com/google/flatbuffers/archive/refs/tags/v23.5.26.zip;59422c3b5e573dd192fead2834d25951f1c1670c
- fp16;https://github.com/Maratyszcza/FP16/archive/0a92994d729ff76a58f692d3028ca1b64b145d91.zip;b985f6985a05a1c03ff1bb71190f66d8f98a1494
- fxdiv;https://github.com/Maratyszcza/FXdiv/archive/63058eff77e11aa15bf531df5dd34395ec3017c8.zip;a5658f4036402dbca7cebee32be57fb8149811e1

compute_tools/src/bin/compute_ctl.rs

@@ -124,10 +124,6 @@ struct Cli {
     /// Interval in seconds for collecting installed extensions statistics
     #[arg(long, default_value = "3600")]
     pub installed_extensions_collection_interval: u64,
-
-    /// Run in development mode, skipping VM-specific operations like process termination
-    #[arg(long, action = clap::ArgAction::SetTrue)]
-    pub dev: bool,
 }
 
 impl Cli {
@@ -163,7 +159,7 @@ fn main() -> Result<()> {
         .build()?;
     let _rt_guard = runtime.enter();
 
-    runtime.block_on(init(cli.dev))?;
+    runtime.block_on(init())?;
 
     // enable core dumping for all child processes
     setrlimit(Resource::CORE, rlimit::INFINITY, rlimit::INFINITY)?;
@@ -202,13 +198,13 @@ fn main() -> Result<()> {
     deinit_and_exit(exit_code);
 }
 
-async fn init(dev_mode: bool) -> Result<()> {
+async fn init() -> Result<()> {
     init_tracing_and_logging(DEFAULT_LOG_LEVEL).await?;
 
     let mut signals = Signals::new([SIGINT, SIGTERM, SIGQUIT])?;
     thread::spawn(move || {
         for sig in signals.forever() {
-            handle_exit_signal(sig, dev_mode);
+            handle_exit_signal(sig);
         }
     });
 
@@ -267,9 +263,9 @@ fn deinit_and_exit(exit_code: Option<i32>) -> ! {
 /// When compute_ctl is killed, send also termination signal to sync-safekeepers
 /// to prevent leakage. TODO: it is better to convert compute_ctl to async and
 /// wait for termination which would be easy then.
-fn handle_exit_signal(sig: i32, dev_mode: bool) {
+fn handle_exit_signal(sig: i32) {
     info!("received {sig} termination signal");
-    forward_termination_signal(dev_mode);
+    forward_termination_signal();
     exit(1);
 }
 

compute_tools/src/bin/fast_import.rs

@@ -486,8 +486,10 @@ async fn cmd_pgdata(
     };
 
     let superuser = "cloud_admin";
-    let destination_connstring =
-        format!("host=localhost port={pg_port} user={superuser} dbname=neondb");
+    let destination_connstring = format!(
+        "host=localhost port={} user={} dbname=neondb",
+        pg_port, superuser
+    );
 
     let pgdata_dir = workdir.join("pgdata");
     let mut proc = PostgresProcess::new(pgdata_dir.clone(), pg_bin_dir.clone(), pg_lib_dir.clone());

compute_tools/src/bin/fast_import/s3_uri.rs

@@ -69,7 +69,7 @@ impl clap::builder::TypedValueParser for S3Uri {
         S3Uri::from_str(value_str).map_err(|e| {
             clap::Error::raw(
                 clap::error::ErrorKind::InvalidValue,
-                format!("Failed to parse S3 URI: {e}"),
+                format!("Failed to parse S3 URI: {}", e),
             )
         })
     }

compute_tools/src/catalog.rs

@@ -22,7 +22,7 @@ pub async fn get_dbs_and_roles(compute: &Arc<ComputeNode>) -> anyhow::Result<Cat
 
     spawn(async move {
         if let Err(e) = connection.await {
-            eprintln!("connection error: {e}");
+            eprintln!("connection error: {}", e);
         }
     });
 
@@ -119,7 +119,7 @@ pub async fn get_database_schema(
         _ => {
             let mut lines = stderr_reader.lines();
             if let Some(line) = lines.next_line().await? {
-                if line.contains(&format!("FATAL:  database \"{dbname}\" does not exist")) {
+                if line.contains(&format!("FATAL:  database \"{}\" does not exist", dbname)) {
                     return Err(SchemaDumpError::DatabaseDoesNotExist);
                 }
                 warn!("pg_dump stderr: {}", line)

compute_tools/src/compute.rs

@@ -35,7 +35,6 @@ use url::Url;
 use utils::id::{TenantId, TimelineId};
 use utils::lsn::Lsn;
 use utils::measured_stream::MeasuredReader;
-use utils::pid_file;
 
 use crate::configurator::launch_configurator;
 use crate::disk_quota::set_disk_quota;
@@ -45,7 +44,6 @@ use crate::lsn_lease::launch_lsn_lease_bg_task_for_static;
 use crate::metrics::COMPUTE_CTL_UP;
 use crate::monitor::launch_monitor;
 use crate::pg_helpers::*;
-use crate::pgbouncer::*;
 use crate::rsyslog::{
     PostgresLogsRsyslogConfig, configure_audit_rsyslog, configure_postgres_logs_export,
     launch_pgaudit_gc,
@@ -163,10 +161,6 @@ pub struct ComputeState {
     pub lfc_prewarm_state: LfcPrewarmState,
     pub lfc_offload_state: LfcOffloadState,
 
-    /// WAL flush LSN that is set after terminating Postgres and syncing safekeepers if
-    /// mode == ComputeMode::Primary. None otherwise
-    pub terminate_flush_lsn: Option<Lsn>,
-
     pub metrics: ComputeMetrics,
 }
 
@@ -182,7 +176,6 @@ impl ComputeState {
             metrics: ComputeMetrics::default(),
             lfc_prewarm_state: LfcPrewarmState::default(),
             lfc_offload_state: LfcOffloadState::default(),
-            terminate_flush_lsn: None,
         }
     }
 
@@ -222,45 +215,6 @@ pub struct ParsedSpec {
     pub endpoint_storage_token: Option<String>,
 }
 
-impl ParsedSpec {
-    pub fn validate(&self) -> Result<(), String> {
-        // Only Primary nodes are using safekeeper_connstrings, and at the moment
-        // this method only validates that part of the specs.
-        if self.spec.mode != ComputeMode::Primary {
-            return Ok(());
-        }
-
-        // While it seems like a good idea to check for an odd number of entries in
-        // the safekeepers connection string, changes to the list of safekeepers might
-        // incur appending a new server to a list of 3, in which case a list of 4
-        // entries is okay in production.
-        //
-        // Still we want unique entries, and at least one entry in the vector
-        if self.safekeeper_connstrings.is_empty() {
-            return Err(String::from("safekeeper_connstrings is empty"));
-        }
-
-        // check for uniqueness of the connection strings in the set
-        let mut connstrings = self.safekeeper_connstrings.clone();
-
-        connstrings.sort();
-        let mut previous = &connstrings[0];
-
-        for current in connstrings.iter().skip(1) {
-            // duplicate entry?
-            if current == previous {
-                return Err(format!(
-                    "duplicate entry in safekeeper_connstrings: {current}!",
-                ));
-            }
-
-            previous = current;
-        }
-
-        Ok(())
-    }
-}
-
 impl TryFrom<ComputeSpec> for ParsedSpec {
     type Error = String;
     fn try_from(spec: ComputeSpec) -> Result<Self, String> {
@@ -290,7 +244,6 @@ impl TryFrom<ComputeSpec> for ParsedSpec {
         } else {
             spec.safekeeper_connstrings.clone()
         };
-
         let storage_auth_token = spec.storage_auth_token.clone();
         let tenant_id: TenantId = if let Some(tenant_id) = spec.tenant_id {
             tenant_id
@@ -325,7 +278,7 @@ impl TryFrom<ComputeSpec> for ParsedSpec {
             .clone()
             .or_else(|| spec.cluster.settings.find("neon.endpoint_storage_token"));
 
-        let res = ParsedSpec {
+        Ok(ParsedSpec {
             spec,
             pageserver_connstr,
             safekeeper_connstrings,
@@ -334,11 +287,7 @@ impl TryFrom<ComputeSpec> for ParsedSpec {
             timeline_id,
             endpoint_storage_addr,
             endpoint_storage_token,
-        };
-
-        // Now check validity of the parsed specification
-        res.validate()?;
-        Ok(res)
+        })
     }
 }
 
@@ -405,11 +354,14 @@ impl ComputeNode {
         // that can affect `compute_ctl` and prevent it from properly configuring the database schema.
         // Unset them via connection string options before connecting to the database.
         // N.B. keep it in sync with `ZENITH_OPTIONS` in `get_maintenance_client()`.
+        //
+        // TODO(ololobus): we currently pass `-c default_transaction_read_only=off` from control plane
+        // as well. After rolling out this code, we can remove this parameter from control plane.
+        // In the meantime, double-passing is fine, the last value is applied.
+        // See: <https://github.com/neondatabase/cloud/blob/133dd8c4dbbba40edfbad475bf6a45073ca63faf/goapp/controlplane/internal/pkg/compute/provisioner/provisioner_common.go#L70>
         const EXTRA_OPTIONS: &str = "-c role=cloud_admin -c default_transaction_read_only=off -c search_path=public -c statement_timeout=0";
         let options = match conn_conf.get_options() {
-            // Allow the control plane to override any options set by the
-            // compute
-            Some(options) => format!("{EXTRA_OPTIONS} {options}"),
+            Some(options) => format!("{} {}", options, EXTRA_OPTIONS),
             None => EXTRA_OPTIONS.to_string(),
         };
         conn_conf.options(&options);
@@ -537,21 +489,12 @@ impl ComputeNode {
         // Reap the postgres process
         delay_exit |= this.cleanup_after_postgres_exit()?;
 
-        // /terminate returns LSN. If we don't sleep at all, connection will break and we
-        // won't get result. If we sleep too much, tests will take significantly longer
-        // and Github Action run will error out
-        let sleep_duration = if delay_exit {
-            Duration::from_secs(30)
-        } else {
-            Duration::from_millis(300)
-        };
-
         // If launch failed, keep serving HTTP requests for a while, so the cloud
         // control plane can get the actual error.
         if delay_exit {
             info!("giving control plane 30s to collect the error before shutdown");
+            std::thread::sleep(Duration::from_secs(30));
         }
-        std::thread::sleep(sleep_duration);
         Ok(exit_code)
     }
 
@@ -923,25 +866,20 @@ impl ComputeNode {
         // Maybe sync safekeepers again, to speed up next startup
         let compute_state = self.state.lock().unwrap().clone();
         let pspec = compute_state.pspec.as_ref().expect("spec must be set");
-        let lsn = if matches!(pspec.spec.mode, compute_api::spec::ComputeMode::Primary) {
+        if matches!(pspec.spec.mode, compute_api::spec::ComputeMode::Primary) {
             info!("syncing safekeepers on shutdown");
             let storage_auth_token = pspec.storage_auth_token.clone();
             let lsn = self.sync_safekeepers(storage_auth_token)?;
-            info!(%lsn, "synced safekeepers");
-            Some(lsn)
-        } else {
-            info!("not primary, not syncing safekeepers");
-            None
-        };
+            info!("synced safekeepers at lsn {lsn}");
+        }
 
         let mut delay_exit = false;
         let mut state = self.state.lock().unwrap();
-        state.terminate_flush_lsn = lsn;
-        if let ComputeStatus::TerminationPending { mode } = state.status {
+        if state.status == ComputeStatus::TerminationPending {
             state.status = ComputeStatus::Terminated;
             self.state_changed.notify_all();
             // we were asked to terminate gracefully, don't exit to avoid restart
-            delay_exit = mode == compute_api::responses::TerminateMode::Fast
+            delay_exit = true
         }
         drop(state);
 
@@ -1126,7 +1064,7 @@ impl ComputeNode {
         let sk_configs = sk_connstrs.into_iter().map(|connstr| {
             // Format connstr
             let id = connstr.clone();
-            let connstr = format!("postgresql://no_user@{connstr}");
+            let connstr = format!("postgresql://no_user@{}", connstr);
             let options = format!(
                 "-c timeline_id={} tenant_id={}",
                 pspec.timeline_id, pspec.tenant_id
@@ -1489,7 +1427,7 @@ impl ComputeNode {
                 let (mut client, connection) = conf.connect(NoTls).await?;
                 tokio::spawn(async move {
                     if let Err(e) = connection.await {
-                        eprintln!("connection error: {e}");
+                        eprintln!("connection error: {}", e);
                     }
                 });
 
@@ -1632,7 +1570,7 @@ impl ComputeNode {
                 Ok((mut client, connection)) => {
                     tokio::spawn(async move {
                         if let Err(e) = connection.await {
-                            eprintln!("connection error: {e}");
+                            eprintln!("connection error: {}", e);
                         }
                     });
                     if let Err(e) = handle_migrations(&mut client).await {
@@ -1812,7 +1750,7 @@ impl ComputeNode {
 
                             // exit loop
                             ComputeStatus::Failed
-                            | ComputeStatus::TerminationPending { .. }
+                            | ComputeStatus::TerminationPending
                             | ComputeStatus::Terminated => break 'cert_update,
 
                             // wait
@@ -1936,7 +1874,7 @@ impl ComputeNode {
         let (client, connection) = connect_result.unwrap();
         tokio::spawn(async move {
             if let Err(e) = connection.await {
-                eprintln!("connection error: {e}");
+                eprintln!("connection error: {}", e);
             }
         });
         let result = client
@@ -2105,7 +2043,7 @@ LIMIT 100",
             db_client
                 .simple_query(&query)
                 .await
-                .with_context(|| format!("Failed to execute query: {query}"))?;
+                .with_context(|| format!("Failed to execute query: {}", query))?;
         }
 
         Ok(())
@@ -2132,7 +2070,7 @@ LIMIT 100",
         let version: Option<ExtVersion> = db_client
             .query_opt(version_query, &[&ext_name])
             .await
-            .with_context(|| format!("Failed to execute query: {version_query}"))?
+            .with_context(|| format!("Failed to execute query: {}", version_query))?
             .map(|row| row.get(0));
 
         // sanitize the inputs as postgres idents.
@@ -2147,14 +2085,14 @@ LIMIT 100",
             db_client
                 .simple_query(&query)
                 .await
-                .with_context(|| format!("Failed to execute query: {query}"))?;
+                .with_context(|| format!("Failed to execute query: {}", query))?;
         } else {
             let query =
                 format!("CREATE EXTENSION IF NOT EXISTS {ext_name} WITH VERSION {quoted_version}");
             db_client
                 .simple_query(&query)
                 .await
-                .with_context(|| format!("Failed to execute query: {query}"))?;
+                .with_context(|| format!("Failed to execute query: {}", query))?;
         }
 
         Ok(ext_version)
@@ -2313,68 +2251,12 @@ pub async fn installed_extensions(conf: tokio_postgres::Config) -> Result<()> {
     Ok(())
 }
 
-pub fn forward_termination_signal(dev_mode: bool) {
+pub fn forward_termination_signal() {
     let ss_pid = SYNC_SAFEKEEPERS_PID.load(Ordering::SeqCst);
     if ss_pid != 0 {
         let ss_pid = nix::unistd::Pid::from_raw(ss_pid as i32);
         kill(ss_pid, Signal::SIGTERM).ok();
     }
-
-    if !dev_mode {
-        //  Terminate pgbouncer with SIGKILL
-        match pid_file::read(PGBOUNCER_PIDFILE.into()) {
-            Ok(pid_file::PidFileRead::LockedByOtherProcess(pid)) => {
-                info!("sending SIGKILL to pgbouncer process pid: {}", pid);
-                if let Err(e) = kill(pid, Signal::SIGKILL) {
-                    error!("failed to terminate pgbouncer: {}", e);
-                }
-            }
-            // pgbouncer does not lock the pid file, so we read and kill the process directly
-            Ok(pid_file::PidFileRead::NotHeldByAnyProcess(_)) => {
-                if let Ok(pid_str) = std::fs::read_to_string(PGBOUNCER_PIDFILE) {
-                    if let Ok(pid) = pid_str.trim().parse::<i32>() {
-                        info!(
-                            "sending SIGKILL to pgbouncer process pid: {} (from unlocked pid file)",
-                            pid
-                        );
-                        if let Err(e) = kill(Pid::from_raw(pid), Signal::SIGKILL) {
-                            error!("failed to terminate pgbouncer: {}", e);
-                        }
-                    }
-                } else {
-                    info!("pgbouncer pid file exists but process not running");
-                }
-            }
-            Ok(pid_file::PidFileRead::NotExist) => {
-                info!("pgbouncer pid file not found, process may not be running");
-            }
-            Err(e) => {
-                error!("error reading pgbouncer pid file: {}", e);
-            }
-        }
-
-        // Terminate local_proxy
-        match pid_file::read("/etc/local_proxy/pid".into()) {
-            Ok(pid_file::PidFileRead::LockedByOtherProcess(pid)) => {
-                info!("sending SIGTERM to local_proxy process pid: {}", pid);
-                if let Err(e) = kill(pid, Signal::SIGTERM) {
-                    error!("failed to terminate local_proxy: {}", e);
-                }
-            }
-            Ok(pid_file::PidFileRead::NotHeldByAnyProcess(_)) => {
-                info!("local_proxy PID file exists but process not running");
-            }
-            Ok(pid_file::PidFileRead::NotExist) => {
-                info!("local_proxy PID file not found, process may not be running");
-            }
-            Err(e) => {
-                error!("error reading local_proxy PID file: {}", e);
-            }
-        }
-    } else {
-        info!("Skipping pgbouncer and local_proxy termination because in dev mode");
-    }
-
     let pg_pid = PG_PID.load(Ordering::SeqCst);
     if pg_pid != 0 {
         let pg_pid = nix::unistd::Pid::from_raw(pg_pid as i32);
@@ -2407,21 +2289,3 @@ impl<T: 'static> JoinSetExt<T> for tokio::task::JoinSet<T> {
         })
     }
 }
-
-#[cfg(test)]
-mod tests {
-    use std::fs::File;
-
-    use super::*;
-
-    #[test]
-    fn duplicate_safekeeper_connstring() {
-        let file = File::open("tests/cluster_spec.json").unwrap();
-        let spec: ComputeSpec = serde_json::from_reader(file).unwrap();
-
-        match ParsedSpec::try_from(spec.clone()) {
-            Ok(_p) => panic!("Failed to detect duplicate entry"),
-            Err(e) => assert!(e.starts_with("duplicate entry in safekeeper_connstrings:")),
-        };
-    }
-}

compute_tools/src/config.rs

@@ -51,7 +51,7 @@ pub fn write_postgres_conf(
 
     // Write the postgresql.conf content from the spec file as is.
     if let Some(conf) = &spec.cluster.postgresql_conf {
-        writeln!(file, "{conf}")?;
+        writeln!(file, "{}", conf)?;
     }
 
     // Add options for connecting to storage
@@ -70,7 +70,7 @@ pub fn write_postgres_conf(
         );
         // If generation is given, prepend sk list with g#number:
         if let Some(generation) = spec.safekeepers_generation {
-            write!(neon_safekeepers_value, "g#{generation}:")?;
+            write!(neon_safekeepers_value, "g#{}:", generation)?;
         }
         neon_safekeepers_value.push_str(&spec.safekeeper_connstrings.join(","));
         writeln!(
@@ -109,8 +109,8 @@ pub fn write_postgres_conf(
         tls::update_key_path_blocking(pgdata_path, tls_config);
 
         // these are the default, but good to be explicit.
-        writeln!(file, "ssl_cert_file = '{SERVER_CRT}'")?;
-        writeln!(file, "ssl_key_file = '{SERVER_KEY}'")?;
+        writeln!(file, "ssl_cert_file = '{}'", SERVER_CRT)?;
+        writeln!(file, "ssl_key_file = '{}'", SERVER_KEY)?;
     }
 
     // Locales
@@ -191,7 +191,8 @@ pub fn write_postgres_conf(
                 }
                 writeln!(
                     file,
-                    "shared_preload_libraries='{libs}{extra_shared_preload_libraries}'"
+                    "shared_preload_libraries='{}{}'",
+                    libs, extra_shared_preload_libraries
                 )?;
             } else {
                 // Typically, this should be unreacheable,
@@ -243,7 +244,8 @@ pub fn write_postgres_conf(
                 }
                 writeln!(
                     file,
-                    "shared_preload_libraries='{libs}{extra_shared_preload_libraries}'"
+                    "shared_preload_libraries='{}{}'",
+                    libs, extra_shared_preload_libraries
                 )?;
             } else {
                 // Typically, this should be unreacheable,
@@ -261,7 +263,7 @@ pub fn write_postgres_conf(
         }
     }
 
-    writeln!(file, "neon.extension_server_port={extension_server_port}")?;
+    writeln!(file, "neon.extension_server_port={}", extension_server_port)?;
 
     if spec.drop_subscriptions_before_start {
         writeln!(file, "neon.disable_logical_replication_subscribers=true")?;
@@ -289,7 +291,7 @@ where
 {
     let path = pgdata_path.join("compute_ctl_temp_override.conf");
     let mut file = File::create(path)?;
-    write!(file, "{options}")?;
+    write!(file, "{}", options)?;
 
     let res = exec();
 

compute_tools/src/extension_server.rs

@@ -310,7 +310,10 @@ async fn download_extension_tar(remote_ext_base_url: &Url, ext_path: &str) -> Re
 async fn do_extension_server_request(uri: Url) -> Result<Bytes, (String, String)> {
     let resp = reqwest::get(uri).await.map_err(|e| {
         (
-            format!("could not perform remote extensions server request: {e:?}"),
+            format!(
+                "could not perform remote extensions server request: {:?}",
+                e
+            ),
             UNKNOWN_HTTP_STATUS.to_string(),
         )
     })?;
@@ -320,7 +323,7 @@ async fn do_extension_server_request(uri: Url) -> Result<Bytes, (String, String)
         StatusCode::OK => match resp.bytes().await {
             Ok(resp) => Ok(resp),
             Err(e) => Err((
-                format!("could not read remote extensions server response: {e:?}"),
+                format!("could not read remote extensions server response: {:?}", e),
                 // It's fine to return and report error with status as 200 OK,
                 // because we still failed to read the response.
                 status.to_string(),
@@ -331,7 +334,10 @@ async fn do_extension_server_request(uri: Url) -> Result<Bytes, (String, String)
             status.to_string(),
         )),
         _ => Err((
-            format!("unexpected remote extensions server response status code: {status}"),
+            format!(
+                "unexpected remote extensions server response status code: {}",
+                status
+            ),
             status.to_string(),
         )),
     }

compute_tools/src/http/routes/configure.rs

@@ -65,7 +65,7 @@ pub(in crate::http) async fn configure(
 
             if state.status == ComputeStatus::Failed {
                 let err = state.error.as_ref().map_or("unknown error", |x| x);
-                let msg = format!("compute configuration failed: {err:?}");
+                let msg = format!("compute configuration failed: {:?}", err);
                 return Err(msg);
             }
         }

compute_tools/src/http/routes/terminate.rs

@@ -1,42 +1,32 @@
-use crate::compute::{ComputeNode, forward_termination_signal};
-use crate::http::JsonResponse;
-use axum::extract::State;
-use axum::response::Response;
-use axum_extra::extract::OptionalQuery;
-use compute_api::responses::{ComputeStatus, TerminateResponse};
-use http::StatusCode;
-use serde::Deserialize;
 use std::sync::Arc;
+
+use axum::extract::State;
+use axum::response::{IntoResponse, Response};
+use compute_api::responses::ComputeStatus;
+use http::StatusCode;
 use tokio::task;
 use tracing::info;
 
-#[derive(Deserialize, Default)]
-pub struct TerminateQuery {
-    mode: compute_api::responses::TerminateMode,
-}
+use crate::compute::{ComputeNode, forward_termination_signal};
+use crate::http::JsonResponse;
 
 /// Terminate the compute.
-pub(in crate::http) async fn terminate(
-    State(compute): State<Arc<ComputeNode>>,
-    OptionalQuery(terminate): OptionalQuery<TerminateQuery>,
-) -> Response {
-    let mode = terminate.unwrap_or_default().mode;
+pub(in crate::http) async fn terminate(State(compute): State<Arc<ComputeNode>>) -> Response {
     {
         let mut state = compute.state.lock().unwrap();
         if state.status == ComputeStatus::Terminated {
-            return JsonResponse::success(StatusCode::CREATED, state.terminate_flush_lsn);
+            return StatusCode::CREATED.into_response();
         }
 
         if !matches!(state.status, ComputeStatus::Empty | ComputeStatus::Running) {
             return JsonResponse::invalid_status(state.status);
         }
-        state.set_status(
-            ComputeStatus::TerminationPending { mode },
-            &compute.state_changed,
-        );
+
+        state.set_status(ComputeStatus::TerminationPending, &compute.state_changed);
+        drop(state);
     }
 
-    forward_termination_signal(false);
+    forward_termination_signal();
     info!("sent signal and notified waiters");
 
     // Spawn a blocking thread to wait for compute to become Terminated.
@@ -44,7 +34,7 @@ pub(in crate::http) async fn terminate(
     // be able to serve other requests while some particular request
     // is waiting for compute to finish configuration.
     let c = compute.clone();
-    let lsn = task::spawn_blocking(move || {
+    task::spawn_blocking(move || {
         let mut state = c.state.lock().unwrap();
         while state.status != ComputeStatus::Terminated {
             state = c.state_changed.wait(state).unwrap();
@@ -54,10 +44,11 @@ pub(in crate::http) async fn terminate(
                 state.status
             );
         }
-        state.terminate_flush_lsn
     })
     .await
     .unwrap();
+
     info!("terminated Postgres");
-    JsonResponse::success(StatusCode::OK, TerminateResponse { lsn })
+
+    StatusCode::OK.into_response()
 }

compute_tools/src/installed_extensions.rs

@@ -43,7 +43,7 @@ pub async fn get_installed_extensions(mut conf: Config) -> Result<InstalledExten
         let (mut client, connection) = conf.connect(NoTls).await?;
         tokio::spawn(async move {
             if let Err(e) = connection.await {
-                eprintln!("connection error: {e}");
+                eprintln!("connection error: {}", e);
             }
         });
 
@@ -57,7 +57,7 @@ pub async fn get_installed_extensions(mut conf: Config) -> Result<InstalledExten
         let (client, connection) = conf.connect(NoTls).await?;
         tokio::spawn(async move {
             if let Err(e) = connection.await {
-                eprintln!("connection error: {e}");
+                eprintln!("connection error: {}", e);
             }
         });
 

compute_tools/src/lib.rs

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

compute_tools/src/lsn_lease.rs

@@ -130,7 +130,7 @@ fn try_acquire_lsn_lease(
         lsn: Lsn,
     ) -> Result<Option<SystemTime>> {
         let mut client = config.connect(NoTls)?;
-        let cmd = format!("lease lsn {tenant_shard_id} {timeline_id} {lsn} ");
+        let cmd = format!("lease lsn {} {} {} ", tenant_shard_id, timeline_id, lsn);
         let res = client.simple_query(&cmd)?;
         let msg = match res.first() {
             Some(msg) => msg,

compute_tools/src/monitor.rs

@@ -83,9 +83,7 @@ impl ComputeMonitor {
         let compute_status = self.compute.get_status();
         if matches!(
             compute_status,
-            ComputeStatus::Terminated
-                | ComputeStatus::TerminationPending { .. }
-                | ComputeStatus::Failed
+            ComputeStatus::Terminated | ComputeStatus::TerminationPending | ComputeStatus::Failed
         ) {
             info!(
                 "compute is in {} status, stopping compute monitor",

compute_tools/src/pg_helpers.rs

@@ -36,9 +36,9 @@ pub fn escape_literal(s: &str) -> String {
     let res = s.replace('\'', "''").replace('\\', "\\\\");
 
     if res.contains('\\') {
-        format!("E'{res}'")
+        format!("E'{}'", res)
     } else {
-        format!("'{res}'")
+        format!("'{}'", res)
     }
 }
 
@@ -46,7 +46,7 @@ pub fn escape_literal(s: &str) -> String {
 /// with `'{}'` is not required, as it returns a ready-to-use config string.
 pub fn escape_conf_value(s: &str) -> String {
     let res = s.replace('\'', "''").replace('\\', "\\\\");
-    format!("'{res}'")
+    format!("'{}'", res)
 }
 
 pub trait GenericOptionExt {
@@ -446,7 +446,7 @@ pub async fn tune_pgbouncer(
         let mut pgbouncer_connstr =
             "host=localhost port=6432 dbname=pgbouncer user=postgres sslmode=disable".to_string();
         if let Ok(pass) = std::env::var("PGBOUNCER_PASSWORD") {
-            pgbouncer_connstr.push_str(format!(" password={pass}").as_str());
+            pgbouncer_connstr.push_str(format!(" password={}", pass).as_str());
         }
         pgbouncer_connstr
     };
@@ -464,7 +464,7 @@ pub async fn tune_pgbouncer(
             Ok((client, connection)) => {
                 tokio::spawn(async move {
                     if let Err(e) = connection.await {
-                        eprintln!("connection error: {e}");
+                        eprintln!("connection error: {}", e);
                     }
                 });
                 break client;

compute_tools/src/pgbouncer.rs

@@ -1 +0,0 @@
-pub const PGBOUNCER_PIDFILE: &str = "/tmp/pgbouncer.pid";

compute_tools/src/spec.rs

@@ -23,12 +23,12 @@ fn do_control_plane_request(
 ) -> Result<ControlPlaneConfigResponse, (bool, String, String)> {
     let resp = reqwest::blocking::Client::new()
         .get(uri)
-        .header("Authorization", format!("Bearer {jwt}"))
+        .header("Authorization", format!("Bearer {}", jwt))
         .send()
         .map_err(|e| {
             (
                 true,
-                format!("could not perform request to control plane: {e:?}"),
+                format!("could not perform request to control plane: {:?}", e),
                 UNKNOWN_HTTP_STATUS.to_string(),
             )
         })?;
@@ -39,7 +39,7 @@ fn do_control_plane_request(
             Ok(spec_resp) => Ok(spec_resp),
             Err(e) => Err((
                 true,
-                format!("could not deserialize control plane response: {e:?}"),
+                format!("could not deserialize control plane response: {:?}", e),
                 status.to_string(),
             )),
         },
@@ -62,7 +62,7 @@ fn do_control_plane_request(
         // or some internal failure happened. Doesn't make much sense to retry in this case.
         _ => Err((
             false,
-            format!("unexpected control plane response status code: {status}"),
+            format!("unexpected control plane response status code: {}", status),
             status.to_string(),
         )),
     }

compute_tools/src/spec_apply.rs

@@ -933,53 +933,56 @@ async fn get_operations<'a>(
                 PerDatabasePhase::DeleteDBRoleReferences => {
                     let ctx = ctx.read().await;
 
-                    let operations = spec
-                        .delta_operations
-                        .iter()
-                        .flatten()
-                        .filter(|op| op.action == "delete_role")
-                        .filter_map(move |op| {
-                            if db.is_owned_by(&op.name) {
-                                return None;
-                            }
-                            if !ctx.roles.contains_key(&op.name) {
-                                return None;
-                            }
-                            let quoted = op.name.pg_quote();
-                            let new_owner = match &db {
-                                DB::SystemDB => PgIdent::from("cloud_admin").pg_quote(),
-                                DB::UserDB(db) => db.owner.pg_quote(),
-                            };
-                            let (escaped_role, outer_tag) = op.name.pg_quote_dollar();
+                    let operations =
+                        spec.delta_operations
+                            .iter()
+                            .flatten()
+                            .filter(|op| op.action == "delete_role")
+                            .filter_map(move |op| {
+                                if db.is_owned_by(&op.name) {
+                                    return None;
+                                }
+                                if !ctx.roles.contains_key(&op.name) {
+                                    return None;
+                                }
+                                let quoted = op.name.pg_quote();
+                                let new_owner = match &db {
+                                    DB::SystemDB => PgIdent::from("cloud_admin").pg_quote(),
+                                    DB::UserDB(db) => db.owner.pg_quote(),
+                                };
+                                let (escaped_role, outer_tag) = op.name.pg_quote_dollar();
 
-                            Some(vec![
-                                // This will reassign all dependent objects to the db owner
-                                Operation {
-                                    query: format!("REASSIGN OWNED BY {quoted} TO {new_owner}",),
-                                    comment: None,
-                                },
-                                // Revoke some potentially blocking privileges (Neon-specific currently)
-                                Operation {
-                                    query: format!(
-                                        include_str!("sql/pre_drop_role_revoke_privileges.sql"),
-                                        // N.B. this has to be properly dollar-escaped with `pg_quote_dollar()`
-                                        role_name = escaped_role,
-                                        outer_tag = outer_tag,
-                                    ),
-                                    comment: None,
-                                },
-                                // This now will only drop privileges of the role
-                                // TODO: this is obviously not 100% true because of the above case,
-                                // there could be still some privileges that are not revoked. Maybe this
-                                // only drops privileges that were granted *by this* role, not *to this* role,
-                                // but this has to be checked.
-                                Operation {
-                                    query: format!("DROP OWNED BY {quoted}"),
-                                    comment: None,
-                                },
-                            ])
-                        })
-                        .flatten();
+                                Some(vec![
+                                    // This will reassign all dependent objects to the db owner
+                                    Operation {
+                                        query: format!(
+                                            "REASSIGN OWNED BY {} TO {}",
+                                            quoted, new_owner,
+                                        ),
+                                        comment: None,
+                                    },
+                                    // Revoke some potentially blocking privileges (Neon-specific currently)
+                                    Operation {
+                                        query: format!(
+                                            include_str!("sql/pre_drop_role_revoke_privileges.sql"),
+                                            // N.B. this has to be properly dollar-escaped with `pg_quote_dollar()`
+                                            role_name = escaped_role,
+                                            outer_tag = outer_tag,
+                                        ),
+                                        comment: None,
+                                    },
+                                    // This now will only drop privileges of the role
+                                    // TODO: this is obviously not 100% true because of the above case,
+                                    // there could be still some privileges that are not revoked. Maybe this
+                                    // only drops privileges that were granted *by this* role, not *to this* role,
+                                    // but this has to be checked.
+                                    Operation {
+                                        query: format!("DROP OWNED BY {}", quoted),
+                                        comment: None,
+                                    },
+                                ])
+                            })
+                            .flatten();
 
                     Ok(Box::new(operations))
                 }

compute_tools/src/sync_sk.rs

@@ -27,7 +27,7 @@ pub async fn ping_safekeeper(
     let (client, conn) = config.connect(tokio_postgres::NoTls).await?;
     tokio::spawn(async move {
         if let Err(e) = conn.await {
-            eprintln!("connection error: {e}");
+            eprintln!("connection error: {}", e);
         }
     });
 

compute_tools/tests/README.md

@@ -1,6 +0,0 @@
-### Test files
-
-The file `cluster_spec.json` has been copied over from libs/compute_api
-tests, with some edits:
-
-  - the neon.safekeepers setting contains a duplicate value

compute_tools/tests/cluster_spec.json

@@ -1,245 +0,0 @@
-{
-  "format_version": 1.0,
-
-  "timestamp": "2021-05-23T18:25:43.511Z",
-  "operation_uuid": "0f657b36-4b0f-4a2d-9c2e-1dcd615e7d8b",
-
-  "cluster": {
-    "cluster_id": "test-cluster-42",
-    "name": "Zenith Test",
-    "state": "restarted",
-    "roles": [
-      {
-        "name": "postgres",
-        "encrypted_password": "6b1d16b78004bbd51fa06af9eda75972",
-        "options": null
-      },
-      {
-        "name": "alexk",
-        "encrypted_password": null,
-        "options": null
-      },
-      {
-        "name": "zenith \"new\"",
-        "encrypted_password": "5b1d16b78004bbd51fa06af9eda75972",
-        "options": null
-      },
-      {
-        "name": "zen",
-        "encrypted_password": "9b1d16b78004bbd51fa06af9eda75972"
-      },
-      {
-        "name": "\"name\";\\n select 1;",
-        "encrypted_password": "5b1d16b78004bbd51fa06af9eda75972"
-      },
-      {
-        "name": "MyRole",
-        "encrypted_password": "5b1d16b78004bbd51fa06af9eda75972"
-      }
-    ],
-    "databases": [
-      {
-        "name": "DB2",
-        "owner": "alexk",
-        "options": [
-          {
-            "name": "LC_COLLATE",
-            "value": "C",
-            "vartype": "string"
-          },
-          {
-            "name": "LC_CTYPE",
-            "value": "C",
-            "vartype": "string"
-          },
-          {
-            "name": "TEMPLATE",
-            "value": "template0",
-            "vartype": "enum"
-          }
-        ]
-      },
-      {
-        "name": "zenith",
-        "owner": "MyRole"
-      },
-      {
-        "name": "zen",
-        "owner": "zen"
-      }
-    ],
-    "settings": [
-      {
-        "name": "fsync",
-        "value": "off",
-        "vartype": "bool"
-      },
-      {
-        "name": "wal_level",
-        "value": "logical",
-        "vartype": "enum"
-      },
-      {
-        "name": "hot_standby",
-        "value": "on",
-        "vartype": "bool"
-      },
-      {
-        "name": "prewarm_lfc_on_startup",
-        "value": "off",
-        "vartype": "bool"
-      },
-      {
-        "name": "neon.safekeepers",
-        "value": "127.0.0.1:6502,127.0.0.1:6503,127.0.0.1:6501,127.0.0.1:6502",
-        "vartype": "string"
-      },
-      {
-        "name": "wal_log_hints",
-        "value": "on",
-        "vartype": "bool"
-      },
-      {
-        "name": "log_connections",
-        "value": "on",
-        "vartype": "bool"
-      },
-      {
-        "name": "shared_buffers",
-        "value": "32768",
-        "vartype": "integer"
-      },
-      {
-        "name": "port",
-        "value": "55432",
-        "vartype": "integer"
-      },
-      {
-        "name": "max_connections",
-        "value": "100",
-        "vartype": "integer"
-      },
-      {
-        "name": "max_wal_senders",
-        "value": "10",
-        "vartype": "integer"
-      },
-      {
-        "name": "listen_addresses",
-        "value": "0.0.0.0",
-        "vartype": "string"
-      },
-      {
-        "name": "wal_sender_timeout",
-        "value": "0",
-        "vartype": "integer"
-      },
-      {
-        "name": "password_encryption",
-        "value": "md5",
-        "vartype": "enum"
-      },
-      {
-        "name": "maintenance_work_mem",
-        "value": "65536",
-        "vartype": "integer"
-      },
-      {
-        "name": "max_parallel_workers",
-        "value": "8",
-        "vartype": "integer"
-      },
-      {
-        "name": "max_worker_processes",
-        "value": "8",
-        "vartype": "integer"
-      },
-      {
-        "name": "neon.tenant_id",
-        "value": "b0554b632bd4d547a63b86c3630317e8",
-        "vartype": "string"
-      },
-      {
-        "name": "max_replication_slots",
-        "value": "10",
-        "vartype": "integer"
-      },
-      {
-        "name": "neon.timeline_id",
-        "value": "2414a61ffc94e428f14b5758fe308e13",
-        "vartype": "string"
-      },
-      {
-        "name": "shared_preload_libraries",
-        "value": "neon",
-        "vartype": "string"
-      },
-      {
-        "name": "synchronous_standby_names",
-        "value": "walproposer",
-        "vartype": "string"
-      },
-      {
-        "name": "neon.pageserver_connstring",
-        "value": "host=127.0.0.1 port=6400",
-        "vartype": "string"
-      },
-      {
-        "name": "test.escaping",
-        "value": "here's a backslash \\ and a quote ' and a double-quote \" hooray",
-        "vartype": "string"
-      }
-    ]
-  },
-  "delta_operations": [
-    {
-      "action": "delete_db",
-      "name": "zenith_test"
-    },
-    {
-      "action": "rename_db",
-      "name": "DB",
-      "new_name": "DB2"
-    },
-    {
-      "action": "delete_role",
-      "name": "zenith2"
-    },
-    {
-      "action": "rename_role",
-      "name": "zenith new",
-      "new_name": "zenith \"new\""
-    }
-  ],
-  "remote_extensions": {
-    "library_index": {
-      "postgis-3": "postgis",
-      "libpgrouting-3.4": "postgis",
-      "postgis_raster-3": "postgis",
-      "postgis_sfcgal-3": "postgis",
-      "postgis_topology-3": "postgis",
-      "address_standardizer-3": "postgis"
-    },
-    "extension_data": {
-      "postgis": {
-        "archive_path": "5834329303/v15/extensions/postgis.tar.zst",
-        "control_data": {
-          "postgis.control": "# postgis extension\ncomment = ''PostGIS geometry and geography spatial types and functions''\ndefault_version = ''3.3.2''\nmodule_pathname = ''$libdir/postgis-3''\nrelocatable = false\ntrusted = true\n",
-          "pgrouting.control": "# pgRouting Extension\ncomment = ''pgRouting Extension''\ndefault_version = ''3.4.2''\nmodule_pathname = ''$libdir/libpgrouting-3.4''\nrelocatable = true\nrequires = ''plpgsql''\nrequires = ''postgis''\ntrusted = true\n",
-          "postgis_raster.control": "# postgis_raster extension\ncomment = ''PostGIS raster types and functions''\ndefault_version = ''3.3.2''\nmodule_pathname = ''$libdir/postgis_raster-3''\nrelocatable = false\nrequires = postgis\ntrusted = true\n",
-          "postgis_sfcgal.control": "# postgis topology extension\ncomment = ''PostGIS SFCGAL functions''\ndefault_version = ''3.3.2''\nrelocatable = true\nrequires = postgis\ntrusted = true\n",
-          "postgis_topology.control": "# postgis topology extension\ncomment = ''PostGIS topology spatial types and functions''\ndefault_version = ''3.3.2''\nrelocatable = false\nschema = topology\nrequires = postgis\ntrusted = true\n",
-          "address_standardizer.control": "# address_standardizer extension\ncomment = ''Used to parse an address into constituent elements. Generally used to support geocoding address normalization step.''\ndefault_version = ''3.3.2''\nrelocatable = true\ntrusted = true\n",
-          "postgis_tiger_geocoder.control": "# postgis tiger geocoder extension\ncomment = ''PostGIS tiger geocoder and reverse geocoder''\ndefault_version = ''3.3.2''\nrelocatable = false\nschema = tiger\nrequires = ''postgis,fuzzystrmatch''\nsuperuser= false\ntrusted = true\n",
-          "address_standardizer_data_us.control": "# address standardizer us dataset\ncomment = ''Address Standardizer US dataset example''\ndefault_version = ''3.3.2''\nrelocatable = true\ntrusted = true\n"
-        }
-      }
-    },
-    "custom_extensions": [],
-    "public_extensions": ["postgis"]
-  },
-  "pgbouncer_settings": {
-    "default_pool_size": "42",
-    "pool_mode": "session"
-  }
-}

control_plane/src/bin/neon_local.rs

@@ -18,7 +18,7 @@ use clap::Parser;
 use compute_api::requests::ComputeClaimsScope;
 use compute_api::spec::ComputeMode;
 use control_plane::broker::StorageBroker;
-use control_plane::endpoint::{ComputeControlPlane, EndpointTerminateMode, PageserverProtocol};
+use control_plane::endpoint::ComputeControlPlane;
 use control_plane::endpoint_storage::{ENDPOINT_STORAGE_DEFAULT_ADDR, EndpointStorage};
 use control_plane::local_env;
 use control_plane::local_env::{
@@ -605,14 +605,6 @@ struct EndpointCreateCmdArgs {
     #[clap(long, help = "Postgres version")]
     pg_version: u32,
 
-    /// Use gRPC to communicate with Pageservers, by generating grpc:// connstrings.
-    ///
-    /// Specified on creation such that it's retained across reconfiguration and restarts.
-    ///
-    /// NB: not yet supported by computes.
-    #[clap(long)]
-    grpc: bool,
-
     #[clap(
         long,
         help = "If set, the node will be a hot replica on the specified timeline",
@@ -672,13 +664,6 @@ struct EndpointStartCmdArgs {
     #[clap(short = 't', long, value_parser= humantime::parse_duration, help = "timeout until we fail the command")]
     #[arg(default_value = "90s")]
     start_timeout: Duration,
-
-    #[clap(
-        long,
-        help = "Run in development mode, skipping VM-specific operations like process termination",
-        action = clap::ArgAction::SetTrue
-    )]
-    dev: bool,
 }
 
 #[derive(clap::Args)]
@@ -711,9 +696,10 @@ struct EndpointStopCmdArgs {
     )]
     destroy: bool,
 
-    #[clap(long, help = "Postgres shutdown mode")]
-    #[clap(default_value = "fast")]
-    mode: EndpointTerminateMode,
+    #[clap(long, help = "Postgres shutdown mode, passed to \"pg_ctl -m <mode>\"")]
+    #[arg(value_parser(["smart", "fast", "immediate"]))]
+    #[arg(default_value = "fast")]
+    mode: String,
 }
 
 #[derive(clap::Args)]
@@ -919,7 +905,7 @@ fn print_timeline(
             br_sym = "┗━";
         }
 
-        print!("{br_sym} @{ancestor_lsn}: ");
+        print!("{} @{}: ", br_sym, ancestor_lsn);
     }
 
     // Finally print a timeline id and name with new line
@@ -1465,7 +1451,6 @@ async fn handle_endpoint(subcmd: &EndpointCmd, env: &local_env::LocalEnv) -> Res
                 args.internal_http_port,
                 args.pg_version,
                 mode,
-                args.grpc,
                 !args.update_catalog,
                 false,
             )?;
@@ -1506,20 +1491,13 @@ async fn handle_endpoint(subcmd: &EndpointCmd, env: &local_env::LocalEnv) -> Res
 
             let (pageservers, stripe_size) = if let Some(pageserver_id) = pageserver_id {
                 let conf = env.get_pageserver_conf(pageserver_id).unwrap();
-                // Use gRPC if requested.
-                let pageserver = if endpoint.grpc {
-                    let grpc_addr = conf.listen_grpc_addr.as_ref().expect("bad config");
-                    let (host, port) = parse_host_port(grpc_addr)?;
-                    let port = port.unwrap_or(DEFAULT_PAGESERVER_GRPC_PORT);
-                    (PageserverProtocol::Grpc, host, port)
-                } else {
-                    let (host, port) = parse_host_port(&conf.listen_pg_addr)?;
-                    let port = port.unwrap_or(5432);
-                    (PageserverProtocol::Libpq, host, port)
-                };
-                // If caller is telling us what pageserver to use, this is not a tenant which is
-                // fully managed by storage controller, therefore not sharded.
-                (vec![pageserver], DEFAULT_STRIPE_SIZE)
+                let parsed = parse_host_port(&conf.listen_pg_addr).expect("Bad config");
+                (
+                    vec![(parsed.0, parsed.1.unwrap_or(5432))],
+                    // If caller is telling us what pageserver to use, this is not a tenant which is
+                    // full managed by storage controller, therefore not sharded.
+                    DEFAULT_STRIPE_SIZE,
+                )
             } else {
                 // Look up the currently attached location of the tenant, and its striping metadata,
                 // to pass these on to postgres.
@@ -1538,20 +1516,11 @@ async fn handle_endpoint(subcmd: &EndpointCmd, env: &local_env::LocalEnv) -> Res
                                 .await?;
                         }
 
-                        let pageserver = if endpoint.grpc {
-                            (
-                                PageserverProtocol::Grpc,
-                                Host::parse(&shard.listen_grpc_addr.expect("no gRPC address"))?,
-                                shard.listen_grpc_port.expect("no gRPC port"),
-                            )
-                        } else {
-                            (
-                                PageserverProtocol::Libpq,
-                                Host::parse(&shard.listen_pg_addr)?,
-                                shard.listen_pg_port,
-                            )
-                        };
-                        anyhow::Ok(pageserver)
+                        anyhow::Ok((
+                            Host::parse(&shard.listen_pg_addr)
+                                .expect("Storage controller reported bad hostname"),
+                            shard.listen_pg_port,
+                        ))
                     }),
                 )
                 .await?;
@@ -1596,7 +1565,6 @@ async fn handle_endpoint(subcmd: &EndpointCmd, env: &local_env::LocalEnv) -> Res
                     stripe_size.0 as usize,
                     args.create_test_user,
                     args.start_timeout,
-                    args.dev,
                 )
                 .await?;
         }
@@ -1607,19 +1575,11 @@ async fn handle_endpoint(subcmd: &EndpointCmd, env: &local_env::LocalEnv) -> Res
                 .get(endpoint_id.as_str())
                 .with_context(|| format!("postgres endpoint {endpoint_id} is not found"))?;
             let pageservers = if let Some(ps_id) = args.endpoint_pageserver_id {
-                let conf = env.get_pageserver_conf(ps_id)?;
-                // Use gRPC if requested.
-                let pageserver = if endpoint.grpc {
-                    let grpc_addr = conf.listen_grpc_addr.as_ref().expect("bad config");
-                    let (host, port) = parse_host_port(grpc_addr)?;
-                    let port = port.unwrap_or(DEFAULT_PAGESERVER_GRPC_PORT);
-                    (PageserverProtocol::Grpc, host, port)
-                } else {
-                    let (host, port) = parse_host_port(&conf.listen_pg_addr)?;
-                    let port = port.unwrap_or(5432);
-                    (PageserverProtocol::Libpq, host, port)
-                };
-                vec![pageserver]
+                let pageserver = PageServerNode::from_env(env, env.get_pageserver_conf(ps_id)?);
+                vec![(
+                    pageserver.pg_connection_config.host().clone(),
+                    pageserver.pg_connection_config.port(),
+                )]
             } else {
                 let storage_controller = StorageController::from_env(env);
                 storage_controller
@@ -1628,21 +1588,11 @@ async fn handle_endpoint(subcmd: &EndpointCmd, env: &local_env::LocalEnv) -> Res
                     .shards
                     .into_iter()
                     .map(|shard| {
-                        // Use gRPC if requested.
-                        if endpoint.grpc {
-                            (
-                                PageserverProtocol::Grpc,
-                                Host::parse(&shard.listen_grpc_addr.expect("no gRPC address"))
-                                    .expect("bad hostname"),
-                                shard.listen_grpc_port.expect("no gRPC port"),
-                            )
-                        } else {
-                            (
-                                PageserverProtocol::Libpq,
-                                Host::parse(&shard.listen_pg_addr).expect("bad hostname"),
-                                shard.listen_pg_port,
-                            )
-                        }
+                        (
+                            Host::parse(&shard.listen_pg_addr)
+                                .expect("Storage controller reported malformed host"),
+                            shard.listen_pg_port,
+                        )
                     })
                     .collect::<Vec<_>>()
             };
@@ -1657,10 +1607,7 @@ async fn handle_endpoint(subcmd: &EndpointCmd, env: &local_env::LocalEnv) -> Res
                 .endpoints
                 .get(endpoint_id)
                 .with_context(|| format!("postgres endpoint {endpoint_id} is not found"))?;
-            match endpoint.stop(args.mode, args.destroy).await?.lsn {
-                Some(lsn) => println!("{lsn}"),
-                None => println!("null"),
-            }
+            endpoint.stop(&args.mode, args.destroy)?;
         }
         EndpointCmd::GenerateJwt(args) => {
             let endpoint = {
@@ -1742,7 +1689,7 @@ async fn handle_pageserver(subcmd: &PageserverCmd, env: &local_env::LocalEnv) ->
                 StopMode::Immediate => true,
             };
             if let Err(e) = get_pageserver(env, args.pageserver_id)?.stop(immediate) {
-                eprintln!("pageserver stop failed: {e}");
+                eprintln!("pageserver stop failed: {}", e);
                 exit(1);
             }
         }
@@ -1751,7 +1698,7 @@ async fn handle_pageserver(subcmd: &PageserverCmd, env: &local_env::LocalEnv) ->
             let pageserver = get_pageserver(env, args.pageserver_id)?;
             //TODO what shutdown strategy should we use here?
             if let Err(e) = pageserver.stop(false) {
-                eprintln!("pageserver stop failed: {e}");
+                eprintln!("pageserver stop failed: {}", e);
                 exit(1);
             }
 
@@ -1768,7 +1715,7 @@ async fn handle_pageserver(subcmd: &PageserverCmd, env: &local_env::LocalEnv) ->
             {
                 Ok(_) => println!("Page server is up and running"),
                 Err(err) => {
-                    eprintln!("Page server is not available: {err}");
+                    eprintln!("Page server is not available: {}", err);
                     exit(1);
                 }
             }
@@ -1805,7 +1752,7 @@ async fn handle_storage_controller(
                 },
             };
             if let Err(e) = svc.stop(stop_args).await {
-                eprintln!("stop failed: {e}");
+                eprintln!("stop failed: {}", e);
                 exit(1);
             }
         }
@@ -1827,7 +1774,7 @@ async fn handle_safekeeper(subcmd: &SafekeeperCmd, env: &local_env::LocalEnv) ->
             let safekeeper = get_safekeeper(env, args.id)?;
 
             if let Err(e) = safekeeper.start(&args.extra_opt, &args.start_timeout).await {
-                eprintln!("safekeeper start failed: {e}");
+                eprintln!("safekeeper start failed: {}", e);
                 exit(1);
             }
         }
@@ -1839,7 +1786,7 @@ async fn handle_safekeeper(subcmd: &SafekeeperCmd, env: &local_env::LocalEnv) ->
                 StopMode::Immediate => true,
             };
             if let Err(e) = safekeeper.stop(immediate) {
-                eprintln!("safekeeper stop failed: {e}");
+                eprintln!("safekeeper stop failed: {}", e);
                 exit(1);
             }
         }
@@ -1852,12 +1799,12 @@ async fn handle_safekeeper(subcmd: &SafekeeperCmd, env: &local_env::LocalEnv) ->
             };
 
             if let Err(e) = safekeeper.stop(immediate) {
-                eprintln!("safekeeper stop failed: {e}");
+                eprintln!("safekeeper stop failed: {}", e);
                 exit(1);
             }
 
             if let Err(e) = safekeeper.start(&args.extra_opt, &args.start_timeout).await {
-                eprintln!("safekeeper start failed: {e}");
+                eprintln!("safekeeper start failed: {}", e);
                 exit(1);
             }
         }
@@ -2092,16 +2039,11 @@ async fn handle_stop_all(args: &StopCmdArgs, env: &local_env::LocalEnv) -> Resul
 }
 
 async fn try_stop_all(env: &local_env::LocalEnv, immediate: bool) {
-    let mode = if immediate {
-        EndpointTerminateMode::Immediate
-    } else {
-        EndpointTerminateMode::Fast
-    };
     // Stop all endpoints
     match ComputeControlPlane::load(env.clone()) {
         Ok(cplane) => {
             for (_k, node) in cplane.endpoints {
-                if let Err(e) = node.stop(mode, false).await {
+                if let Err(e) = node.stop(if immediate { "immediate" } else { "fast" }, false) {
                     eprintln!("postgres stop failed: {e:#}");
                 }
             }
@@ -2113,7 +2055,7 @@ async fn try_stop_all(env: &local_env::LocalEnv, immediate: bool) {
 
     let storage = EndpointStorage::from_env(env);
     if let Err(e) = storage.stop(immediate) {
-        eprintln!("endpoint_storage stop failed: {e:#}");
+        eprintln!("endpoint_storage stop failed: {:#}", e);
     }
 
     for ps_conf in &env.pageservers {

control_plane/src/endpoint.rs

@@ -37,7 +37,6 @@
 //! ```
 //!
 use std::collections::BTreeMap;
-use std::fmt::Display;
 use std::net::{IpAddr, Ipv4Addr, SocketAddr, TcpStream};
 use std::path::PathBuf;
 use std::process::Command;
@@ -46,14 +45,11 @@ use std::sync::Arc;
 use std::time::{Duration, Instant};
 
 use anyhow::{Context, Result, anyhow, bail};
-use base64::Engine;
-use base64::prelude::BASE64_URL_SAFE_NO_PAD;
 use compute_api::requests::{
     COMPUTE_AUDIENCE, ComputeClaims, ComputeClaimsScope, ConfigurationRequest,
 };
 use compute_api::responses::{
-    ComputeConfig, ComputeCtlConfig, ComputeStatus, ComputeStatusResponse, TerminateResponse,
-    TlsConfig,
+    ComputeConfig, ComputeCtlConfig, ComputeStatus, ComputeStatusResponse, TlsConfig,
 };
 use compute_api::spec::{
     Cluster, ComputeAudit, ComputeFeature, ComputeMode, ComputeSpec, Database, PgIdent,
@@ -78,6 +74,7 @@ use utils::id::{NodeId, TenantId, TimelineId};
 
 use crate::local_env::LocalEnv;
 use crate::postgresql_conf::PostgresConf;
+use crate::storage_controller::StorageController;
 
 // contents of a endpoint.json file
 #[derive(Serialize, Deserialize, PartialEq, Eq, Clone, Debug)]
@@ -90,7 +87,6 @@ pub struct EndpointConf {
     external_http_port: u16,
     internal_http_port: u16,
     pg_version: u32,
-    grpc: bool,
     skip_pg_catalog_updates: bool,
     reconfigure_concurrency: usize,
     drop_subscriptions_before_start: bool,
@@ -168,7 +164,7 @@ impl ComputeControlPlane {
                     public_key_use: Some(PublicKeyUse::Signature),
                     key_operations: Some(vec![KeyOperations::Verify]),
                     key_algorithm: Some(KeyAlgorithm::EdDSA),
-                    key_id: Some(BASE64_URL_SAFE_NO_PAD.encode(key_hash)),
+                    key_id: Some(base64::encode_config(key_hash, base64::URL_SAFE_NO_PAD)),
                     x509_url: None::<String>,
                     x509_chain: None::<Vec<String>>,
                     x509_sha1_fingerprint: None::<String>,
@@ -177,7 +173,7 @@ impl ComputeControlPlane {
                 algorithm: AlgorithmParameters::OctetKeyPair(OctetKeyPairParameters {
                     key_type: OctetKeyPairType::OctetKeyPair,
                     curve: EllipticCurve::Ed25519,
-                    x: BASE64_URL_SAFE_NO_PAD.encode(public_key),
+                    x: base64::encode_config(public_key, base64::URL_SAFE_NO_PAD),
                 }),
             }],
         })
@@ -194,7 +190,6 @@ impl ComputeControlPlane {
         internal_http_port: Option<u16>,
         pg_version: u32,
         mode: ComputeMode,
-        grpc: bool,
         skip_pg_catalog_updates: bool,
         drop_subscriptions_before_start: bool,
     ) -> Result<Arc<Endpoint>> {
@@ -229,7 +224,6 @@ impl ComputeControlPlane {
             // we also skip catalog updates in the cloud.
             skip_pg_catalog_updates,
             drop_subscriptions_before_start,
-            grpc,
             reconfigure_concurrency: 1,
             features: vec![],
             cluster: None,
@@ -248,7 +242,6 @@ impl ComputeControlPlane {
                 internal_http_port,
                 pg_port,
                 pg_version,
-                grpc,
                 skip_pg_catalog_updates,
                 drop_subscriptions_before_start,
                 reconfigure_concurrency: 1,
@@ -303,8 +296,6 @@ pub struct Endpoint {
     pub tenant_id: TenantId,
     pub timeline_id: TimelineId,
     pub mode: ComputeMode,
-    /// If true, the endpoint should use gRPC to communicate with Pageservers.
-    pub grpc: bool,
 
     // port and address of the Postgres server and `compute_ctl`'s HTTP APIs
     pub pg_address: SocketAddr,
@@ -340,58 +331,15 @@ pub enum EndpointStatus {
     RunningNoPidfile,
 }
 
-impl Display for EndpointStatus {
+impl std::fmt::Display for EndpointStatus {
     fn fmt(&self, writer: &mut std::fmt::Formatter) -> std::fmt::Result {
-        writer.write_str(match self {
+        let s = match self {
             Self::Running => "running",
             Self::Stopped => "stopped",
             Self::Crashed => "crashed",
             Self::RunningNoPidfile => "running, no pidfile",
-        })
-    }
-}
-
-#[derive(Default, Clone, Copy, clap::ValueEnum)]
-pub enum EndpointTerminateMode {
-    #[default]
-    /// Use pg_ctl stop -m fast
-    Fast,
-    /// Use pg_ctl stop -m immediate
-    Immediate,
-    /// Use /terminate?mode=immediate
-    ImmediateTerminate,
-}
-
-impl std::fmt::Display for EndpointTerminateMode {
-    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
-        f.write_str(match &self {
-            EndpointTerminateMode::Fast => "fast",
-            EndpointTerminateMode::Immediate => "immediate",
-            EndpointTerminateMode::ImmediateTerminate => "immediate-terminate",
-        })
-    }
-}
-
-/// Protocol used to connect to a Pageserver.
-#[derive(Clone, Copy, Debug)]
-pub enum PageserverProtocol {
-    Libpq,
-    Grpc,
-}
-
-impl PageserverProtocol {
-    /// Returns the URL scheme for the protocol, used in connstrings.
-    pub fn scheme(&self) -> &'static str {
-        match self {
-            Self::Libpq => "postgresql",
-            Self::Grpc => "grpc",
-        }
-    }
-}
-
-impl Display for PageserverProtocol {
-    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
-        f.write_str(self.scheme())
+        };
+        write!(writer, "{}", s)
     }
 }
 
@@ -430,7 +378,6 @@ impl Endpoint {
             mode: conf.mode,
             tenant_id: conf.tenant_id,
             pg_version: conf.pg_version,
-            grpc: conf.grpc,
             skip_pg_catalog_updates: conf.skip_pg_catalog_updates,
             reconfigure_concurrency: conf.reconfigure_concurrency,
             drop_subscriptions_before_start: conf.drop_subscriptions_before_start,
@@ -659,10 +606,10 @@ impl Endpoint {
         }
     }
 
-    fn build_pageserver_connstr(pageservers: &[(PageserverProtocol, Host, u16)]) -> String {
+    fn build_pageserver_connstr(pageservers: &[(Host, u16)]) -> String {
         pageservers
             .iter()
-            .map(|(scheme, host, port)| format!("{scheme}://no_user@{host}:{port}"))
+            .map(|(host, port)| format!("postgresql://no_user@{host}:{port}"))
             .collect::<Vec<_>>()
             .join(",")
     }
@@ -707,12 +654,11 @@ impl Endpoint {
         endpoint_storage_addr: String,
         safekeepers_generation: Option<SafekeeperGeneration>,
         safekeepers: Vec<NodeId>,
-        pageservers: Vec<(PageserverProtocol, Host, u16)>,
+        pageservers: Vec<(Host, u16)>,
         remote_ext_base_url: Option<&String>,
         shard_stripe_size: usize,
         create_test_user: bool,
         start_timeout: Duration,
-        dev: bool,
     ) -> Result<()> {
         if self.status() == EndpointStatus::Running {
             anyhow::bail!("The endpoint is already running");
@@ -846,10 +792,10 @@ impl Endpoint {
 
         // Launch compute_ctl
         let conn_str = self.connstr("cloud_admin", "postgres");
-        println!("Starting postgres node at '{conn_str}'");
+        println!("Starting postgres node at '{}'", conn_str);
         if create_test_user {
             let conn_str = self.connstr("test", "neondb");
-            println!("Also at '{conn_str}'");
+            println!("Also at '{}'", conn_str);
         }
         let mut cmd = Command::new(self.env.neon_distrib_dir.join("compute_ctl"));
         cmd.args([
@@ -883,10 +829,6 @@ impl Endpoint {
             cmd.args(["--remote-ext-base-url", remote_ext_base_url]);
         }
 
-        if dev {
-            cmd.arg("--dev");
-        }
-
         let child = cmd.spawn()?;
         // set up a scopeguard to kill & wait for the child in case we panic or bail below
         let child = scopeguard::guard(child, |mut child| {
@@ -939,7 +881,7 @@ impl Endpoint {
                         ComputeStatus::Empty
                         | ComputeStatus::ConfigurationPending
                         | ComputeStatus::Configuration
-                        | ComputeStatus::TerminationPending { .. }
+                        | ComputeStatus::TerminationPending
                         | ComputeStatus::Terminated => {
                             bail!("unexpected compute status: {:?}", state.status)
                         }
@@ -948,7 +890,8 @@ impl Endpoint {
                 Err(e) => {
                     if Instant::now().duration_since(start_at) > start_timeout {
                         return Err(e).context(format!(
-                            "timed out {start_timeout:?} waiting to connect to compute_ctl HTTP",
+                            "timed out {:?} waiting to connect to compute_ctl HTTP",
+                            start_timeout,
                         ));
                     }
                 }
@@ -987,7 +930,7 @@ impl Endpoint {
             // reqwest does not export its error construction utility functions, so let's craft the message ourselves
             let url = response.url().to_owned();
             let msg = match response.text().await {
-                Ok(err_body) => format!("Error: {err_body}"),
+                Ok(err_body) => format!("Error: {}", err_body),
                 Err(_) => format!("Http error ({}) at {}.", status.as_u16(), url),
             };
             Err(anyhow::anyhow!(msg))
@@ -996,12 +939,10 @@ impl Endpoint {
 
     pub async fn reconfigure(
         &self,
-        pageservers: Vec<(PageserverProtocol, Host, u16)>,
+        mut pageservers: Vec<(Host, u16)>,
         stripe_size: Option<ShardStripeSize>,
         safekeepers: Option<Vec<NodeId>>,
     ) -> Result<()> {
-        anyhow::ensure!(!pageservers.is_empty(), "no pageservers provided");
-
         let (mut spec, compute_ctl_config) = {
             let config_path = self.endpoint_path().join("config.json");
             let file = std::fs::File::open(config_path)?;
@@ -1013,7 +954,25 @@ impl Endpoint {
         let postgresql_conf = self.read_postgresql_conf()?;
         spec.cluster.postgresql_conf = Some(postgresql_conf);
 
+        // If we weren't given explicit pageservers, query the storage controller
+        if pageservers.is_empty() {
+            let storage_controller = StorageController::from_env(&self.env);
+            let locate_result = storage_controller.tenant_locate(self.tenant_id).await?;
+            pageservers = locate_result
+                .shards
+                .into_iter()
+                .map(|shard| {
+                    (
+                        Host::parse(&shard.listen_pg_addr)
+                            .expect("Storage controller reported bad hostname"),
+                        shard.listen_pg_port,
+                    )
+                })
+                .collect::<Vec<_>>();
+        }
+
         let pageserver_connstr = Self::build_pageserver_connstr(&pageservers);
+        assert!(!pageserver_connstr.is_empty());
         spec.pageserver_connstring = Some(pageserver_connstr);
         if stripe_size.is_some() {
             spec.shard_stripe_size = stripe_size.map(|s| s.0 as usize);
@@ -1053,34 +1012,15 @@ impl Endpoint {
         } else {
             let url = response.url().to_owned();
             let msg = match response.text().await {
-                Ok(err_body) => format!("Error: {err_body}"),
+                Ok(err_body) => format!("Error: {}", err_body),
                 Err(_) => format!("Http error ({}) at {}.", status.as_u16(), url),
             };
             Err(anyhow::anyhow!(msg))
         }
     }
 
-    pub async fn stop(
-        &self,
-        mode: EndpointTerminateMode,
-        destroy: bool,
-    ) -> Result<TerminateResponse> {
-        // pg_ctl stop is fast but doesn't allow us to collect LSN. /terminate is
-        // slow, and test runs time out. Solution: special mode "immediate-terminate"
-        // which uses /terminate
-        let response = if let EndpointTerminateMode::ImmediateTerminate = mode {
-            let ip = self.external_http_address.ip();
-            let port = self.external_http_address.port();
-            let url = format!("http://{ip}:{port}/terminate?mode=immediate");
-            let token = self.generate_jwt(Some(ComputeClaimsScope::Admin))?;
-            let request = reqwest::Client::new().post(url).bearer_auth(token);
-            let response = request.send().await.context("/terminate")?;
-            let text = response.text().await.context("/terminate result")?;
-            serde_json::from_str(&text).with_context(|| format!("deserializing {text}"))?
-        } else {
-            self.pg_ctl(&["-m", &mode.to_string(), "stop"], &None)?;
-            TerminateResponse { lsn: None }
-        };
+    pub fn stop(&self, mode: &str, destroy: bool) -> Result<()> {
+        self.pg_ctl(&["-m", mode, "stop"], &None)?;
 
         // Also wait for the compute_ctl process to die. It might have some
         // cleanup work to do after postgres stops, like syncing safekeepers,
@@ -1090,7 +1030,7 @@ impl Endpoint {
         // waiting. Sometimes we do *not* want this cleanup: tests intentionally
         // do stop when majority of safekeepers is down, so sync-safekeepers
         // would hang otherwise. This could be a separate flag though.
-        let send_sigterm = destroy || !matches!(mode, EndpointTerminateMode::Fast);
+        let send_sigterm = destroy || mode == "immediate";
         self.wait_for_compute_ctl_to_exit(send_sigterm)?;
         if destroy {
             println!(
@@ -1099,7 +1039,7 @@ impl Endpoint {
             );
             std::fs::remove_dir_all(self.endpoint_path())?;
         }
-        Ok(response)
+        Ok(())
     }
 
     pub fn connstr(&self, user: &str, db_name: &str) -> String {

control_plane/src/local_env.rs

@@ -209,10 +209,6 @@ pub struct NeonStorageControllerConf {
     pub use_https_safekeeper_api: bool,
 
     pub use_local_compute_notifications: bool,
-
-    pub timeline_safekeeper_count: Option<i64>,
-
-    pub kick_secondary_downloads: Option<bool>,
 }
 
 impl NeonStorageControllerConf {
@@ -240,11 +236,9 @@ impl Default for NeonStorageControllerConf {
             heartbeat_interval: Self::DEFAULT_HEARTBEAT_INTERVAL,
             long_reconcile_threshold: None,
             use_https_pageserver_api: false,
-            timelines_onto_safekeepers: true,
+            timelines_onto_safekeepers: false,
             use_https_safekeeper_api: false,
             use_local_compute_notifications: true,
-            timeline_safekeeper_count: None,
-            kick_secondary_downloads: None,
         }
     }
 }
@@ -260,7 +254,7 @@ impl Default for EndpointStorageConf {
 impl NeonBroker {
     pub fn client_url(&self) -> Url {
         let url = if let Some(addr) = self.listen_https_addr {
-            format!("https://{addr}")
+            format!("https://{}", addr)
         } else {
             format!(
                 "http://{}",
@@ -733,7 +727,7 @@ impl LocalEnv {
                 let config_toml_path = dentry.path().join("pageserver.toml");
                 let config_toml: PageserverConfigTomlSubset = toml_edit::de::from_str(
                     &std::fs::read_to_string(&config_toml_path)
-                        .with_context(|| format!("read {config_toml_path:?}"))?,
+                        .with_context(|| format!("read {:?}", config_toml_path))?,
                 )
                 .context("parse pageserver.toml")?;
                 let identity_toml_path = dentry.path().join("identity.toml");
@@ -743,7 +737,7 @@ impl LocalEnv {
                 }
                 let identity_toml: IdentityTomlSubset = toml_edit::de::from_str(
                     &std::fs::read_to_string(&identity_toml_path)
-                        .with_context(|| format!("read {identity_toml_path:?}"))?,
+                        .with_context(|| format!("read {:?}", identity_toml_path))?,
                 )
                 .context("parse identity.toml")?;
                 let PageserverConfigTomlSubset {

control_plane/src/pageserver.rs

@@ -16,7 +16,6 @@ use std::time::Duration;
 
 use anyhow::{Context, bail};
 use camino::Utf8PathBuf;
-use pageserver_api::config::{DEFAULT_GRPC_LISTEN_PORT, DEFAULT_HTTP_LISTEN_PORT};
 use pageserver_api::models::{self, TenantInfo, TimelineInfo};
 use pageserver_api::shard::TenantShardId;
 use pageserver_client::mgmt_api;
@@ -121,7 +120,7 @@ impl PageServerNode {
                 .env
                 .generate_auth_token(&Claims::new(None, Scope::GenerationsApi))
                 .unwrap();
-            overrides.push(format!("control_plane_api_token='{jwt_token}'"));
+            overrides.push(format!("control_plane_api_token='{}'", jwt_token));
         }
 
         if !conf.other.contains_key("remote_storage") {
@@ -253,10 +252,9 @@ impl PageServerNode {
         // the storage controller
         let metadata_path = datadir.join("metadata.json");
 
-        let http_host = "localhost".to_string();
-        let (_, http_port) =
+        let (_http_host, http_port) =
             parse_host_port(&self.conf.listen_http_addr).expect("Unable to parse listen_http_addr");
-        let http_port = http_port.unwrap_or(DEFAULT_HTTP_LISTEN_PORT);
+        let http_port = http_port.unwrap_or(9898);
 
         let https_port = match self.conf.listen_https_addr.as_ref() {
             Some(https_addr) => {
@@ -267,13 +265,6 @@ impl PageServerNode {
             None => None,
         };
 
-        let (mut grpc_host, mut grpc_port) = (None, None);
-        if let Some(grpc_addr) = &self.conf.listen_grpc_addr {
-            let (_, port) = parse_host_port(grpc_addr).expect("Unable to parse listen_grpc_addr");
-            grpc_host = Some("localhost".to_string());
-            grpc_port = Some(port.unwrap_or(DEFAULT_GRPC_LISTEN_PORT));
-        }
-
         // Intentionally hand-craft JSON: this acts as an implicit format compat test
         // in case the pageserver-side structure is edited, and reflects the real life
         // situation: the metadata is written by some other script.
@@ -282,9 +273,7 @@ impl PageServerNode {
             serde_json::to_vec(&pageserver_api::config::NodeMetadata {
                 postgres_host: "localhost".to_string(),
                 postgres_port: self.pg_connection_config.port(),
-                grpc_host,
-                grpc_port,
-                http_host,
+                http_host: "localhost".to_string(),
                 http_port,
                 https_port,
                 other: HashMap::from([(

control_plane/src/safekeeper.rs

@@ -143,7 +143,7 @@ impl SafekeeperNode {
         let id_string = id.to_string();
         // TODO: add availability_zone to the config.
         // Right now we just specify any value here and use it to check metrics in tests.
-        let availability_zone = format!("sk-{id_string}");
+        let availability_zone = format!("sk-{}", id_string);
 
         let mut args = vec![
             "-D".to_owned(),

control_plane/src/storage_controller.rs

@@ -167,7 +167,7 @@ impl StorageController {
     fn storage_controller_instance_dir(&self, instance_id: u8) -> PathBuf {
         self.env
             .base_data_dir
-            .join(format!("storage_controller_{instance_id}"))
+            .join(format!("storage_controller_{}", instance_id))
     }
 
     fn pid_file(&self, instance_id: u8) -> Utf8PathBuf {
@@ -220,7 +220,7 @@ impl StorageController {
             "-d",
             DB_NAME,
             "-p",
-            &format!("{postgres_port}"),
+            &format!("{}", postgres_port),
         ];
         let pg_lib_dir = self.get_pg_lib_dir().await.unwrap();
         let envs = [
@@ -263,7 +263,7 @@ impl StorageController {
                 "-h",
                 "localhost",
                 "-p",
-                &format!("{postgres_port}"),
+                &format!("{}", postgres_port),
                 "-U",
                 &username(),
                 "-O",
@@ -425,7 +425,7 @@ impl StorageController {
             // from `LocalEnv`'s config file (`.neon/config`).
             tokio::fs::write(
                 &pg_data_path.join("postgresql.conf"),
-                format!("port = {postgres_port}\nfsync=off\n"),
+                format!("port = {}\nfsync=off\n", postgres_port),
             )
             .await?;
 
@@ -477,7 +477,7 @@ impl StorageController {
             self.setup_database(postgres_port).await?;
         }
 
-        let database_url = format!("postgresql://localhost:{postgres_port}/{DB_NAME}");
+        let database_url = format!("postgresql://localhost:{}/{DB_NAME}", postgres_port);
 
         // We support running a startup SQL script to fiddle with the database before we launch storcon.
         // This is used by the test suite.
@@ -508,7 +508,7 @@ impl StorageController {
         drop(client);
         conn.await??;
 
-        let addr = format!("{host}:{listen_port}");
+        let addr = format!("{}:{}", host, listen_port);
         let address_for_peers = Uri::builder()
             .scheme(scheme)
             .authority(addr.clone())
@@ -557,10 +557,6 @@ impl StorageController {
             args.push("--use-local-compute-notifications".to_string());
         }
 
-        if let Some(value) = self.config.kick_secondary_downloads {
-            args.push(format!("--kick-secondary-downloads={value}"));
-        }
-
         if let Some(ssl_ca_file) = self.env.ssl_ca_cert_path() {
             args.push(format!("--ssl-ca-file={}", ssl_ca_file.to_str().unwrap()));
         }
@@ -632,10 +628,6 @@ impl StorageController {
             args.push("--timelines-onto-safekeepers".to_string());
         }
 
-        if let Some(sk_cnt) = self.config.timeline_safekeeper_count {
-            args.push(format!("--timeline-safekeeper-count={sk_cnt}"));
-        }
-
         println!("Starting storage controller");
 
         background_process::start_process(
@@ -810,9 +802,9 @@ impl StorageController {
             builder = builder.json(&body)
         }
         if let Some(private_key) = &self.private_key {
-            println!("Getting claims for path {path}");
+            println!("Getting claims for path {}", path);
             if let Some(required_claims) = Self::get_claims_for_path(&path)? {
-                println!("Got claims {required_claims:?} for path {path}");
+                println!("Got claims {:?} for path {}", required_claims, path);
                 let jwt_token = encode_from_key_file(&required_claims, private_key)?;
                 builder = builder.header(
                     reqwest::header::AUTHORIZATION,

control_plane/storcon_cli/src/main.rs

@@ -36,10 +36,6 @@ enum Command {
         listen_pg_addr: String,
         #[arg(long)]
         listen_pg_port: u16,
-        #[arg(long)]
-        listen_grpc_addr: Option<String>,
-        #[arg(long)]
-        listen_grpc_port: Option<u16>,
 
         #[arg(long)]
         listen_http_addr: String,
@@ -422,8 +418,6 @@ async fn main() -> anyhow::Result<()> {
             node_id,
             listen_pg_addr,
             listen_pg_port,
-            listen_grpc_addr,
-            listen_grpc_port,
             listen_http_addr,
             listen_http_port,
             listen_https_port,
@@ -437,8 +431,6 @@ async fn main() -> anyhow::Result<()> {
                         node_id,
                         listen_pg_addr,
                         listen_pg_port,
-                        listen_grpc_addr,
-                        listen_grpc_port,
                         listen_http_addr,
                         listen_http_port,
                         listen_https_port,
@@ -649,7 +641,7 @@ async fn main() -> anyhow::Result<()> {
                 response
                     .new_shards
                     .iter()
-                    .map(|s| format!("{s:?}"))
+                    .map(|s| format!("{:?}", s))
                     .collect::<Vec<_>>()
                     .join(",")
             );
@@ -771,8 +763,8 @@ async fn main() -> anyhow::Result<()> {
 
             println!("Tenant {tenant_id}");
             let mut table = comfy_table::Table::new();
-            table.add_row(["Policy", &format!("{policy:?}")]);
-            table.add_row(["Stripe size", &format!("{stripe_size:?}")]);
+            table.add_row(["Policy", &format!("{:?}", policy)]);
+            table.add_row(["Stripe size", &format!("{:?}", stripe_size)]);
             table.add_row(["Config", &serde_json::to_string_pretty(&config).unwrap()]);
             println!("{table}");
             println!("Shards:");
@@ -789,7 +781,7 @@ async fn main() -> anyhow::Result<()> {
                 let secondary = shard
                     .node_secondary
                     .iter()
-                    .map(|n| format!("{n}"))
+                    .map(|n| format!("{}", n))
                     .collect::<Vec<_>>()
                     .join(",");
 
@@ -863,7 +855,7 @@ async fn main() -> anyhow::Result<()> {
                 }
             } else {
                 // Make it obvious to the user that since they've omitted an AZ, we're clearing it
-                eprintln!("Clearing preferred AZ for tenant {tenant_id}");
+                eprintln!("Clearing preferred AZ for tenant {}", tenant_id);
             }
 
             // Construct a request that modifies all the tenant's shards
@@ -1134,7 +1126,8 @@ async fn main() -> anyhow::Result<()> {
                     Err((tenant_shard_id, from, to, error)) => {
                         failure += 1;
                         println!(
-                            "Failed to migrate {tenant_shard_id} from node {from} to node {to}: {error}"
+                            "Failed to migrate {} from node {} to node {}: {}",
+                            tenant_shard_id, from, to, error
                         );
                     }
                 }
@@ -1276,7 +1269,8 @@ async fn main() -> anyhow::Result<()> {
             concurrency,
         } => {
             let mut path = format!(
-                "/v1/tenant/{tenant_shard_id}/timeline/{timeline_id}/download_heatmap_layers",
+                "/v1/tenant/{}/timeline/{}/download_heatmap_layers",
+                tenant_shard_id, timeline_id,
             );
 
             if let Some(c) = concurrency {
@@ -1301,7 +1295,8 @@ async fn watch_tenant_shard(
 ) -> anyhow::Result<()> {
     if let Some(until_migrated_to) = until_migrated_to {
         println!(
-            "Waiting for tenant shard {tenant_shard_id} to be migrated to node {until_migrated_to}"
+            "Waiting for tenant shard {} to be migrated to node {}",
+            tenant_shard_id, until_migrated_to
         );
     }
 
@@ -1324,7 +1319,7 @@ async fn watch_tenant_shard(
             "attached: {} secondary: {} {}",
             shard
                 .node_attached
-                .map(|n| format!("{n}"))
+                .map(|n| format!("{}", n))
                 .unwrap_or("none".to_string()),
             shard
                 .node_secondary
@@ -1338,12 +1333,15 @@ async fn watch_tenant_shard(
                 "(reconciler idle)"
             }
         );
-        println!("{summary}");
+        println!("{}", summary);
 
         // Maybe drop out if we finished migration
         if let Some(until_migrated_to) = until_migrated_to {
             if shard.node_attached == Some(until_migrated_to) && !shard.is_reconciling {
-                println!("Tenant shard {tenant_shard_id} is now on node {until_migrated_to}");
+                println!(
+                    "Tenant shard {} is now on node {}",
+                    tenant_shard_id, until_migrated_to
+                );
                 break;
             }
         }

docker-compose/compute_wrapper/shell/compute.sh

@@ -95,4 +95,3 @@ echo "Start compute node"
      -b /usr/local/bin/postgres                              \
      --compute-id "compute-${RANDOM}"                          \
      --config "${CONFIG_FILE}"
-     --dev

endpoint_storage/src/app.rs

@@ -374,7 +374,7 @@ MC4CAQAwBQYDK2VwBCIEID/Drmc1AA6U/znNRWpF3zEGegOATQxfkdWxitcOMsIH
             let request = Request::builder()
                 .uri(format!("/{tenant}/{timeline}/{endpoint}/sub/path/key"))
                 .method(method)
-                .header("Authorization", format!("Bearer {token}"))
+                .header("Authorization", format!("Bearer {}", token))
                 .body(Body::empty())
                 .unwrap();
             let status = ServiceExt::ready(&mut app)

endpoint_storage/src/main.rs

@@ -31,12 +31,13 @@ struct Args {
 }
 
 #[derive(serde::Deserialize)]
+#[serde(tag = "type")]
 struct Config {
     #[serde(default = "listen")]
     listen: std::net::SocketAddr,
     pemfile: camino::Utf8PathBuf,
     #[serde(flatten)]
-    storage_kind: remote_storage::TypedRemoteStorageKind,
+    storage_config: remote_storage::RemoteStorageConfig,
     #[serde(default = "max_upload_file_limit")]
     max_upload_file_limit: usize,
 }
@@ -69,8 +70,7 @@ async fn main() -> anyhow::Result<()> {
     let listener = tokio::net::TcpListener::bind(config.listen).await.unwrap();
     info!("listening on {}", listener.local_addr().unwrap());
 
-    let storage =
-        remote_storage::GenericRemoteStorage::from_storage_kind(config.storage_kind).await?;
+    let storage = remote_storage::GenericRemoteStorage::from_config(&config.storage_config).await?;
     let cancel = tokio_util::sync::CancellationToken::new();
     if !args.no_s3_check_on_startup {
         app::check_storage_permissions(&storage, cancel.clone()).await?;

libs/compute_api/src/requests.rs

@@ -16,7 +16,6 @@ pub static COMPUTE_AUDIENCE: &str = "compute";
 pub enum ComputeClaimsScope {
     /// An admin-scoped token allows access to all of `compute_ctl`'s authorized
     /// facilities.
-    #[serde(rename = "compute_ctl:admin")]
     Admin,
 }
 
@@ -25,7 +24,7 @@ impl FromStr for ComputeClaimsScope {
 
     fn from_str(s: &str) -> Result<Self, Self::Err> {
         match s {
-            "compute_ctl:admin" => Ok(ComputeClaimsScope::Admin),
+            "admin" => Ok(ComputeClaimsScope::Admin),
             _ => Err(anyhow::anyhow!("invalid compute claims scope \"{s}\"")),
         }
     }
@@ -81,23 +80,3 @@ pub struct SetRoleGrantsRequest {
     pub privileges: Vec<Privilege>,
     pub role: PgIdent,
 }
-
-#[cfg(test)]
-mod test {
-    use std::str::FromStr;
-
-    use crate::requests::ComputeClaimsScope;
-
-    /// Confirm that whether we parse the scope by string or through serde, the
-    /// same values parse to the same enum variant.
-    #[test]
-    fn compute_request_scopes() {
-        const ADMIN_SCOPE: &str = "compute_ctl:admin";
-
-        let from_serde: ComputeClaimsScope =
-            serde_json::from_str(&format!("\"{ADMIN_SCOPE}\"")).unwrap();
-        let from_str = ComputeClaimsScope::from_str(ADMIN_SCOPE).unwrap();
-
-        assert_eq!(from_serde, from_str);
-    }
-}

libs/compute_api/src/responses.rs

@@ -83,16 +83,6 @@ pub struct ComputeStatusResponse {
     pub error: Option<String>,
 }
 
-#[derive(Serialize, Clone, Copy, Debug, Deserialize, PartialEq, Eq, Default)]
-#[serde(rename_all = "snake_case")]
-pub enum TerminateMode {
-    #[default]
-    /// wait 30s till returning from /terminate to allow control plane to get the error
-    Fast,
-    /// return from /terminate immediately as soon as all components are terminated
-    Immediate,
-}
-
 #[derive(Serialize, Clone, Copy, Debug, Deserialize, PartialEq, Eq)]
 #[serde(rename_all = "snake_case")]
 pub enum ComputeStatus {
@@ -113,16 +103,11 @@ pub enum ComputeStatus {
     // control-plane to terminate it.
     Failed,
     // Termination requested
-    TerminationPending { mode: TerminateMode },
+    TerminationPending,
     // Terminated Postgres
     Terminated,
 }
 
-#[derive(Deserialize, Serialize)]
-pub struct TerminateResponse {
-    pub lsn: Option<utils::lsn::Lsn>,
-}
-
 impl Display for ComputeStatus {
     fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
         match self {
@@ -132,7 +117,7 @@ impl Display for ComputeStatus {
             ComputeStatus::Running => f.write_str("running"),
             ComputeStatus::Configuration => f.write_str("configuration"),
             ComputeStatus::Failed => f.write_str("failed"),
-            ComputeStatus::TerminationPending { .. } => f.write_str("termination-pending"),
+            ComputeStatus::TerminationPending => f.write_str("termination-pending"),
             ComputeStatus::Terminated => f.write_str("terminated"),
         }
     }

libs/desim/src/executor.rs

@@ -71,7 +71,7 @@ impl Runtime {
                     debug!("thread panicked: {:?}", e);
                     let mut result = ctx.result.lock();
                     if result.0 == -1 {
-                        *result = (256, format!("thread panicked: {e:?}"));
+                        *result = (256, format!("thread panicked: {:?}", e));
                     }
                 });
             }
@@ -419,13 +419,13 @@ pub fn now() -> u64 {
     with_thread_context(|ctx| ctx.clock.get().unwrap().now())
 }
 
-pub fn exit(code: i32, msg: String) -> ! {
+pub fn exit(code: i32, msg: String) {
     with_thread_context(|ctx| {
         ctx.allow_panic.store(true, Ordering::SeqCst);
         let mut result = ctx.result.lock();
         *result = (code, msg);
         panic!("exit");
-    })
+    });
 }
 
 pub(crate) fn get_thread_ctx() -> Arc<ThreadContext> {

libs/desim/src/proto.rs

@@ -47,8 +47,8 @@ impl Debug for AnyMessage {
         match self {
             AnyMessage::None => write!(f, "None"),
             AnyMessage::InternalConnect => write!(f, "InternalConnect"),
-            AnyMessage::Just32(v) => write!(f, "Just32({v})"),
-            AnyMessage::ReplCell(v) => write!(f, "ReplCell({v:?})"),
+            AnyMessage::Just32(v) => write!(f, "Just32({})", v),
+            AnyMessage::ReplCell(v) => write!(f, "ReplCell({:?})", v),
             AnyMessage::Bytes(v) => write!(f, "Bytes({})", hex::encode(v)),
             AnyMessage::LSN(v) => write!(f, "LSN({})", Lsn(*v)),
         }

libs/http-utils/src/endpoint.rs

@@ -582,14 +582,14 @@ pub fn attach_openapi_ui(
                             deepLinking: true,
                             showExtensions: true,
                             showCommonExtensions: true,
-                            url: "{spec_mount_path}",
+                            url: "{}",
                         }})
                         window.ui = ui;
                     }};
                 </script>
                 </body>
                 </html>
-            "#))).unwrap())
+            "#, spec_mount_path))).unwrap())
              })
         )
 }
@@ -696,7 +696,7 @@ mod tests {
         let remote_addr = SocketAddr::new(IpAddr::from_str("127.0.0.1").unwrap(), 80);
         let mut service = builder.build(remote_addr);
         if let Err(e) = poll_fn(|ctx| service.poll_ready(ctx)).await {
-            panic!("request service is not ready: {e:?}");
+            panic!("request service is not ready: {:?}", e);
         }
 
         let mut req: Request<Body> = Request::default();
@@ -716,7 +716,7 @@ mod tests {
         let remote_addr = SocketAddr::new(IpAddr::from_str("127.0.0.1").unwrap(), 80);
         let mut service = builder.build(remote_addr);
         if let Err(e) = poll_fn(|ctx| service.poll_ready(ctx)).await {
-            panic!("request service is not ready: {e:?}");
+            panic!("request service is not ready: {:?}", e);
         }
 
         let req: Request<Body> = Request::default();

libs/neon-shmem/src/lib.rs

@@ -86,7 +86,7 @@ impl ShmemHandle {
         // somewhat smaller than that, because with anything close to that, you'll run out of
         // memory anyway.
         if max_size >= 1 << 48 {
-            panic!("max size {max_size} too large");
+            panic!("max size {} too large", max_size);
         }
         if initial_size > max_size {
             panic!("initial size {initial_size} larger than max size {max_size}");
@@ -279,7 +279,7 @@ mod tests {
     fn assert_range(ptr: *const u8, expected: u8, range: Range<usize>) {
         for i in range {
             let b = unsafe { *(ptr.add(i)) };
-            assert_eq!(expected, b, "unexpected byte at offset {i}");
+            assert_eq!(expected, b, "unexpected byte at offset {}", i);
         }
     }
 

libs/pageserver_api/Cargo.toml

@@ -17,7 +17,7 @@ anyhow.workspace = true
 bytes.workspace = true
 byteorder.workspace = true
 utils.workspace = true
-postgres_ffi_types.workspace = true
+postgres_ffi.workspace = true
 enum-map.workspace = true
 strum.workspace = true
 strum_macros.workspace = true

libs/pageserver_api/src/config.rs

@@ -12,7 +12,6 @@ pub const DEFAULT_HTTP_LISTEN_ADDR: &str = formatcp!("127.0.0.1:{DEFAULT_HTTP_LI
 pub const DEFAULT_GRPC_LISTEN_PORT: u16 = 51051; // storage-broker already uses 50051
 
 use std::collections::HashMap;
-use std::fmt::Display;
 use std::num::{NonZeroU64, NonZeroUsize};
 use std::str::FromStr;
 use std::time::Duration;
@@ -25,17 +24,16 @@ use utils::logging::LogFormat;
 use crate::models::{ImageCompressionAlgorithm, LsnLease};
 
 // Certain metadata (e.g. externally-addressable name, AZ) is delivered
-// as a separate structure.  This information is not needed by the pageserver
+// as a separate structure.  This information is not neeed by the pageserver
 // itself, it is only used for registering the pageserver with the control
 // plane and/or storage controller.
+//
 #[derive(PartialEq, Eq, Debug, serde::Serialize, serde::Deserialize)]
 pub struct NodeMetadata {
     #[serde(rename = "host")]
     pub postgres_host: String,
     #[serde(rename = "port")]
     pub postgres_port: u16,
-    pub grpc_host: Option<String>,
-    pub grpc_port: Option<u16>,
     pub http_host: String,
     pub http_port: u16,
     pub https_port: Option<u16>,
@@ -46,23 +44,6 @@ pub struct NodeMetadata {
     pub other: HashMap<String, serde_json::Value>,
 }
 
-impl Display for NodeMetadata {
-    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
-        write!(
-            f,
-            "postgresql://{}:{} ",
-            self.postgres_host, self.postgres_port
-        )?;
-        if let Some(grpc_host) = &self.grpc_host {
-            let grpc_port = self.grpc_port.unwrap_or_default();
-            write!(f, "grpc://{grpc_host}:{grpc_port} ")?;
-        }
-        write!(f, "http://{}:{} ", self.http_host, self.http_port)?;
-        write!(f, "other:{:?}", self.other)?;
-        Ok(())
-    }
-}
-
 /// PostHog integration config.
 #[derive(Debug, Clone, PartialEq, Eq, serde::Serialize, serde::Deserialize)]
 pub struct PostHogConfig {
@@ -76,10 +57,6 @@ pub struct PostHogConfig {
     pub private_api_url: String,
     /// Public API URL
     pub public_api_url: String,
-    /// Refresh interval for the feature flag spec
-    #[serde(skip_serializing_if = "Option::is_none")]
-    #[serde(with = "humantime_serde")]
-    pub refresh_interval: Option<Duration>,
 }
 
 /// `pageserver.toml`
@@ -360,21 +337,16 @@ pub struct TimelineImportConfig {
 pub struct BasebackupCacheConfig {
     #[serde(with = "humantime_serde")]
     pub cleanup_period: Duration,
-    /// Maximum total size of basebackup cache entries on disk in bytes.
-    /// The cache may slightly exceed this limit because we do not know
-    /// the exact size of the cache entry untill it's written to disk.
-    pub max_total_size_bytes: u64,
-    // TODO(diko): support max_entry_size_bytes.
-    // pub max_entry_size_bytes: u64,
-    pub max_size_entries: usize,
+    // FIXME: Support max_size_bytes.
+    // pub max_size_bytes: usize,
+    pub max_size_entries: i64,
 }
 
 impl Default for BasebackupCacheConfig {
     fn default() -> Self {
         Self {
             cleanup_period: Duration::from_secs(60),
-            max_total_size_bytes: 1024 * 1024 * 1024, // 1 GiB
-            // max_entry_size_bytes: 16 * 1024 * 1024,   // 16 MiB
+            // max_size_bytes: 1024 * 1024 * 1024, // 1 GiB
             max_size_entries: 1000,
         }
     }
@@ -820,7 +792,7 @@ pub mod tenant_conf_defaults {
     // By default ingest enough WAL for two new L0 layers before checking if new image
     // image layers should be created.
     pub const DEFAULT_IMAGE_LAYER_CREATION_CHECK_THRESHOLD: u8 = 2;
-    pub const DEFAULT_GC_COMPACTION_ENABLED: bool = true;
+    pub const DEFAULT_GC_COMPACTION_ENABLED: bool = false;
     pub const DEFAULT_GC_COMPACTION_VERIFICATION: bool = true;
     pub const DEFAULT_GC_COMPACTION_INITIAL_THRESHOLD_KB: u64 = 5 * 1024 * 1024; // 5GB
     pub const DEFAULT_GC_COMPACTION_RATIO_PERCENT: u64 = 100;

libs/pageserver_api/src/config/tests.rs

@@ -14,8 +14,6 @@ fn test_node_metadata_v1_backward_compatibilty() {
         NodeMetadata {
             postgres_host: "localhost".to_string(),
             postgres_port: 23,
-            grpc_host: None,
-            grpc_port: None,
             http_host: "localhost".to_string(),
             http_port: 42,
             https_port: None,
@@ -39,35 +37,6 @@ fn test_node_metadata_v2_backward_compatibilty() {
         NodeMetadata {
             postgres_host: "localhost".to_string(),
             postgres_port: 23,
-            grpc_host: None,
-            grpc_port: None,
-            http_host: "localhost".to_string(),
-            http_port: 42,
-            https_port: Some(123),
-            other: HashMap::new(),
-        }
-    )
-}
-
-#[test]
-fn test_node_metadata_v3_backward_compatibilty() {
-    let v3 = serde_json::to_vec(&serde_json::json!({
-        "host": "localhost",
-        "port": 23,
-        "grpc_host": "localhost",
-        "grpc_port": 51,
-        "http_host": "localhost",
-        "http_port": 42,
-        "https_port": 123,
-    }));
-
-    assert_eq!(
-        serde_json::from_slice::<NodeMetadata>(&v3.unwrap()).unwrap(),
-        NodeMetadata {
-            postgres_host: "localhost".to_string(),
-            postgres_port: 23,
-            grpc_host: Some("localhost".to_string()),
-            grpc_port: Some(51),
             http_host: "localhost".to_string(),
             http_port: 42,
             https_port: Some(123),

libs/pageserver_api/src/controller_api.rs

@@ -52,8 +52,6 @@ pub struct NodeRegisterRequest {
 
     pub listen_pg_addr: String,
     pub listen_pg_port: u16,
-    pub listen_grpc_addr: Option<String>,
-    pub listen_grpc_port: Option<u16>,
 
     pub listen_http_addr: String,
     pub listen_http_port: u16,
@@ -103,8 +101,6 @@ pub struct TenantLocateResponseShard {
 
     pub listen_pg_addr: String,
     pub listen_pg_port: u16,
-    pub listen_grpc_addr: Option<String>,
-    pub listen_grpc_port: Option<u16>,
 
     pub listen_http_addr: String,
     pub listen_http_port: u16,
@@ -156,8 +152,6 @@ pub struct NodeDescribeResponse {
 
     pub listen_pg_addr: String,
     pub listen_pg_port: u16,
-    pub listen_grpc_addr: Option<String>,
-    pub listen_grpc_port: Option<u16>,
 }
 
 #[derive(Serialize, Deserialize, Debug)]
@@ -577,7 +571,8 @@ mod test {
         let err = serde_json::from_value::<TenantCreateRequest>(create_request).unwrap_err();
         assert!(
             err.to_string().contains("unknown field `unknown_field`"),
-            "expect unknown field `unknown_field` error, got: {err}"
+            "expect unknown field `unknown_field` error, got: {}",
+            err
         );
     }
 

libs/pageserver_api/src/key.rs

@@ -4,8 +4,8 @@ use std::ops::Range;
 use anyhow::{Result, bail};
 use byteorder::{BE, ByteOrder};
 use bytes::Bytes;
-use postgres_ffi_types::forknum::{FSM_FORKNUM, VISIBILITYMAP_FORKNUM};
-use postgres_ffi_types::{Oid, RepOriginId};
+use postgres_ffi::relfile_utils::{FSM_FORKNUM, VISIBILITYMAP_FORKNUM};
+use postgres_ffi::{Oid, RepOriginId};
 use serde::{Deserialize, Serialize};
 use utils::const_assert;
 
@@ -194,7 +194,7 @@ impl Key {
     /// will be rejected on the write path.
     #[allow(dead_code)]
     pub fn is_valid_key_on_write_path_strong(&self) -> bool {
-        use postgres_ffi_types::constants::{DEFAULTTABLESPACE_OID, GLOBALTABLESPACE_OID};
+        use postgres_ffi::pg_constants::{DEFAULTTABLESPACE_OID, GLOBALTABLESPACE_OID};
         if !self.is_i128_representable() {
             return false;
         }

libs/pageserver_api/src/keyspace.rs

@@ -1,6 +1,7 @@
 use std::ops::Range;
 
 use itertools::Itertools;
+use postgres_ffi::BLCKSZ;
 
 use crate::key::Key;
 use crate::shard::{ShardCount, ShardIdentity};
@@ -268,13 +269,9 @@ impl KeySpace {
     /// Partition a key space into roughly chunks of roughly 'target_size' bytes
     /// in each partition.
     ///
-    pub fn partition(
-        &self,
-        shard_identity: &ShardIdentity,
-        target_size: u64,
-        block_size: u64,
-    ) -> KeyPartitioning {
-        let target_nblocks = (target_size / block_size) as u32;
+    pub fn partition(&self, shard_identity: &ShardIdentity, target_size: u64) -> KeyPartitioning {
+        // Assume that each value is 8k in size.
+        let target_nblocks = (target_size / BLCKSZ as u64) as u32;
 
         let mut parts = Vec::new();
         let mut current_part = Vec::new();
@@ -334,7 +331,8 @@ impl KeySpace {
                     std::cmp::max(range.start, prev.start) < std::cmp::min(range.end, prev.end);
                 assert!(
                     !overlap,
-                    "Attempt to merge ovelapping keyspaces: {prev:?} overlaps {range:?}"
+                    "Attempt to merge ovelapping keyspaces: {:?} overlaps {:?}",
+                    prev, range
                 );
             }
 
@@ -1103,7 +1101,7 @@ mod tests {
             // total range contains at least one shard-local page
             let all_nonzero = fragments.iter().all(|f| f.0 > 0);
             if !all_nonzero {
-                eprintln!("Found a zero-length fragment: {fragments:?}");
+                eprintln!("Found a zero-length fragment: {:?}", fragments);
             }
             assert!(all_nonzero);
         } else {

libs/pageserver_api/src/lib.rs

@@ -5,10 +5,11 @@ pub mod controller_api;
 pub mod key;
 pub mod keyspace;
 pub mod models;
-pub mod pagestream_api;
+pub mod record;
 pub mod reltag;
 pub mod shard;
 /// Public API types
 pub mod upcall_api;
+pub mod value;
 
 pub mod config;

libs/pageserver_api/src/models.rs

@@ -5,12 +5,16 @@ pub mod utilization;
 use core::ops::Range;
 use std::collections::HashMap;
 use std::fmt::Display;
+use std::io::{BufRead, Read};
 use std::num::{NonZeroU32, NonZeroU64, NonZeroUsize};
 use std::str::FromStr;
 use std::time::{Duration, SystemTime};
 
+use byteorder::{BigEndian, ReadBytesExt};
+use bytes::{Buf, BufMut, Bytes, BytesMut};
 #[cfg(feature = "testing")]
 use camino::Utf8PathBuf;
+use postgres_ffi::BLCKSZ;
 use serde::{Deserialize, Deserializer, Serialize, Serializer};
 use serde_with::serde_as;
 pub use utilization::PageserverUtilization;
@@ -20,6 +24,7 @@ use utils::{completion, serde_system_time};
 
 use crate::config::Ratio;
 use crate::key::{CompactKey, Key};
+use crate::reltag::RelTag;
 use crate::shard::{DEFAULT_STRIPE_SIZE, ShardCount, ShardStripeSize, TenantShardId};
 
 /// The state of a tenant in this pageserver.
@@ -1182,7 +1187,7 @@ impl Display for ImageCompressionAlgorithm {
             ImageCompressionAlgorithm::Disabled => write!(f, "disabled"),
             ImageCompressionAlgorithm::Zstd { level } => {
                 if let Some(level) = level {
-                    write!(f, "zstd({level})")
+                    write!(f, "zstd({})", level)
                 } else {
                     write!(f, "zstd")
                 }
@@ -1902,6 +1907,219 @@ pub struct ScanDisposableKeysResponse {
     pub not_disposable_count: usize,
 }
 
+// Wrapped in libpq CopyData
+#[derive(PartialEq, Eq, Debug)]
+pub enum PagestreamFeMessage {
+    Exists(PagestreamExistsRequest),
+    Nblocks(PagestreamNblocksRequest),
+    GetPage(PagestreamGetPageRequest),
+    DbSize(PagestreamDbSizeRequest),
+    GetSlruSegment(PagestreamGetSlruSegmentRequest),
+    #[cfg(feature = "testing")]
+    Test(PagestreamTestRequest),
+}
+
+// Wrapped in libpq CopyData
+#[derive(Debug, strum_macros::EnumProperty)]
+pub enum PagestreamBeMessage {
+    Exists(PagestreamExistsResponse),
+    Nblocks(PagestreamNblocksResponse),
+    GetPage(PagestreamGetPageResponse),
+    Error(PagestreamErrorResponse),
+    DbSize(PagestreamDbSizeResponse),
+    GetSlruSegment(PagestreamGetSlruSegmentResponse),
+    #[cfg(feature = "testing")]
+    Test(PagestreamTestResponse),
+}
+
+// Keep in sync with `pagestore_client.h`
+#[repr(u8)]
+enum PagestreamFeMessageTag {
+    Exists = 0,
+    Nblocks = 1,
+    GetPage = 2,
+    DbSize = 3,
+    GetSlruSegment = 4,
+    /* future tags above this line */
+    /// For testing purposes, not available in production.
+    #[cfg(feature = "testing")]
+    Test = 99,
+}
+
+// Keep in sync with `pagestore_client.h`
+#[repr(u8)]
+enum PagestreamBeMessageTag {
+    Exists = 100,
+    Nblocks = 101,
+    GetPage = 102,
+    Error = 103,
+    DbSize = 104,
+    GetSlruSegment = 105,
+    /* future tags above this line */
+    /// For testing purposes, not available in production.
+    #[cfg(feature = "testing")]
+    Test = 199,
+}
+
+impl TryFrom<u8> for PagestreamFeMessageTag {
+    type Error = u8;
+    fn try_from(value: u8) -> Result<Self, u8> {
+        match value {
+            0 => Ok(PagestreamFeMessageTag::Exists),
+            1 => Ok(PagestreamFeMessageTag::Nblocks),
+            2 => Ok(PagestreamFeMessageTag::GetPage),
+            3 => Ok(PagestreamFeMessageTag::DbSize),
+            4 => Ok(PagestreamFeMessageTag::GetSlruSegment),
+            #[cfg(feature = "testing")]
+            99 => Ok(PagestreamFeMessageTag::Test),
+            _ => Err(value),
+        }
+    }
+}
+
+impl TryFrom<u8> for PagestreamBeMessageTag {
+    type Error = u8;
+    fn try_from(value: u8) -> Result<Self, u8> {
+        match value {
+            100 => Ok(PagestreamBeMessageTag::Exists),
+            101 => Ok(PagestreamBeMessageTag::Nblocks),
+            102 => Ok(PagestreamBeMessageTag::GetPage),
+            103 => Ok(PagestreamBeMessageTag::Error),
+            104 => Ok(PagestreamBeMessageTag::DbSize),
+            105 => Ok(PagestreamBeMessageTag::GetSlruSegment),
+            #[cfg(feature = "testing")]
+            199 => Ok(PagestreamBeMessageTag::Test),
+            _ => Err(value),
+        }
+    }
+}
+
+// A GetPage request contains two LSN values:
+//
+// request_lsn: Get the page version at this point in time.  Lsn::Max is a special value that means
+// "get the latest version present". It's used by the primary server, which knows that no one else
+// is writing WAL. 'not_modified_since' must be set to a proper value even if request_lsn is
+// Lsn::Max. Standby servers use the current replay LSN as the request LSN.
+//
+// not_modified_since: Hint to the pageserver that the client knows that the page has not been
+// modified between 'not_modified_since' and the request LSN. It's always correct to set
+// 'not_modified_since equal' to 'request_lsn' (unless Lsn::Max is used as the 'request_lsn'), but
+// passing an earlier LSN can speed up the request, by allowing the pageserver to process the
+// request without waiting for 'request_lsn' to arrive.
+//
+// The now-defunct V1 interface contained only one LSN, and a boolean 'latest' flag. The V1 interface was
+// sufficient for the primary; the 'lsn' was equivalent to the 'not_modified_since' value, and
+// 'latest' was set to true. The V2 interface was added because there was no correct way for a
+// standby to request a page at a particular non-latest LSN, and also include the
+// 'not_modified_since' hint. That led to an awkward choice of either using an old LSN in the
+// request, if the standby knows that the page hasn't been modified since, and risk getting an error
+// if that LSN has fallen behind the GC horizon, or requesting the current replay LSN, which could
+// require the pageserver unnecessarily to wait for the WAL to arrive up to that point. The new V2
+// interface allows sending both LSNs, and let the pageserver do the right thing. There was no
+// difference in the responses between V1 and V2.
+//
+// V3 version of protocol adds request ID to all requests. This request ID is also included in response
+// as well as other fields from requests, which allows to verify that we receive response for our request.
+// We copy fields from request to response to make checking more reliable: request ID is formed from process ID
+// and local counter, so in principle there can be duplicated requests IDs if process PID is reused.
+//
+#[derive(Debug, PartialEq, Eq, Clone, Copy)]
+pub enum PagestreamProtocolVersion {
+    V2,
+    V3,
+}
+
+pub type RequestId = u64;
+
+#[derive(Debug, Default, PartialEq, Eq, Clone, Copy)]
+pub struct PagestreamRequest {
+    pub reqid: RequestId,
+    pub request_lsn: Lsn,
+    pub not_modified_since: Lsn,
+}
+
+#[derive(Debug, PartialEq, Eq, Clone, Copy)]
+pub struct PagestreamExistsRequest {
+    pub hdr: PagestreamRequest,
+    pub rel: RelTag,
+}
+
+#[derive(Debug, PartialEq, Eq, Clone, Copy)]
+pub struct PagestreamNblocksRequest {
+    pub hdr: PagestreamRequest,
+    pub rel: RelTag,
+}
+
+#[derive(Debug, Default, PartialEq, Eq, Clone, Copy)]
+pub struct PagestreamGetPageRequest {
+    pub hdr: PagestreamRequest,
+    pub rel: RelTag,
+    pub blkno: u32,
+}
+
+#[derive(Debug, PartialEq, Eq, Clone, Copy)]
+pub struct PagestreamDbSizeRequest {
+    pub hdr: PagestreamRequest,
+    pub dbnode: u32,
+}
+
+#[derive(Debug, PartialEq, Eq, Clone, Copy)]
+pub struct PagestreamGetSlruSegmentRequest {
+    pub hdr: PagestreamRequest,
+    pub kind: u8,
+    pub segno: u32,
+}
+
+#[derive(Debug)]
+pub struct PagestreamExistsResponse {
+    pub req: PagestreamExistsRequest,
+    pub exists: bool,
+}
+
+#[derive(Debug)]
+pub struct PagestreamNblocksResponse {
+    pub req: PagestreamNblocksRequest,
+    pub n_blocks: u32,
+}
+
+#[derive(Debug)]
+pub struct PagestreamGetPageResponse {
+    pub req: PagestreamGetPageRequest,
+    pub page: Bytes,
+}
+
+#[derive(Debug)]
+pub struct PagestreamGetSlruSegmentResponse {
+    pub req: PagestreamGetSlruSegmentRequest,
+    pub segment: Bytes,
+}
+
+#[derive(Debug)]
+pub struct PagestreamErrorResponse {
+    pub req: PagestreamRequest,
+    pub message: String,
+}
+
+#[derive(Debug)]
+pub struct PagestreamDbSizeResponse {
+    pub req: PagestreamDbSizeRequest,
+    pub db_size: i64,
+}
+
+#[cfg(feature = "testing")]
+#[derive(Debug, PartialEq, Eq, Clone)]
+pub struct PagestreamTestRequest {
+    pub hdr: PagestreamRequest,
+    pub batch_key: u64,
+    pub message: String,
+}
+
+#[cfg(feature = "testing")]
+#[derive(Debug)]
+pub struct PagestreamTestResponse {
+    pub req: PagestreamTestRequest,
+}
+
 // This is a cut-down version of TenantHistorySize from the pageserver crate, omitting fields
 // that require pageserver-internal types.  It is sufficient to get the total size.
 #[derive(Serialize, Deserialize, Debug)]
@@ -1913,6 +2131,506 @@ pub struct TenantHistorySize {
     pub size: Option<u64>,
 }
 
+impl PagestreamFeMessage {
+    /// Serialize a compute -> pageserver message. This is currently only used in testing
+    /// tools. Always uses protocol version 3.
+    pub fn serialize(&self) -> Bytes {
+        let mut bytes = BytesMut::new();
+
+        match self {
+            Self::Exists(req) => {
+                bytes.put_u8(PagestreamFeMessageTag::Exists as u8);
+                bytes.put_u64(req.hdr.reqid);
+                bytes.put_u64(req.hdr.request_lsn.0);
+                bytes.put_u64(req.hdr.not_modified_since.0);
+                bytes.put_u32(req.rel.spcnode);
+                bytes.put_u32(req.rel.dbnode);
+                bytes.put_u32(req.rel.relnode);
+                bytes.put_u8(req.rel.forknum);
+            }
+
+            Self::Nblocks(req) => {
+                bytes.put_u8(PagestreamFeMessageTag::Nblocks as u8);
+                bytes.put_u64(req.hdr.reqid);
+                bytes.put_u64(req.hdr.request_lsn.0);
+                bytes.put_u64(req.hdr.not_modified_since.0);
+                bytes.put_u32(req.rel.spcnode);
+                bytes.put_u32(req.rel.dbnode);
+                bytes.put_u32(req.rel.relnode);
+                bytes.put_u8(req.rel.forknum);
+            }
+
+            Self::GetPage(req) => {
+                bytes.put_u8(PagestreamFeMessageTag::GetPage as u8);
+                bytes.put_u64(req.hdr.reqid);
+                bytes.put_u64(req.hdr.request_lsn.0);
+                bytes.put_u64(req.hdr.not_modified_since.0);
+                bytes.put_u32(req.rel.spcnode);
+                bytes.put_u32(req.rel.dbnode);
+                bytes.put_u32(req.rel.relnode);
+                bytes.put_u8(req.rel.forknum);
+                bytes.put_u32(req.blkno);
+            }
+
+            Self::DbSize(req) => {
+                bytes.put_u8(PagestreamFeMessageTag::DbSize as u8);
+                bytes.put_u64(req.hdr.reqid);
+                bytes.put_u64(req.hdr.request_lsn.0);
+                bytes.put_u64(req.hdr.not_modified_since.0);
+                bytes.put_u32(req.dbnode);
+            }
+
+            Self::GetSlruSegment(req) => {
+                bytes.put_u8(PagestreamFeMessageTag::GetSlruSegment as u8);
+                bytes.put_u64(req.hdr.reqid);
+                bytes.put_u64(req.hdr.request_lsn.0);
+                bytes.put_u64(req.hdr.not_modified_since.0);
+                bytes.put_u8(req.kind);
+                bytes.put_u32(req.segno);
+            }
+            #[cfg(feature = "testing")]
+            Self::Test(req) => {
+                bytes.put_u8(PagestreamFeMessageTag::Test as u8);
+                bytes.put_u64(req.hdr.reqid);
+                bytes.put_u64(req.hdr.request_lsn.0);
+                bytes.put_u64(req.hdr.not_modified_since.0);
+                bytes.put_u64(req.batch_key);
+                let message = req.message.as_bytes();
+                bytes.put_u64(message.len() as u64);
+                bytes.put_slice(message);
+            }
+        }
+
+        bytes.into()
+    }
+
+    pub fn parse<R: std::io::Read>(
+        body: &mut R,
+        protocol_version: PagestreamProtocolVersion,
+    ) -> anyhow::Result<PagestreamFeMessage> {
+        // these correspond to the NeonMessageTag enum in pagestore_client.h
+        //
+        // TODO: consider using protobuf or serde bincode for less error prone
+        // serialization.
+        let msg_tag = body.read_u8()?;
+        let (reqid, request_lsn, not_modified_since) = match protocol_version {
+            PagestreamProtocolVersion::V2 => (
+                0,
+                Lsn::from(body.read_u64::<BigEndian>()?),
+                Lsn::from(body.read_u64::<BigEndian>()?),
+            ),
+            PagestreamProtocolVersion::V3 => (
+                body.read_u64::<BigEndian>()?,
+                Lsn::from(body.read_u64::<BigEndian>()?),
+                Lsn::from(body.read_u64::<BigEndian>()?),
+            ),
+        };
+
+        match PagestreamFeMessageTag::try_from(msg_tag)
+            .map_err(|tag: u8| anyhow::anyhow!("invalid tag {tag}"))?
+        {
+            PagestreamFeMessageTag::Exists => {
+                Ok(PagestreamFeMessage::Exists(PagestreamExistsRequest {
+                    hdr: PagestreamRequest {
+                        reqid,
+                        request_lsn,
+                        not_modified_since,
+                    },
+                    rel: RelTag {
+                        spcnode: body.read_u32::<BigEndian>()?,
+                        dbnode: body.read_u32::<BigEndian>()?,
+                        relnode: body.read_u32::<BigEndian>()?,
+                        forknum: body.read_u8()?,
+                    },
+                }))
+            }
+            PagestreamFeMessageTag::Nblocks => {
+                Ok(PagestreamFeMessage::Nblocks(PagestreamNblocksRequest {
+                    hdr: PagestreamRequest {
+                        reqid,
+                        request_lsn,
+                        not_modified_since,
+                    },
+                    rel: RelTag {
+                        spcnode: body.read_u32::<BigEndian>()?,
+                        dbnode: body.read_u32::<BigEndian>()?,
+                        relnode: body.read_u32::<BigEndian>()?,
+                        forknum: body.read_u8()?,
+                    },
+                }))
+            }
+            PagestreamFeMessageTag::GetPage => {
+                Ok(PagestreamFeMessage::GetPage(PagestreamGetPageRequest {
+                    hdr: PagestreamRequest {
+                        reqid,
+                        request_lsn,
+                        not_modified_since,
+                    },
+                    rel: RelTag {
+                        spcnode: body.read_u32::<BigEndian>()?,
+                        dbnode: body.read_u32::<BigEndian>()?,
+                        relnode: body.read_u32::<BigEndian>()?,
+                        forknum: body.read_u8()?,
+                    },
+                    blkno: body.read_u32::<BigEndian>()?,
+                }))
+            }
+            PagestreamFeMessageTag::DbSize => {
+                Ok(PagestreamFeMessage::DbSize(PagestreamDbSizeRequest {
+                    hdr: PagestreamRequest {
+                        reqid,
+                        request_lsn,
+                        not_modified_since,
+                    },
+                    dbnode: body.read_u32::<BigEndian>()?,
+                }))
+            }
+            PagestreamFeMessageTag::GetSlruSegment => Ok(PagestreamFeMessage::GetSlruSegment(
+                PagestreamGetSlruSegmentRequest {
+                    hdr: PagestreamRequest {
+                        reqid,
+                        request_lsn,
+                        not_modified_since,
+                    },
+                    kind: body.read_u8()?,
+                    segno: body.read_u32::<BigEndian>()?,
+                },
+            )),
+            #[cfg(feature = "testing")]
+            PagestreamFeMessageTag::Test => Ok(PagestreamFeMessage::Test(PagestreamTestRequest {
+                hdr: PagestreamRequest {
+                    reqid,
+                    request_lsn,
+                    not_modified_since,
+                },
+                batch_key: body.read_u64::<BigEndian>()?,
+                message: {
+                    let len = body.read_u64::<BigEndian>()?;
+                    let mut buf = vec![0; len as usize];
+                    body.read_exact(&mut buf)?;
+                    String::from_utf8(buf)?
+                },
+            })),
+        }
+    }
+}
+
+impl PagestreamBeMessage {
+    pub fn serialize(&self, protocol_version: PagestreamProtocolVersion) -> Bytes {
+        let mut bytes = BytesMut::new();
+
+        use PagestreamBeMessageTag as Tag;
+        match protocol_version {
+            PagestreamProtocolVersion::V2 => {
+                match self {
+                    Self::Exists(resp) => {
+                        bytes.put_u8(Tag::Exists as u8);
+                        bytes.put_u8(resp.exists as u8);
+                    }
+
+                    Self::Nblocks(resp) => {
+                        bytes.put_u8(Tag::Nblocks as u8);
+                        bytes.put_u32(resp.n_blocks);
+                    }
+
+                    Self::GetPage(resp) => {
+                        bytes.put_u8(Tag::GetPage as u8);
+                        bytes.put(&resp.page[..])
+                    }
+
+                    Self::Error(resp) => {
+                        bytes.put_u8(Tag::Error as u8);
+                        bytes.put(resp.message.as_bytes());
+                        bytes.put_u8(0); // null terminator
+                    }
+                    Self::DbSize(resp) => {
+                        bytes.put_u8(Tag::DbSize as u8);
+                        bytes.put_i64(resp.db_size);
+                    }
+
+                    Self::GetSlruSegment(resp) => {
+                        bytes.put_u8(Tag::GetSlruSegment as u8);
+                        bytes.put_u32((resp.segment.len() / BLCKSZ as usize) as u32);
+                        bytes.put(&resp.segment[..]);
+                    }
+
+                    #[cfg(feature = "testing")]
+                    Self::Test(resp) => {
+                        bytes.put_u8(Tag::Test as u8);
+                        bytes.put_u64(resp.req.batch_key);
+                        let message = resp.req.message.as_bytes();
+                        bytes.put_u64(message.len() as u64);
+                        bytes.put_slice(message);
+                    }
+                }
+            }
+            PagestreamProtocolVersion::V3 => {
+                match self {
+                    Self::Exists(resp) => {
+                        bytes.put_u8(Tag::Exists as u8);
+                        bytes.put_u64(resp.req.hdr.reqid);
+                        bytes.put_u64(resp.req.hdr.request_lsn.0);
+                        bytes.put_u64(resp.req.hdr.not_modified_since.0);
+                        bytes.put_u32(resp.req.rel.spcnode);
+                        bytes.put_u32(resp.req.rel.dbnode);
+                        bytes.put_u32(resp.req.rel.relnode);
+                        bytes.put_u8(resp.req.rel.forknum);
+                        bytes.put_u8(resp.exists as u8);
+                    }
+
+                    Self::Nblocks(resp) => {
+                        bytes.put_u8(Tag::Nblocks as u8);
+                        bytes.put_u64(resp.req.hdr.reqid);
+                        bytes.put_u64(resp.req.hdr.request_lsn.0);
+                        bytes.put_u64(resp.req.hdr.not_modified_since.0);
+                        bytes.put_u32(resp.req.rel.spcnode);
+                        bytes.put_u32(resp.req.rel.dbnode);
+                        bytes.put_u32(resp.req.rel.relnode);
+                        bytes.put_u8(resp.req.rel.forknum);
+                        bytes.put_u32(resp.n_blocks);
+                    }
+
+                    Self::GetPage(resp) => {
+                        bytes.put_u8(Tag::GetPage as u8);
+                        bytes.put_u64(resp.req.hdr.reqid);
+                        bytes.put_u64(resp.req.hdr.request_lsn.0);
+                        bytes.put_u64(resp.req.hdr.not_modified_since.0);
+                        bytes.put_u32(resp.req.rel.spcnode);
+                        bytes.put_u32(resp.req.rel.dbnode);
+                        bytes.put_u32(resp.req.rel.relnode);
+                        bytes.put_u8(resp.req.rel.forknum);
+                        bytes.put_u32(resp.req.blkno);
+                        bytes.put(&resp.page[..])
+                    }
+
+                    Self::Error(resp) => {
+                        bytes.put_u8(Tag::Error as u8);
+                        bytes.put_u64(resp.req.reqid);
+                        bytes.put_u64(resp.req.request_lsn.0);
+                        bytes.put_u64(resp.req.not_modified_since.0);
+                        bytes.put(resp.message.as_bytes());
+                        bytes.put_u8(0); // null terminator
+                    }
+                    Self::DbSize(resp) => {
+                        bytes.put_u8(Tag::DbSize as u8);
+                        bytes.put_u64(resp.req.hdr.reqid);
+                        bytes.put_u64(resp.req.hdr.request_lsn.0);
+                        bytes.put_u64(resp.req.hdr.not_modified_since.0);
+                        bytes.put_u32(resp.req.dbnode);
+                        bytes.put_i64(resp.db_size);
+                    }
+
+                    Self::GetSlruSegment(resp) => {
+                        bytes.put_u8(Tag::GetSlruSegment as u8);
+                        bytes.put_u64(resp.req.hdr.reqid);
+                        bytes.put_u64(resp.req.hdr.request_lsn.0);
+                        bytes.put_u64(resp.req.hdr.not_modified_since.0);
+                        bytes.put_u8(resp.req.kind);
+                        bytes.put_u32(resp.req.segno);
+                        bytes.put_u32((resp.segment.len() / BLCKSZ as usize) as u32);
+                        bytes.put(&resp.segment[..]);
+                    }
+
+                    #[cfg(feature = "testing")]
+                    Self::Test(resp) => {
+                        bytes.put_u8(Tag::Test as u8);
+                        bytes.put_u64(resp.req.hdr.reqid);
+                        bytes.put_u64(resp.req.hdr.request_lsn.0);
+                        bytes.put_u64(resp.req.hdr.not_modified_since.0);
+                        bytes.put_u64(resp.req.batch_key);
+                        let message = resp.req.message.as_bytes();
+                        bytes.put_u64(message.len() as u64);
+                        bytes.put_slice(message);
+                    }
+                }
+            }
+        }
+        bytes.into()
+    }
+
+    pub fn deserialize(buf: Bytes) -> anyhow::Result<Self> {
+        let mut buf = buf.reader();
+        let msg_tag = buf.read_u8()?;
+
+        use PagestreamBeMessageTag as Tag;
+        let ok =
+            match Tag::try_from(msg_tag).map_err(|tag: u8| anyhow::anyhow!("invalid tag {tag}"))? {
+                Tag::Exists => {
+                    let reqid = buf.read_u64::<BigEndian>()?;
+                    let request_lsn = Lsn(buf.read_u64::<BigEndian>()?);
+                    let not_modified_since = Lsn(buf.read_u64::<BigEndian>()?);
+                    let rel = RelTag {
+                        spcnode: buf.read_u32::<BigEndian>()?,
+                        dbnode: buf.read_u32::<BigEndian>()?,
+                        relnode: buf.read_u32::<BigEndian>()?,
+                        forknum: buf.read_u8()?,
+                    };
+                    let exists = buf.read_u8()? != 0;
+                    Self::Exists(PagestreamExistsResponse {
+                        req: PagestreamExistsRequest {
+                            hdr: PagestreamRequest {
+                                reqid,
+                                request_lsn,
+                                not_modified_since,
+                            },
+                            rel,
+                        },
+                        exists,
+                    })
+                }
+                Tag::Nblocks => {
+                    let reqid = buf.read_u64::<BigEndian>()?;
+                    let request_lsn = Lsn(buf.read_u64::<BigEndian>()?);
+                    let not_modified_since = Lsn(buf.read_u64::<BigEndian>()?);
+                    let rel = RelTag {
+                        spcnode: buf.read_u32::<BigEndian>()?,
+                        dbnode: buf.read_u32::<BigEndian>()?,
+                        relnode: buf.read_u32::<BigEndian>()?,
+                        forknum: buf.read_u8()?,
+                    };
+                    let n_blocks = buf.read_u32::<BigEndian>()?;
+                    Self::Nblocks(PagestreamNblocksResponse {
+                        req: PagestreamNblocksRequest {
+                            hdr: PagestreamRequest {
+                                reqid,
+                                request_lsn,
+                                not_modified_since,
+                            },
+                            rel,
+                        },
+                        n_blocks,
+                    })
+                }
+                Tag::GetPage => {
+                    let reqid = buf.read_u64::<BigEndian>()?;
+                    let request_lsn = Lsn(buf.read_u64::<BigEndian>()?);
+                    let not_modified_since = Lsn(buf.read_u64::<BigEndian>()?);
+                    let rel = RelTag {
+                        spcnode: buf.read_u32::<BigEndian>()?,
+                        dbnode: buf.read_u32::<BigEndian>()?,
+                        relnode: buf.read_u32::<BigEndian>()?,
+                        forknum: buf.read_u8()?,
+                    };
+                    let blkno = buf.read_u32::<BigEndian>()?;
+                    let mut page = vec![0; 8192]; // TODO: use MaybeUninit
+                    buf.read_exact(&mut page)?;
+                    Self::GetPage(PagestreamGetPageResponse {
+                        req: PagestreamGetPageRequest {
+                            hdr: PagestreamRequest {
+                                reqid,
+                                request_lsn,
+                                not_modified_since,
+                            },
+                            rel,
+                            blkno,
+                        },
+                        page: page.into(),
+                    })
+                }
+                Tag::Error => {
+                    let reqid = buf.read_u64::<BigEndian>()?;
+                    let request_lsn = Lsn(buf.read_u64::<BigEndian>()?);
+                    let not_modified_since = Lsn(buf.read_u64::<BigEndian>()?);
+                    let mut msg = Vec::new();
+                    buf.read_until(0, &mut msg)?;
+                    let cstring = std::ffi::CString::from_vec_with_nul(msg)?;
+                    let rust_str = cstring.to_str()?;
+                    Self::Error(PagestreamErrorResponse {
+                        req: PagestreamRequest {
+                            reqid,
+                            request_lsn,
+                            not_modified_since,
+                        },
+                        message: rust_str.to_owned(),
+                    })
+                }
+                Tag::DbSize => {
+                    let reqid = buf.read_u64::<BigEndian>()?;
+                    let request_lsn = Lsn(buf.read_u64::<BigEndian>()?);
+                    let not_modified_since = Lsn(buf.read_u64::<BigEndian>()?);
+                    let dbnode = buf.read_u32::<BigEndian>()?;
+                    let db_size = buf.read_i64::<BigEndian>()?;
+                    Self::DbSize(PagestreamDbSizeResponse {
+                        req: PagestreamDbSizeRequest {
+                            hdr: PagestreamRequest {
+                                reqid,
+                                request_lsn,
+                                not_modified_since,
+                            },
+                            dbnode,
+                        },
+                        db_size,
+                    })
+                }
+                Tag::GetSlruSegment => {
+                    let reqid = buf.read_u64::<BigEndian>()?;
+                    let request_lsn = Lsn(buf.read_u64::<BigEndian>()?);
+                    let not_modified_since = Lsn(buf.read_u64::<BigEndian>()?);
+                    let kind = buf.read_u8()?;
+                    let segno = buf.read_u32::<BigEndian>()?;
+                    let n_blocks = buf.read_u32::<BigEndian>()?;
+                    let mut segment = vec![0; n_blocks as usize * BLCKSZ as usize];
+                    buf.read_exact(&mut segment)?;
+                    Self::GetSlruSegment(PagestreamGetSlruSegmentResponse {
+                        req: PagestreamGetSlruSegmentRequest {
+                            hdr: PagestreamRequest {
+                                reqid,
+                                request_lsn,
+                                not_modified_since,
+                            },
+                            kind,
+                            segno,
+                        },
+                        segment: segment.into(),
+                    })
+                }
+                #[cfg(feature = "testing")]
+                Tag::Test => {
+                    let reqid = buf.read_u64::<BigEndian>()?;
+                    let request_lsn = Lsn(buf.read_u64::<BigEndian>()?);
+                    let not_modified_since = Lsn(buf.read_u64::<BigEndian>()?);
+                    let batch_key = buf.read_u64::<BigEndian>()?;
+                    let len = buf.read_u64::<BigEndian>()?;
+                    let mut msg = vec![0; len as usize];
+                    buf.read_exact(&mut msg)?;
+                    let message = String::from_utf8(msg)?;
+                    Self::Test(PagestreamTestResponse {
+                        req: PagestreamTestRequest {
+                            hdr: PagestreamRequest {
+                                reqid,
+                                request_lsn,
+                                not_modified_since,
+                            },
+                            batch_key,
+                            message,
+                        },
+                    })
+                }
+            };
+        let remaining = buf.into_inner();
+        if !remaining.is_empty() {
+            anyhow::bail!(
+                "remaining bytes in msg with tag={msg_tag}: {}",
+                remaining.len()
+            );
+        }
+        Ok(ok)
+    }
+
+    pub fn kind(&self) -> &'static str {
+        match self {
+            Self::Exists(_) => "Exists",
+            Self::Nblocks(_) => "Nblocks",
+            Self::GetPage(_) => "GetPage",
+            Self::Error(_) => "Error",
+            Self::DbSize(_) => "DbSize",
+            Self::GetSlruSegment(_) => "GetSlruSegment",
+            #[cfg(feature = "testing")]
+            Self::Test(_) => "Test",
+        }
+    }
+}
+
 #[derive(Debug, Serialize, Deserialize)]
 pub struct PageTraceEvent {
     pub key: CompactKey,
@@ -1938,6 +2656,68 @@ mod tests {
 
     use super::*;
 
+    #[test]
+    fn test_pagestream() {
+        // Test serialization/deserialization of PagestreamFeMessage
+        let messages = vec![
+            PagestreamFeMessage::Exists(PagestreamExistsRequest {
+                hdr: PagestreamRequest {
+                    reqid: 0,
+                    request_lsn: Lsn(4),
+                    not_modified_since: Lsn(3),
+                },
+                rel: RelTag {
+                    forknum: 1,
+                    spcnode: 2,
+                    dbnode: 3,
+                    relnode: 4,
+                },
+            }),
+            PagestreamFeMessage::Nblocks(PagestreamNblocksRequest {
+                hdr: PagestreamRequest {
+                    reqid: 0,
+                    request_lsn: Lsn(4),
+                    not_modified_since: Lsn(4),
+                },
+                rel: RelTag {
+                    forknum: 1,
+                    spcnode: 2,
+                    dbnode: 3,
+                    relnode: 4,
+                },
+            }),
+            PagestreamFeMessage::GetPage(PagestreamGetPageRequest {
+                hdr: PagestreamRequest {
+                    reqid: 0,
+                    request_lsn: Lsn(4),
+                    not_modified_since: Lsn(3),
+                },
+                rel: RelTag {
+                    forknum: 1,
+                    spcnode: 2,
+                    dbnode: 3,
+                    relnode: 4,
+                },
+                blkno: 7,
+            }),
+            PagestreamFeMessage::DbSize(PagestreamDbSizeRequest {
+                hdr: PagestreamRequest {
+                    reqid: 0,
+                    request_lsn: Lsn(4),
+                    not_modified_since: Lsn(3),
+                },
+                dbnode: 7,
+            }),
+        ];
+        for msg in messages {
+            let bytes = msg.serialize();
+            let reconstructed =
+                PagestreamFeMessage::parse(&mut bytes.reader(), PagestreamProtocolVersion::V3)
+                    .unwrap();
+            assert!(msg == reconstructed);
+        }
+    }
+
     #[test]
     fn test_tenantinfo_serde() {
         // Test serialization/deserialization of TenantInfo
@@ -2011,7 +2791,8 @@ mod tests {
         let err = serde_json::from_value::<TenantConfigRequest>(config_request).unwrap_err();
         assert!(
             err.to_string().contains("unknown field `unknown_field`"),
-            "expect unknown field `unknown_field` error, got: {err}"
+            "expect unknown field `unknown_field` error, got: {}",
+            err
         );
     }
 

libs/pageserver_api/src/pagestream_api.rs

@@ -1,798 +0,0 @@
-//! Rust definitions of the libpq-based pagestream API
-//!
-//! See also the C implementation of the same API in pgxn/neon/pagestore_client.h
-
-use std::io::{BufRead, Read};
-
-use crate::reltag::RelTag;
-
-use byteorder::{BigEndian, ReadBytesExt};
-use bytes::{Buf, BufMut, Bytes, BytesMut};
-use utils::lsn::Lsn;
-
-/// Block size.
-///
-/// XXX: We assume 8k block size in the SLRU fetch API. It's not great to hardcode
-/// that in the protocol, because Postgres supports different block sizes as a compile
-/// time option.
-const BLCKSZ: usize = 8192;
-
-// Wrapped in libpq CopyData
-#[derive(PartialEq, Eq, Debug)]
-pub enum PagestreamFeMessage {
-    Exists(PagestreamExistsRequest),
-    Nblocks(PagestreamNblocksRequest),
-    GetPage(PagestreamGetPageRequest),
-    DbSize(PagestreamDbSizeRequest),
-    GetSlruSegment(PagestreamGetSlruSegmentRequest),
-    #[cfg(feature = "testing")]
-    Test(PagestreamTestRequest),
-}
-
-// Wrapped in libpq CopyData
-#[derive(Debug, strum_macros::EnumProperty)]
-pub enum PagestreamBeMessage {
-    Exists(PagestreamExistsResponse),
-    Nblocks(PagestreamNblocksResponse),
-    GetPage(PagestreamGetPageResponse),
-    Error(PagestreamErrorResponse),
-    DbSize(PagestreamDbSizeResponse),
-    GetSlruSegment(PagestreamGetSlruSegmentResponse),
-    #[cfg(feature = "testing")]
-    Test(PagestreamTestResponse),
-}
-
-// Keep in sync with `pagestore_client.h`
-#[repr(u8)]
-enum PagestreamFeMessageTag {
-    Exists = 0,
-    Nblocks = 1,
-    GetPage = 2,
-    DbSize = 3,
-    GetSlruSegment = 4,
-    /* future tags above this line */
-    /// For testing purposes, not available in production.
-    #[cfg(feature = "testing")]
-    Test = 99,
-}
-
-// Keep in sync with `pagestore_client.h`
-#[repr(u8)]
-enum PagestreamBeMessageTag {
-    Exists = 100,
-    Nblocks = 101,
-    GetPage = 102,
-    Error = 103,
-    DbSize = 104,
-    GetSlruSegment = 105,
-    /* future tags above this line */
-    /// For testing purposes, not available in production.
-    #[cfg(feature = "testing")]
-    Test = 199,
-}
-
-impl TryFrom<u8> for PagestreamFeMessageTag {
-    type Error = u8;
-    fn try_from(value: u8) -> Result<Self, u8> {
-        match value {
-            0 => Ok(PagestreamFeMessageTag::Exists),
-            1 => Ok(PagestreamFeMessageTag::Nblocks),
-            2 => Ok(PagestreamFeMessageTag::GetPage),
-            3 => Ok(PagestreamFeMessageTag::DbSize),
-            4 => Ok(PagestreamFeMessageTag::GetSlruSegment),
-            #[cfg(feature = "testing")]
-            99 => Ok(PagestreamFeMessageTag::Test),
-            _ => Err(value),
-        }
-    }
-}
-
-impl TryFrom<u8> for PagestreamBeMessageTag {
-    type Error = u8;
-    fn try_from(value: u8) -> Result<Self, u8> {
-        match value {
-            100 => Ok(PagestreamBeMessageTag::Exists),
-            101 => Ok(PagestreamBeMessageTag::Nblocks),
-            102 => Ok(PagestreamBeMessageTag::GetPage),
-            103 => Ok(PagestreamBeMessageTag::Error),
-            104 => Ok(PagestreamBeMessageTag::DbSize),
-            105 => Ok(PagestreamBeMessageTag::GetSlruSegment),
-            #[cfg(feature = "testing")]
-            199 => Ok(PagestreamBeMessageTag::Test),
-            _ => Err(value),
-        }
-    }
-}
-
-// A GetPage request contains two LSN values:
-//
-// request_lsn: Get the page version at this point in time.  Lsn::Max is a special value that means
-// "get the latest version present". It's used by the primary server, which knows that no one else
-// is writing WAL. 'not_modified_since' must be set to a proper value even if request_lsn is
-// Lsn::Max. Standby servers use the current replay LSN as the request LSN.
-//
-// not_modified_since: Hint to the pageserver that the client knows that the page has not been
-// modified between 'not_modified_since' and the request LSN. It's always correct to set
-// 'not_modified_since equal' to 'request_lsn' (unless Lsn::Max is used as the 'request_lsn'), but
-// passing an earlier LSN can speed up the request, by allowing the pageserver to process the
-// request without waiting for 'request_lsn' to arrive.
-//
-// The now-defunct V1 interface contained only one LSN, and a boolean 'latest' flag. The V1 interface was
-// sufficient for the primary; the 'lsn' was equivalent to the 'not_modified_since' value, and
-// 'latest' was set to true. The V2 interface was added because there was no correct way for a
-// standby to request a page at a particular non-latest LSN, and also include the
-// 'not_modified_since' hint. That led to an awkward choice of either using an old LSN in the
-// request, if the standby knows that the page hasn't been modified since, and risk getting an error
-// if that LSN has fallen behind the GC horizon, or requesting the current replay LSN, which could
-// require the pageserver unnecessarily to wait for the WAL to arrive up to that point. The new V2
-// interface allows sending both LSNs, and let the pageserver do the right thing. There was no
-// difference in the responses between V1 and V2.
-//
-// V3 version of protocol adds request ID to all requests. This request ID is also included in response
-// as well as other fields from requests, which allows to verify that we receive response for our request.
-// We copy fields from request to response to make checking more reliable: request ID is formed from process ID
-// and local counter, so in principle there can be duplicated requests IDs if process PID is reused.
-//
-#[derive(Debug, PartialEq, Eq, Clone, Copy)]
-pub enum PagestreamProtocolVersion {
-    V2,
-    V3,
-}
-
-pub type RequestId = u64;
-
-#[derive(Debug, Default, PartialEq, Eq, Clone, Copy)]
-pub struct PagestreamRequest {
-    pub reqid: RequestId,
-    pub request_lsn: Lsn,
-    pub not_modified_since: Lsn,
-}
-
-#[derive(Debug, PartialEq, Eq, Clone, Copy)]
-pub struct PagestreamExistsRequest {
-    pub hdr: PagestreamRequest,
-    pub rel: RelTag,
-}
-
-#[derive(Debug, PartialEq, Eq, Clone, Copy)]
-pub struct PagestreamNblocksRequest {
-    pub hdr: PagestreamRequest,
-    pub rel: RelTag,
-}
-
-#[derive(Debug, Default, PartialEq, Eq, Clone, Copy)]
-pub struct PagestreamGetPageRequest {
-    pub hdr: PagestreamRequest,
-    pub rel: RelTag,
-    pub blkno: u32,
-}
-
-#[derive(Debug, PartialEq, Eq, Clone, Copy)]
-pub struct PagestreamDbSizeRequest {
-    pub hdr: PagestreamRequest,
-    pub dbnode: u32,
-}
-
-#[derive(Debug, PartialEq, Eq, Clone, Copy)]
-pub struct PagestreamGetSlruSegmentRequest {
-    pub hdr: PagestreamRequest,
-    pub kind: u8,
-    pub segno: u32,
-}
-
-#[derive(Debug)]
-pub struct PagestreamExistsResponse {
-    pub req: PagestreamExistsRequest,
-    pub exists: bool,
-}
-
-#[derive(Debug)]
-pub struct PagestreamNblocksResponse {
-    pub req: PagestreamNblocksRequest,
-    pub n_blocks: u32,
-}
-
-#[derive(Debug)]
-pub struct PagestreamGetPageResponse {
-    pub req: PagestreamGetPageRequest,
-    pub page: Bytes,
-}
-
-#[derive(Debug)]
-pub struct PagestreamGetSlruSegmentResponse {
-    pub req: PagestreamGetSlruSegmentRequest,
-    pub segment: Bytes,
-}
-
-#[derive(Debug)]
-pub struct PagestreamErrorResponse {
-    pub req: PagestreamRequest,
-    pub message: String,
-}
-
-#[derive(Debug)]
-pub struct PagestreamDbSizeResponse {
-    pub req: PagestreamDbSizeRequest,
-    pub db_size: i64,
-}
-
-#[cfg(feature = "testing")]
-#[derive(Debug, PartialEq, Eq, Clone)]
-pub struct PagestreamTestRequest {
-    pub hdr: PagestreamRequest,
-    pub batch_key: u64,
-    pub message: String,
-}
-
-#[cfg(feature = "testing")]
-#[derive(Debug)]
-pub struct PagestreamTestResponse {
-    pub req: PagestreamTestRequest,
-}
-
-impl PagestreamFeMessage {
-    /// Serialize a compute -> pageserver message. This is currently only used in testing
-    /// tools. Always uses protocol version 3.
-    pub fn serialize(&self) -> Bytes {
-        let mut bytes = BytesMut::new();
-
-        match self {
-            Self::Exists(req) => {
-                bytes.put_u8(PagestreamFeMessageTag::Exists as u8);
-                bytes.put_u64(req.hdr.reqid);
-                bytes.put_u64(req.hdr.request_lsn.0);
-                bytes.put_u64(req.hdr.not_modified_since.0);
-                bytes.put_u32(req.rel.spcnode);
-                bytes.put_u32(req.rel.dbnode);
-                bytes.put_u32(req.rel.relnode);
-                bytes.put_u8(req.rel.forknum);
-            }
-
-            Self::Nblocks(req) => {
-                bytes.put_u8(PagestreamFeMessageTag::Nblocks as u8);
-                bytes.put_u64(req.hdr.reqid);
-                bytes.put_u64(req.hdr.request_lsn.0);
-                bytes.put_u64(req.hdr.not_modified_since.0);
-                bytes.put_u32(req.rel.spcnode);
-                bytes.put_u32(req.rel.dbnode);
-                bytes.put_u32(req.rel.relnode);
-                bytes.put_u8(req.rel.forknum);
-            }
-
-            Self::GetPage(req) => {
-                bytes.put_u8(PagestreamFeMessageTag::GetPage as u8);
-                bytes.put_u64(req.hdr.reqid);
-                bytes.put_u64(req.hdr.request_lsn.0);
-                bytes.put_u64(req.hdr.not_modified_since.0);
-                bytes.put_u32(req.rel.spcnode);
-                bytes.put_u32(req.rel.dbnode);
-                bytes.put_u32(req.rel.relnode);
-                bytes.put_u8(req.rel.forknum);
-                bytes.put_u32(req.blkno);
-            }
-
-            Self::DbSize(req) => {
-                bytes.put_u8(PagestreamFeMessageTag::DbSize as u8);
-                bytes.put_u64(req.hdr.reqid);
-                bytes.put_u64(req.hdr.request_lsn.0);
-                bytes.put_u64(req.hdr.not_modified_since.0);
-                bytes.put_u32(req.dbnode);
-            }
-
-            Self::GetSlruSegment(req) => {
-                bytes.put_u8(PagestreamFeMessageTag::GetSlruSegment as u8);
-                bytes.put_u64(req.hdr.reqid);
-                bytes.put_u64(req.hdr.request_lsn.0);
-                bytes.put_u64(req.hdr.not_modified_since.0);
-                bytes.put_u8(req.kind);
-                bytes.put_u32(req.segno);
-            }
-            #[cfg(feature = "testing")]
-            Self::Test(req) => {
-                bytes.put_u8(PagestreamFeMessageTag::Test as u8);
-                bytes.put_u64(req.hdr.reqid);
-                bytes.put_u64(req.hdr.request_lsn.0);
-                bytes.put_u64(req.hdr.not_modified_since.0);
-                bytes.put_u64(req.batch_key);
-                let message = req.message.as_bytes();
-                bytes.put_u64(message.len() as u64);
-                bytes.put_slice(message);
-            }
-        }
-
-        bytes.into()
-    }
-
-    pub fn parse<R: std::io::Read>(
-        body: &mut R,
-        protocol_version: PagestreamProtocolVersion,
-    ) -> anyhow::Result<PagestreamFeMessage> {
-        // these correspond to the NeonMessageTag enum in pagestore_client.h
-        //
-        // TODO: consider using protobuf or serde bincode for less error prone
-        // serialization.
-        let msg_tag = body.read_u8()?;
-        let (reqid, request_lsn, not_modified_since) = match protocol_version {
-            PagestreamProtocolVersion::V2 => (
-                0,
-                Lsn::from(body.read_u64::<BigEndian>()?),
-                Lsn::from(body.read_u64::<BigEndian>()?),
-            ),
-            PagestreamProtocolVersion::V3 => (
-                body.read_u64::<BigEndian>()?,
-                Lsn::from(body.read_u64::<BigEndian>()?),
-                Lsn::from(body.read_u64::<BigEndian>()?),
-            ),
-        };
-
-        match PagestreamFeMessageTag::try_from(msg_tag)
-            .map_err(|tag: u8| anyhow::anyhow!("invalid tag {tag}"))?
-        {
-            PagestreamFeMessageTag::Exists => {
-                Ok(PagestreamFeMessage::Exists(PagestreamExistsRequest {
-                    hdr: PagestreamRequest {
-                        reqid,
-                        request_lsn,
-                        not_modified_since,
-                    },
-                    rel: RelTag {
-                        spcnode: body.read_u32::<BigEndian>()?,
-                        dbnode: body.read_u32::<BigEndian>()?,
-                        relnode: body.read_u32::<BigEndian>()?,
-                        forknum: body.read_u8()?,
-                    },
-                }))
-            }
-            PagestreamFeMessageTag::Nblocks => {
-                Ok(PagestreamFeMessage::Nblocks(PagestreamNblocksRequest {
-                    hdr: PagestreamRequest {
-                        reqid,
-                        request_lsn,
-                        not_modified_since,
-                    },
-                    rel: RelTag {
-                        spcnode: body.read_u32::<BigEndian>()?,
-                        dbnode: body.read_u32::<BigEndian>()?,
-                        relnode: body.read_u32::<BigEndian>()?,
-                        forknum: body.read_u8()?,
-                    },
-                }))
-            }
-            PagestreamFeMessageTag::GetPage => {
-                Ok(PagestreamFeMessage::GetPage(PagestreamGetPageRequest {
-                    hdr: PagestreamRequest {
-                        reqid,
-                        request_lsn,
-                        not_modified_since,
-                    },
-                    rel: RelTag {
-                        spcnode: body.read_u32::<BigEndian>()?,
-                        dbnode: body.read_u32::<BigEndian>()?,
-                        relnode: body.read_u32::<BigEndian>()?,
-                        forknum: body.read_u8()?,
-                    },
-                    blkno: body.read_u32::<BigEndian>()?,
-                }))
-            }
-            PagestreamFeMessageTag::DbSize => {
-                Ok(PagestreamFeMessage::DbSize(PagestreamDbSizeRequest {
-                    hdr: PagestreamRequest {
-                        reqid,
-                        request_lsn,
-                        not_modified_since,
-                    },
-                    dbnode: body.read_u32::<BigEndian>()?,
-                }))
-            }
-            PagestreamFeMessageTag::GetSlruSegment => Ok(PagestreamFeMessage::GetSlruSegment(
-                PagestreamGetSlruSegmentRequest {
-                    hdr: PagestreamRequest {
-                        reqid,
-                        request_lsn,
-                        not_modified_since,
-                    },
-                    kind: body.read_u8()?,
-                    segno: body.read_u32::<BigEndian>()?,
-                },
-            )),
-            #[cfg(feature = "testing")]
-            PagestreamFeMessageTag::Test => Ok(PagestreamFeMessage::Test(PagestreamTestRequest {
-                hdr: PagestreamRequest {
-                    reqid,
-                    request_lsn,
-                    not_modified_since,
-                },
-                batch_key: body.read_u64::<BigEndian>()?,
-                message: {
-                    let len = body.read_u64::<BigEndian>()?;
-                    let mut buf = vec![0; len as usize];
-                    body.read_exact(&mut buf)?;
-                    String::from_utf8(buf)?
-                },
-            })),
-        }
-    }
-}
-
-impl PagestreamBeMessage {
-    pub fn serialize(&self, protocol_version: PagestreamProtocolVersion) -> Bytes {
-        let mut bytes = BytesMut::new();
-
-        use PagestreamBeMessageTag as Tag;
-        match protocol_version {
-            PagestreamProtocolVersion::V2 => {
-                match self {
-                    Self::Exists(resp) => {
-                        bytes.put_u8(Tag::Exists as u8);
-                        bytes.put_u8(resp.exists as u8);
-                    }
-
-                    Self::Nblocks(resp) => {
-                        bytes.put_u8(Tag::Nblocks as u8);
-                        bytes.put_u32(resp.n_blocks);
-                    }
-
-                    Self::GetPage(resp) => {
-                        bytes.put_u8(Tag::GetPage as u8);
-                        bytes.put(&resp.page[..])
-                    }
-
-                    Self::Error(resp) => {
-                        bytes.put_u8(Tag::Error as u8);
-                        bytes.put(resp.message.as_bytes());
-                        bytes.put_u8(0); // null terminator
-                    }
-                    Self::DbSize(resp) => {
-                        bytes.put_u8(Tag::DbSize as u8);
-                        bytes.put_i64(resp.db_size);
-                    }
-
-                    Self::GetSlruSegment(resp) => {
-                        bytes.put_u8(Tag::GetSlruSegment as u8);
-                        bytes.put_u32((resp.segment.len() / BLCKSZ) as u32);
-                        bytes.put(&resp.segment[..]);
-                    }
-
-                    #[cfg(feature = "testing")]
-                    Self::Test(resp) => {
-                        bytes.put_u8(Tag::Test as u8);
-                        bytes.put_u64(resp.req.batch_key);
-                        let message = resp.req.message.as_bytes();
-                        bytes.put_u64(message.len() as u64);
-                        bytes.put_slice(message);
-                    }
-                }
-            }
-            PagestreamProtocolVersion::V3 => {
-                match self {
-                    Self::Exists(resp) => {
-                        bytes.put_u8(Tag::Exists as u8);
-                        bytes.put_u64(resp.req.hdr.reqid);
-                        bytes.put_u64(resp.req.hdr.request_lsn.0);
-                        bytes.put_u64(resp.req.hdr.not_modified_since.0);
-                        bytes.put_u32(resp.req.rel.spcnode);
-                        bytes.put_u32(resp.req.rel.dbnode);
-                        bytes.put_u32(resp.req.rel.relnode);
-                        bytes.put_u8(resp.req.rel.forknum);
-                        bytes.put_u8(resp.exists as u8);
-                    }
-
-                    Self::Nblocks(resp) => {
-                        bytes.put_u8(Tag::Nblocks as u8);
-                        bytes.put_u64(resp.req.hdr.reqid);
-                        bytes.put_u64(resp.req.hdr.request_lsn.0);
-                        bytes.put_u64(resp.req.hdr.not_modified_since.0);
-                        bytes.put_u32(resp.req.rel.spcnode);
-                        bytes.put_u32(resp.req.rel.dbnode);
-                        bytes.put_u32(resp.req.rel.relnode);
-                        bytes.put_u8(resp.req.rel.forknum);
-                        bytes.put_u32(resp.n_blocks);
-                    }
-
-                    Self::GetPage(resp) => {
-                        bytes.put_u8(Tag::GetPage as u8);
-                        bytes.put_u64(resp.req.hdr.reqid);
-                        bytes.put_u64(resp.req.hdr.request_lsn.0);
-                        bytes.put_u64(resp.req.hdr.not_modified_since.0);
-                        bytes.put_u32(resp.req.rel.spcnode);
-                        bytes.put_u32(resp.req.rel.dbnode);
-                        bytes.put_u32(resp.req.rel.relnode);
-                        bytes.put_u8(resp.req.rel.forknum);
-                        bytes.put_u32(resp.req.blkno);
-                        bytes.put(&resp.page[..])
-                    }
-
-                    Self::Error(resp) => {
-                        bytes.put_u8(Tag::Error as u8);
-                        bytes.put_u64(resp.req.reqid);
-                        bytes.put_u64(resp.req.request_lsn.0);
-                        bytes.put_u64(resp.req.not_modified_since.0);
-                        bytes.put(resp.message.as_bytes());
-                        bytes.put_u8(0); // null terminator
-                    }
-                    Self::DbSize(resp) => {
-                        bytes.put_u8(Tag::DbSize as u8);
-                        bytes.put_u64(resp.req.hdr.reqid);
-                        bytes.put_u64(resp.req.hdr.request_lsn.0);
-                        bytes.put_u64(resp.req.hdr.not_modified_since.0);
-                        bytes.put_u32(resp.req.dbnode);
-                        bytes.put_i64(resp.db_size);
-                    }
-
-                    Self::GetSlruSegment(resp) => {
-                        bytes.put_u8(Tag::GetSlruSegment as u8);
-                        bytes.put_u64(resp.req.hdr.reqid);
-                        bytes.put_u64(resp.req.hdr.request_lsn.0);
-                        bytes.put_u64(resp.req.hdr.not_modified_since.0);
-                        bytes.put_u8(resp.req.kind);
-                        bytes.put_u32(resp.req.segno);
-                        bytes.put_u32((resp.segment.len() / BLCKSZ) as u32);
-                        bytes.put(&resp.segment[..]);
-                    }
-
-                    #[cfg(feature = "testing")]
-                    Self::Test(resp) => {
-                        bytes.put_u8(Tag::Test as u8);
-                        bytes.put_u64(resp.req.hdr.reqid);
-                        bytes.put_u64(resp.req.hdr.request_lsn.0);
-                        bytes.put_u64(resp.req.hdr.not_modified_since.0);
-                        bytes.put_u64(resp.req.batch_key);
-                        let message = resp.req.message.as_bytes();
-                        bytes.put_u64(message.len() as u64);
-                        bytes.put_slice(message);
-                    }
-                }
-            }
-        }
-        bytes.into()
-    }
-
-    pub fn deserialize(buf: Bytes) -> anyhow::Result<Self> {
-        let mut buf = buf.reader();
-        let msg_tag = buf.read_u8()?;
-
-        use PagestreamBeMessageTag as Tag;
-        let ok =
-            match Tag::try_from(msg_tag).map_err(|tag: u8| anyhow::anyhow!("invalid tag {tag}"))? {
-                Tag::Exists => {
-                    let reqid = buf.read_u64::<BigEndian>()?;
-                    let request_lsn = Lsn(buf.read_u64::<BigEndian>()?);
-                    let not_modified_since = Lsn(buf.read_u64::<BigEndian>()?);
-                    let rel = RelTag {
-                        spcnode: buf.read_u32::<BigEndian>()?,
-                        dbnode: buf.read_u32::<BigEndian>()?,
-                        relnode: buf.read_u32::<BigEndian>()?,
-                        forknum: buf.read_u8()?,
-                    };
-                    let exists = buf.read_u8()? != 0;
-                    Self::Exists(PagestreamExistsResponse {
-                        req: PagestreamExistsRequest {
-                            hdr: PagestreamRequest {
-                                reqid,
-                                request_lsn,
-                                not_modified_since,
-                            },
-                            rel,
-                        },
-                        exists,
-                    })
-                }
-                Tag::Nblocks => {
-                    let reqid = buf.read_u64::<BigEndian>()?;
-                    let request_lsn = Lsn(buf.read_u64::<BigEndian>()?);
-                    let not_modified_since = Lsn(buf.read_u64::<BigEndian>()?);
-                    let rel = RelTag {
-                        spcnode: buf.read_u32::<BigEndian>()?,
-                        dbnode: buf.read_u32::<BigEndian>()?,
-                        relnode: buf.read_u32::<BigEndian>()?,
-                        forknum: buf.read_u8()?,
-                    };
-                    let n_blocks = buf.read_u32::<BigEndian>()?;
-                    Self::Nblocks(PagestreamNblocksResponse {
-                        req: PagestreamNblocksRequest {
-                            hdr: PagestreamRequest {
-                                reqid,
-                                request_lsn,
-                                not_modified_since,
-                            },
-                            rel,
-                        },
-                        n_blocks,
-                    })
-                }
-                Tag::GetPage => {
-                    let reqid = buf.read_u64::<BigEndian>()?;
-                    let request_lsn = Lsn(buf.read_u64::<BigEndian>()?);
-                    let not_modified_since = Lsn(buf.read_u64::<BigEndian>()?);
-                    let rel = RelTag {
-                        spcnode: buf.read_u32::<BigEndian>()?,
-                        dbnode: buf.read_u32::<BigEndian>()?,
-                        relnode: buf.read_u32::<BigEndian>()?,
-                        forknum: buf.read_u8()?,
-                    };
-                    let blkno = buf.read_u32::<BigEndian>()?;
-                    let mut page = vec![0; 8192]; // TODO: use MaybeUninit
-                    buf.read_exact(&mut page)?;
-                    Self::GetPage(PagestreamGetPageResponse {
-                        req: PagestreamGetPageRequest {
-                            hdr: PagestreamRequest {
-                                reqid,
-                                request_lsn,
-                                not_modified_since,
-                            },
-                            rel,
-                            blkno,
-                        },
-                        page: page.into(),
-                    })
-                }
-                Tag::Error => {
-                    let reqid = buf.read_u64::<BigEndian>()?;
-                    let request_lsn = Lsn(buf.read_u64::<BigEndian>()?);
-                    let not_modified_since = Lsn(buf.read_u64::<BigEndian>()?);
-                    let mut msg = Vec::new();
-                    buf.read_until(0, &mut msg)?;
-                    let cstring = std::ffi::CString::from_vec_with_nul(msg)?;
-                    let rust_str = cstring.to_str()?;
-                    Self::Error(PagestreamErrorResponse {
-                        req: PagestreamRequest {
-                            reqid,
-                            request_lsn,
-                            not_modified_since,
-                        },
-                        message: rust_str.to_owned(),
-                    })
-                }
-                Tag::DbSize => {
-                    let reqid = buf.read_u64::<BigEndian>()?;
-                    let request_lsn = Lsn(buf.read_u64::<BigEndian>()?);
-                    let not_modified_since = Lsn(buf.read_u64::<BigEndian>()?);
-                    let dbnode = buf.read_u32::<BigEndian>()?;
-                    let db_size = buf.read_i64::<BigEndian>()?;
-                    Self::DbSize(PagestreamDbSizeResponse {
-                        req: PagestreamDbSizeRequest {
-                            hdr: PagestreamRequest {
-                                reqid,
-                                request_lsn,
-                                not_modified_since,
-                            },
-                            dbnode,
-                        },
-                        db_size,
-                    })
-                }
-                Tag::GetSlruSegment => {
-                    let reqid = buf.read_u64::<BigEndian>()?;
-                    let request_lsn = Lsn(buf.read_u64::<BigEndian>()?);
-                    let not_modified_since = Lsn(buf.read_u64::<BigEndian>()?);
-                    let kind = buf.read_u8()?;
-                    let segno = buf.read_u32::<BigEndian>()?;
-                    let n_blocks = buf.read_u32::<BigEndian>()?;
-                    let mut segment = vec![0; n_blocks as usize * BLCKSZ];
-                    buf.read_exact(&mut segment)?;
-                    Self::GetSlruSegment(PagestreamGetSlruSegmentResponse {
-                        req: PagestreamGetSlruSegmentRequest {
-                            hdr: PagestreamRequest {
-                                reqid,
-                                request_lsn,
-                                not_modified_since,
-                            },
-                            kind,
-                            segno,
-                        },
-                        segment: segment.into(),
-                    })
-                }
-                #[cfg(feature = "testing")]
-                Tag::Test => {
-                    let reqid = buf.read_u64::<BigEndian>()?;
-                    let request_lsn = Lsn(buf.read_u64::<BigEndian>()?);
-                    let not_modified_since = Lsn(buf.read_u64::<BigEndian>()?);
-                    let batch_key = buf.read_u64::<BigEndian>()?;
-                    let len = buf.read_u64::<BigEndian>()?;
-                    let mut msg = vec![0; len as usize];
-                    buf.read_exact(&mut msg)?;
-                    let message = String::from_utf8(msg)?;
-                    Self::Test(PagestreamTestResponse {
-                        req: PagestreamTestRequest {
-                            hdr: PagestreamRequest {
-                                reqid,
-                                request_lsn,
-                                not_modified_since,
-                            },
-                            batch_key,
-                            message,
-                        },
-                    })
-                }
-            };
-        let remaining = buf.into_inner();
-        if !remaining.is_empty() {
-            anyhow::bail!(
-                "remaining bytes in msg with tag={msg_tag}: {}",
-                remaining.len()
-            );
-        }
-        Ok(ok)
-    }
-
-    pub fn kind(&self) -> &'static str {
-        match self {
-            Self::Exists(_) => "Exists",
-            Self::Nblocks(_) => "Nblocks",
-            Self::GetPage(_) => "GetPage",
-            Self::Error(_) => "Error",
-            Self::DbSize(_) => "DbSize",
-            Self::GetSlruSegment(_) => "GetSlruSegment",
-            #[cfg(feature = "testing")]
-            Self::Test(_) => "Test",
-        }
-    }
-}
-
-#[cfg(test)]
-mod tests {
-    use super::*;
-
-    #[test]
-    fn test_pagestream() {
-        // Test serialization/deserialization of PagestreamFeMessage
-        let messages = vec![
-            PagestreamFeMessage::Exists(PagestreamExistsRequest {
-                hdr: PagestreamRequest {
-                    reqid: 0,
-                    request_lsn: Lsn(4),
-                    not_modified_since: Lsn(3),
-                },
-                rel: RelTag {
-                    forknum: 1,
-                    spcnode: 2,
-                    dbnode: 3,
-                    relnode: 4,
-                },
-            }),
-            PagestreamFeMessage::Nblocks(PagestreamNblocksRequest {
-                hdr: PagestreamRequest {
-                    reqid: 0,
-                    request_lsn: Lsn(4),
-                    not_modified_since: Lsn(4),
-                },
-                rel: RelTag {
-                    forknum: 1,
-                    spcnode: 2,
-                    dbnode: 3,
-                    relnode: 4,
-                },
-            }),
-            PagestreamFeMessage::GetPage(PagestreamGetPageRequest {
-                hdr: PagestreamRequest {
-                    reqid: 0,
-                    request_lsn: Lsn(4),
-                    not_modified_since: Lsn(3),
-                },
-                rel: RelTag {
-                    forknum: 1,
-                    spcnode: 2,
-                    dbnode: 3,
-                    relnode: 4,
-                },
-                blkno: 7,
-            }),
-            PagestreamFeMessage::DbSize(PagestreamDbSizeRequest {
-                hdr: PagestreamRequest {
-                    reqid: 0,
-                    request_lsn: Lsn(4),
-                    not_modified_since: Lsn(3),
-                },
-                dbnode: 7,
-            }),
-        ];
-        for msg in messages {
-            let bytes = msg.serialize();
-            let reconstructed =
-                PagestreamFeMessage::parse(&mut bytes.reader(), PagestreamProtocolVersion::V3)
-                    .unwrap();
-            assert!(msg == reconstructed);
-        }
-    }
-}

libs/pageserver_api/src/record.rs

@@ -128,6 +128,6 @@ pub fn describe_wal_record(rec: &NeonWalRecord) -> Result<String, DeserializeErr
             will_init,
             describe_postgres_wal_record(rec)?
         )),
-        _ => Ok(format!("{rec:?}")),
+        _ => Ok(format!("{:?}", rec)),
     }
 }

libs/pageserver_api/src/reltag.rs

@@ -1,9 +1,9 @@
 use std::cmp::Ordering;
 use std::fmt;
 
-use postgres_ffi_types::Oid;
-use postgres_ffi_types::constants::GLOBALTABLESPACE_OID;
-use postgres_ffi_types::forknum::{MAIN_FORKNUM, forkname_to_number, forknumber_to_name};
+use postgres_ffi::Oid;
+use postgres_ffi::pg_constants::GLOBALTABLESPACE_OID;
+use postgres_ffi::relfile_utils::{MAIN_FORKNUM, forkname_to_number, forknumber_to_name};
 use serde::{Deserialize, Serialize};
 
 ///

libs/pageserver_api/src/shard.rs

@@ -35,7 +35,7 @@ use std::hash::{Hash, Hasher};
 
 #[doc(inline)]
 pub use ::utils::shard::*;
-use postgres_ffi_types::forknum::INIT_FORKNUM;
+use postgres_ffi::relfile_utils::INIT_FORKNUM;
 use serde::{Deserialize, Serialize};
 
 use crate::key::Key;

libs/pageserver_api/src/upcall_api.rs

@@ -9,7 +9,7 @@ use utils::id::{NodeId, TimelineId};
 
 use crate::controller_api::NodeRegisterRequest;
 use crate::models::{LocationConfigMode, ShardImportStatus};
-use crate::shard::{ShardStripeSize, TenantShardId};
+use crate::shard::TenantShardId;
 
 /// Upcall message sent by the pageserver to the configured `control_plane_api` on
 /// startup.
@@ -23,13 +23,19 @@ pub struct ReAttachRequest {
     pub register: Option<NodeRegisterRequest>,
 }
 
+fn default_mode() -> LocationConfigMode {
+    LocationConfigMode::AttachedSingle
+}
+
 #[derive(Serialize, Deserialize, Debug)]
 pub struct ReAttachResponseTenant {
     pub id: TenantShardId,
     /// Mandatory if LocationConfigMode is None or set to an Attached* mode
     pub r#gen: Option<u32>,
+
+    /// Default value only for backward compat: this field should be set
+    #[serde(default = "default_mode")]
     pub mode: LocationConfigMode,
-    pub stripe_size: ShardStripeSize,
 }
 #[derive(Serialize, Deserialize)]
 pub struct ReAttachResponse {

libs/pageserver_api/src/value.rs

@@ -10,7 +10,7 @@
 use bytes::Bytes;
 use serde::{Deserialize, Serialize};
 
-use crate::models::record::NeonWalRecord;
+use crate::record::NeonWalRecord;
 
 #[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)]
 pub enum Value {

libs/postgres_backend/src/lib.rs

@@ -939,7 +939,7 @@ impl<IO: AsyncRead + AsyncWrite + Unpin> PostgresBackendReader<IO> {
                 FeMessage::CopyFail => Err(CopyStreamHandlerEnd::CopyFail),
                 FeMessage::Terminate => Err(CopyStreamHandlerEnd::Terminate),
                 _ => Err(CopyStreamHandlerEnd::from(ConnectionError::Protocol(
-                    ProtocolError::Protocol(format!("unexpected message in COPY stream {msg:?}")),
+                    ProtocolError::Protocol(format!("unexpected message in COPY stream {:?}", msg)),
                 ))),
             },
             None => Err(CopyStreamHandlerEnd::EOF),

libs/postgres_backend/tests/simple_select.rs

@@ -61,7 +61,7 @@ async fn simple_select() {
     // so spawn it off to run on its own.
     tokio::spawn(async move {
         if let Err(e) = connection.await {
-            eprintln!("connection error: {e}");
+            eprintln!("connection error: {}", e);
         }
     });
 
@@ -137,7 +137,7 @@ async fn simple_select_ssl() {
     // so spawn it off to run on its own.
     tokio::spawn(async move {
         if let Err(e) = connection.await {
-            eprintln!("connection error: {e}");
+            eprintln!("connection error: {}", e);
         }
     });
 

libs/postgres_connection/src/lib.rs

@@ -223,7 +223,7 @@ mod tests_pg_connection_config {
         assert_eq!(cfg.port(), 123);
         assert_eq!(cfg.raw_address(), "stub.host.example:123");
         assert_eq!(
-            format!("{cfg:?}"),
+            format!("{:?}", cfg),
             "PgConnectionConfig { host: Domain(\"stub.host.example\"), port: 123, password: None }"
         );
     }
@@ -239,7 +239,7 @@ mod tests_pg_connection_config {
         assert_eq!(cfg.port(), 123);
         assert_eq!(cfg.raw_address(), "[::1]:123");
         assert_eq!(
-            format!("{cfg:?}"),
+            format!("{:?}", cfg),
             "PgConnectionConfig { host: Ipv6(::1), port: 123, password: None }"
         );
     }
@@ -252,7 +252,7 @@ mod tests_pg_connection_config {
         assert_eq!(cfg.port(), 123);
         assert_eq!(cfg.raw_address(), "stub.host.example:123");
         assert_eq!(
-            format!("{cfg:?}"),
+            format!("{:?}", cfg),
             "PgConnectionConfig { host: Domain(\"stub.host.example\"), port: 123, password: Some(REDACTED-STRING) }"
         );
     }

libs/postgres_ffi/Cargo.toml

@@ -16,7 +16,6 @@ memoffset.workspace = true
 pprof.workspace = true
 thiserror.workspace = true
 serde.workspace = true
-postgres_ffi_types.workspace = true
 utils.workspace = true
 tracing.workspace = true
 

libs/postgres_ffi/src/pg_constants.rs

@@ -11,7 +11,11 @@
 
 use crate::{BLCKSZ, PageHeaderData};
 
-// Note: There are a few more widely-used constants in the postgres_ffi_types::constants crate.
+//
+// From pg_tablespace_d.h
+//
+pub const DEFAULTTABLESPACE_OID: u32 = 1663;
+pub const GLOBALTABLESPACE_OID: u32 = 1664;
 
 // From storage_xlog.h
 pub const XLOG_SMGR_CREATE: u8 = 0x10;

libs/postgres_ffi/src/relfile_utils.rs

@@ -4,7 +4,50 @@
 use once_cell::sync::OnceCell;
 use regex::Regex;
 
-use postgres_ffi_types::forknum::*;
+//
+// Fork numbers, from relpath.h
+//
+pub const MAIN_FORKNUM: u8 = 0;
+pub const FSM_FORKNUM: u8 = 1;
+pub const VISIBILITYMAP_FORKNUM: u8 = 2;
+pub const INIT_FORKNUM: u8 = 3;
+
+#[derive(Debug, Clone, thiserror::Error, PartialEq, Eq)]
+pub enum FilePathError {
+    #[error("invalid relation fork name")]
+    InvalidForkName,
+    #[error("invalid relation data file name")]
+    InvalidFileName,
+}
+
+impl From<core::num::ParseIntError> for FilePathError {
+    fn from(_e: core::num::ParseIntError) -> Self {
+        FilePathError::InvalidFileName
+    }
+}
+
+/// Convert Postgres relation file's fork suffix to fork number.
+pub fn forkname_to_number(forkname: Option<&str>) -> Result<u8, FilePathError> {
+    match forkname {
+        // "main" is not in filenames, it's implicit if the fork name is not present
+        None => Ok(MAIN_FORKNUM),
+        Some("fsm") => Ok(FSM_FORKNUM),
+        Some("vm") => Ok(VISIBILITYMAP_FORKNUM),
+        Some("init") => Ok(INIT_FORKNUM),
+        Some(_) => Err(FilePathError::InvalidForkName),
+    }
+}
+
+/// Convert Postgres fork number to the right suffix of the relation data file.
+pub fn forknumber_to_name(forknum: u8) -> Option<&'static str> {
+    match forknum {
+        MAIN_FORKNUM => None,
+        FSM_FORKNUM => Some("fsm"),
+        VISIBILITYMAP_FORKNUM => Some("vm"),
+        INIT_FORKNUM => Some("init"),
+        _ => Some("UNKNOWN FORKNUM"),
+    }
+}
 
 /// Parse a filename of a relation file. Returns (relfilenode, forknum, segno) tuple.
 ///
@@ -32,9 +75,7 @@ pub fn parse_relfilename(fname: &str) -> Result<(u32, u8, u32), FilePathError> {
         .ok_or(FilePathError::InvalidFileName)?;
 
     let relnode_str = caps.name("relnode").unwrap().as_str();
-    let relnode = relnode_str
-        .parse::<u32>()
-        .map_err(|_e| FilePathError::InvalidFileName)?;
+    let relnode = relnode_str.parse::<u32>()?;
 
     let forkname = caps.name("forkname").map(|f| f.as_str());
     let forknum = forkname_to_number(forkname)?;
@@ -43,11 +84,7 @@ pub fn parse_relfilename(fname: &str) -> Result<(u32, u8, u32), FilePathError> {
     let segno = if segno_match.is_none() {
         0
     } else {
-        segno_match
-            .unwrap()
-            .as_str()
-            .parse::<u32>()
-            .map_err(|_e| FilePathError::InvalidFileName)?
+        segno_match.unwrap().as_str().parse::<u32>()?
     };
 
     Ok((relnode, forknum, segno))

libs/postgres_ffi/src/waldecoder_handler.rs

@@ -114,7 +114,7 @@ impl WalStreamDecoderHandler for WalStreamDecoder {
 
                         let hdr = XLogLongPageHeaderData::from_bytes(&mut self.inputbuf).map_err(
                             |e| WalDecodeError {
-                                msg: format!("long header deserialization failed {e}"),
+                                msg: format!("long header deserialization failed {}", e),
                                 lsn: self.lsn,
                             },
                         )?;
@@ -130,7 +130,7 @@ impl WalStreamDecoderHandler for WalStreamDecoder {
                         let hdr =
                             XLogPageHeaderData::from_bytes(&mut self.inputbuf).map_err(|e| {
                                 WalDecodeError {
-                                    msg: format!("header deserialization failed {e}"),
+                                    msg: format!("header deserialization failed {}", e),
                                     lsn: self.lsn,
                                 }
                             })?;
@@ -155,7 +155,7 @@ impl WalStreamDecoderHandler for WalStreamDecoder {
                     let xl_tot_len = (&self.inputbuf[0..4]).get_u32_le();
                     if (xl_tot_len as usize) < XLOG_SIZE_OF_XLOG_RECORD {
                         return Err(WalDecodeError {
-                            msg: format!("invalid xl_tot_len {xl_tot_len}"),
+                            msg: format!("invalid xl_tot_len {}", xl_tot_len),
                             lsn: self.lsn,
                         });
                     }
@@ -218,7 +218,7 @@ impl WalStreamDecoderHandler for WalStreamDecoder {
         let xlogrec =
             XLogRecord::from_slice(&recordbuf[0..XLOG_SIZE_OF_XLOG_RECORD]).map_err(|e| {
                 WalDecodeError {
-                    msg: format!("xlog record deserialization failed {e}"),
+                    msg: format!("xlog record deserialization failed {}", e),
                     lsn: self.lsn,
                 }
             })?;

libs/postgres_ffi/src/walrecord.rs

@@ -1199,7 +1199,7 @@ pub fn describe_postgres_wal_record(record: &Bytes) -> Result<String, Deserializ
                 pg_constants::XLOG_HEAP2_MULTI_INSERT => "HEAP2 MULTI_INSERT",
                 pg_constants::XLOG_HEAP2_VISIBLE => "HEAP2 VISIBLE",
                 _ => {
-                    unknown_str = format!("HEAP2 UNKNOWN_0x{info:02x}");
+                    unknown_str = format!("HEAP2 UNKNOWN_0x{:02x}", info);
                     &unknown_str
                 }
             }
@@ -1212,7 +1212,7 @@ pub fn describe_postgres_wal_record(record: &Bytes) -> Result<String, Deserializ
                 pg_constants::XLOG_HEAP_UPDATE => "HEAP UPDATE",
                 pg_constants::XLOG_HEAP_HOT_UPDATE => "HEAP HOT_UPDATE",
                 _ => {
-                    unknown_str = format!("HEAP2 UNKNOWN_0x{info:02x}");
+                    unknown_str = format!("HEAP2 UNKNOWN_0x{:02x}", info);
                     &unknown_str
                 }
             }
@@ -1223,7 +1223,7 @@ pub fn describe_postgres_wal_record(record: &Bytes) -> Result<String, Deserializ
                 pg_constants::XLOG_FPI => "XLOG FPI",
                 pg_constants::XLOG_FPI_FOR_HINT => "XLOG FPI_FOR_HINT",
                 _ => {
-                    unknown_str = format!("XLOG UNKNOWN_0x{info:02x}");
+                    unknown_str = format!("XLOG UNKNOWN_0x{:02x}", info);
                     &unknown_str
                 }
             }
@@ -1231,7 +1231,7 @@ pub fn describe_postgres_wal_record(record: &Bytes) -> Result<String, Deserializ
         rmid => {
             let info = xlogrec.xl_info & pg_constants::XLR_RMGR_INFO_MASK;
 
-            unknown_str = format!("UNKNOWN_RM_{rmid} INFO_0x{info:02x}");
+            unknown_str = format!("UNKNOWN_RM_{} INFO_0x{:02x}", rmid, info);
             &unknown_str
         }
     };

libs/postgres_ffi/wal_craft/src/xlog_utils_test.rs

@@ -34,7 +34,7 @@ fn test_end_of_wal<C: crate::Crafter>(test_name: &str) {
     let cfg = Conf {
         pg_version,
         pg_distrib_dir: top_path.join("pg_install"),
-        datadir: top_path.join(format!("test_output/{test_name}-{PG_MAJORVERSION}")),
+        datadir: top_path.join(format!("test_output/{}-{PG_MAJORVERSION}", test_name)),
     };
     if cfg.datadir.exists() {
         fs::remove_dir_all(&cfg.datadir).unwrap();

libs/postgres_ffi_types/Cargo.toml

@@ -1,11 +0,0 @@
-[package]
-name = "postgres_ffi_types"
-version = "0.1.0"
-edition.workspace = true
-license.workspace = true
-
-[dependencies]
-thiserror.workspace = true
-workspace_hack = { version = "0.1", path = "../../workspace_hack" }
-
-[dev-dependencies]

libs/postgres_ffi_types/src/constants.rs

@@ -1,8 +0,0 @@
-//! Misc constants, copied from PostgreSQL headers.
-//!
-//! Any constants included here must be the same in all PostgreSQL versions and unlikely to change
-//! in the future either!
-
-// From pg_tablespace_d.h
-pub const DEFAULTTABLESPACE_OID: u32 = 1663;
-pub const GLOBALTABLESPACE_OID: u32 = 1664;

libs/postgres_ffi_types/src/forknum.rs

@@ -1,36 +0,0 @@
-// Fork numbers, from relpath.h
-pub const MAIN_FORKNUM: u8 = 0;
-pub const FSM_FORKNUM: u8 = 1;
-pub const VISIBILITYMAP_FORKNUM: u8 = 2;
-pub const INIT_FORKNUM: u8 = 3;
-
-#[derive(Debug, Clone, thiserror::Error, PartialEq, Eq)]
-pub enum FilePathError {
-    #[error("invalid relation fork name")]
-    InvalidForkName,
-    #[error("invalid relation data file name")]
-    InvalidFileName,
-}
-
-/// Convert Postgres relation file's fork suffix to fork number.
-pub fn forkname_to_number(forkname: Option<&str>) -> Result<u8, FilePathError> {
-    match forkname {
-        // "main" is not in filenames, it's implicit if the fork name is not present
-        None => Ok(MAIN_FORKNUM),
-        Some("fsm") => Ok(FSM_FORKNUM),
-        Some("vm") => Ok(VISIBILITYMAP_FORKNUM),
-        Some("init") => Ok(INIT_FORKNUM),
-        Some(_) => Err(FilePathError::InvalidForkName),
-    }
-}
-
-/// Convert Postgres fork number to the right suffix of the relation data file.
-pub fn forknumber_to_name(forknum: u8) -> Option<&'static str> {
-    match forknum {
-        MAIN_FORKNUM => None,
-        FSM_FORKNUM => Some("fsm"),
-        VISIBILITYMAP_FORKNUM => Some("vm"),
-        INIT_FORKNUM => Some("init"),
-        _ => Some("UNKNOWN FORKNUM"),
-    }
-}

libs/postgres_ffi_types/src/lib.rs

@@ -1,13 +0,0 @@
-//! This package contains some PostgreSQL constants and datatypes that are the same in all versions
-//! of PostgreSQL and unlikely to change in the future either. These could be derived from the
-//! PostgreSQL headers with 'bindgen', but in order to avoid proliferating the dependency to bindgen
-//! and the PostgreSQL C headers to all services, we prefer to have this small stand-alone crate for
-//! them instead.
-//!
-//! Be mindful in what you add here, as these types are deeply ingrained in the APIs.
-
-pub mod constants;
-pub mod forknum;
-
-pub type Oid = u32;
-pub type RepOriginId = u16;

libs/postgres_initdb/src/lib.rs

@@ -31,15 +31,15 @@ pub enum Error {
 impl fmt::Display for Error {
     fn fmt(&self, f: &mut fmt::Formatter) -> fmt::Result {
         match self {
-            Error::Spawn(e) => write!(f, "Error spawning command: {e:?}"),
+            Error::Spawn(e) => write!(f, "Error spawning command: {:?}", e),
             Error::Failed { status, stderr } => write!(
                 f,
                 "Command failed with status {:?}: {}",
                 status,
                 String::from_utf8_lossy(stderr)
             ),
-            Error::WaitOutput(e) => write!(f, "Error waiting for command output: {e:?}"),
-            Error::Other(e) => write!(f, "Error: {e:?}"),
+            Error::WaitOutput(e) => write!(f, "Error waiting for command output: {:?}", e),
+            Error::Other(e) => write!(f, "Error: {:?}", e),
         }
     }
 }

libs/posthog_client_lite/src/background_loop.rs

@@ -36,10 +36,7 @@ impl FeatureResolverBackgroundLoop {
         // Main loop of updating the feature flags.
         handle.spawn(
             async move {
-                tracing::info!(
-                    "Starting PostHog feature resolver with refresh period: {:?}",
-                    refresh_period
-                );
+                tracing::info!("Starting PostHog feature resolver");
                 let mut ticker = tokio::time::interval(refresh_period);
                 ticker.set_missed_tick_behavior(tokio::time::MissedTickBehavior::Skip);
                 loop {
@@ -58,16 +55,9 @@ impl FeatureResolverBackgroundLoop {
                             continue;
                         }
                     };
-                    let project_id = this.posthog_client.config.project_id.parse::<u64>().ok();
-                    match FeatureStore::new_with_flags(resp.flags, project_id) {
-                        Ok(feature_store) => {
-                            this.feature_store.store(Arc::new(feature_store));
-                            tracing::info!("Feature flag updated");
-                        }
-                        Err(e) => {
-                            tracing::warn!("Cannot process feature flag spec: {}", e);
-                        }
-                    }
+                    let feature_store = FeatureStore::new_with_flags(resp.flags);
+                    this.feature_store.store(Arc::new(feature_store));
+                    tracing::info!("Feature flag updated");
                 }
                 tracing::info!("PostHog feature resolver stopped");
             }

libs/posthog_client_lite/src/lib.rs

@@ -39,9 +39,6 @@ pub struct LocalEvaluationResponse {
 
 #[derive(Deserialize)]
 pub struct LocalEvaluationFlag {
-    #[allow(dead_code)]
-    id: u64,
-    team_id: u64,
     key: String,
     filters: LocalEvaluationFlagFilters,
     active: bool,
@@ -110,32 +107,17 @@ impl FeatureStore {
         }
     }
 
-    pub fn new_with_flags(
-        flags: Vec<LocalEvaluationFlag>,
-        project_id: Option<u64>,
-    ) -> Result<Self, &'static str> {
+    pub fn new_with_flags(flags: Vec<LocalEvaluationFlag>) -> Self {
         let mut store = Self::new();
-        store.set_flags(flags, project_id)?;
-        Ok(store)
+        store.set_flags(flags);
+        store
     }
 
-    pub fn set_flags(
-        &mut self,
-        flags: Vec<LocalEvaluationFlag>,
-        project_id: Option<u64>,
-    ) -> Result<(), &'static str> {
+    pub fn set_flags(&mut self, flags: Vec<LocalEvaluationFlag>) {
         self.flags.clear();
         for flag in flags {
-            if let Some(project_id) = project_id {
-                if flag.team_id != project_id {
-                    return Err(
-                        "Retrieved a spec with different project id, wrong config? Discarding the feature flags.",
-                    );
-                }
-            }
             self.flags.insert(flag.key.clone(), flag);
         }
-        Ok(())
     }
 
     /// Generate a consistent hash for a user ID (e.g., tenant ID).
@@ -168,13 +150,15 @@ impl FeatureStore {
                 let PostHogFlagFilterPropertyValue::String(provided) = provided else {
                     // Left should be a string
                     return Err(PostHogEvaluationError::Internal(format!(
-                        "The left side of the condition is not a string: {provided:?}"
+                        "The left side of the condition is not a string: {:?}",
+                        provided
                     )));
                 };
                 let PostHogFlagFilterPropertyValue::List(requested) = requested else {
                     // Right should be a list of string
                     return Err(PostHogEvaluationError::Internal(format!(
-                        "The right side of the condition is not a list: {requested:?}"
+                        "The right side of the condition is not a list: {:?}",
+                        requested
                     )));
                 };
                 Ok(requested.contains(provided))
@@ -183,12 +167,14 @@ impl FeatureStore {
                 let PostHogFlagFilterPropertyValue::String(requested) = requested else {
                     // Right should be a string
                     return Err(PostHogEvaluationError::Internal(format!(
-                        "The right side of the condition is not a string: {requested:?}"
+                        "The right side of the condition is not a string: {:?}",
+                        requested
                     )));
                 };
                 let Ok(requested) = requested.parse::<f64>() else {
                     return Err(PostHogEvaluationError::Internal(format!(
-                        "Can not parse the right side of the condition as a number: {requested:?}"
+                        "Can not parse the right side of the condition as a number: {:?}",
+                        requested
                     )));
                 };
                 // Left can either be a number or a string
@@ -197,14 +183,16 @@ impl FeatureStore {
                     PostHogFlagFilterPropertyValue::String(provided) => {
                         let Ok(provided) = provided.parse::<f64>() else {
                             return Err(PostHogEvaluationError::Internal(format!(
-                                "Can not parse the left side of the condition as a number: {provided:?}"
+                                "Can not parse the left side of the condition as a number: {:?}",
+                                provided
                             )));
                         };
                         provided
                     }
                     _ => {
                         return Err(PostHogEvaluationError::Internal(format!(
-                            "The left side of the condition is not a number or a string: {provided:?}"
+                            "The left side of the condition is not a number or a string: {:?}",
+                            provided
                         )));
                     }
                 };
@@ -212,12 +200,14 @@ impl FeatureStore {
                     "lt" => Ok(provided < requested),
                     "gt" => Ok(provided > requested),
                     op => Err(PostHogEvaluationError::Internal(format!(
-                        "Unsupported operator: {op}"
+                        "Unsupported operator: {}",
+                        op
                     ))),
                 }
             }
             _ => Err(PostHogEvaluationError::Internal(format!(
-                "Unsupported operator: {operator}"
+                "Unsupported operator: {}",
+                operator
             ))),
         }
     }
@@ -365,7 +355,8 @@ impl FeatureStore {
         if let Some(flag_config) = self.flags.get(flag_key) {
             if !flag_config.active {
                 return Err(PostHogEvaluationError::NotAvailable(format!(
-                    "The feature flag is not active: {flag_key}"
+                    "The feature flag is not active: {}",
+                    flag_key
                 )));
             }
             let Some(ref multivariate) = flag_config.filters.multivariate else {
@@ -392,7 +383,8 @@ impl FeatureStore {
                         // This should not happen because the rollout percentage always adds up to 100, but just in case that PostHog
                         // returned invalid spec, we return an error.
                         return Err(PostHogEvaluationError::Internal(format!(
-                            "Rollout percentage does not add up to 100: {flag_key}"
+                            "Rollout percentage does not add up to 100: {}",
+                            flag_key
                         )));
                     }
                     GroupEvaluationResult::Unmatched => continue,
@@ -403,7 +395,8 @@ impl FeatureStore {
         } else {
             // The feature flag is not available yet
             Err(PostHogEvaluationError::NotAvailable(format!(
-                "Not found in the local evaluation spec: {flag_key}"
+                "Not found in the local evaluation spec: {}",
+                flag_key
             )))
         }
     }
@@ -429,7 +422,8 @@ impl FeatureStore {
         if let Some(flag_config) = self.flags.get(flag_key) {
             if !flag_config.active {
                 return Err(PostHogEvaluationError::NotAvailable(format!(
-                    "The feature flag is not active: {flag_key}"
+                    "The feature flag is not active: {}",
+                    flag_key
                 )));
             }
             if flag_config.filters.multivariate.is_some() {
@@ -444,7 +438,8 @@ impl FeatureStore {
                 match self.evaluate_group(group, hash_on_global_rollout_percentage, properties)? {
                     GroupEvaluationResult::MatchedAndOverride(_) => {
                         return Err(PostHogEvaluationError::Internal(format!(
-                            "Boolean flag cannot have overrides: {flag_key}"
+                            "Boolean flag cannot have overrides: {}",
+                            flag_key
                         )));
                     }
                     GroupEvaluationResult::MatchedAndEvaluate => {
@@ -458,7 +453,8 @@ impl FeatureStore {
         } else {
             // The feature flag is not available yet
             Err(PostHogEvaluationError::NotAvailable(format!(
-                "Not found in the local evaluation spec: {flag_key}"
+                "Not found in the local evaluation spec: {}",
+                flag_key
             )))
         }
     }
@@ -469,7 +465,8 @@ impl FeatureStore {
             Ok(flag_config.filters.multivariate.is_none())
         } else {
             Err(PostHogEvaluationError::NotAvailable(format!(
-                "Not found in the local evaluation spec: {flag_key}"
+                "Not found in the local evaluation spec: {}",
+                flag_key
             )))
         }
     }
@@ -537,13 +534,6 @@ impl PostHogClient {
         })
     }
 
-    /// Check if the server API key is a feature flag secure API key. This key can only be
-    /// used to fetch the feature flag specs and can only be used on a undocumented API
-    /// endpoint.
-    fn is_feature_flag_secure_api_key(&self) -> bool {
-        self.config.server_api_key.starts_with("phs_")
-    }
-
     /// Fetch the feature flag specs from the server.
     ///
     /// This is unfortunately an undocumented API at:
@@ -557,22 +547,10 @@ impl PostHogClient {
     ) -> anyhow::Result<LocalEvaluationResponse> {
         // BASE_URL/api/projects/:project_id/feature_flags/local_evaluation
         // with bearer token of self.server_api_key
-        // OR
-        // BASE_URL/api/feature_flag/local_evaluation/
-        // with bearer token of feature flag specific self.server_api_key
-        let url = if self.is_feature_flag_secure_api_key() {
-            // The new feature local evaluation secure API token
-            format!(
-                "{}/api/feature_flag/local_evaluation",
-                self.config.private_api_url
-            )
-        } else {
-            // The old personal API token
-            format!(
-                "{}/api/projects/{}/feature_flags/local_evaluation",
-                self.config.private_api_url, self.config.project_id
-            )
-        };
+        let url = format!(
+            "{}/api/projects/{}/feature_flags/local_evaluation",
+            self.config.private_api_url, self.config.project_id
+        );
         let response = self
             .client
             .get(url)
@@ -825,7 +803,7 @@ mod tests {
     fn evaluate_multivariate() {
         let mut store = FeatureStore::new();
         let response: LocalEvaluationResponse = serde_json::from_str(data()).unwrap();
-        store.set_flags(response.flags, None).unwrap();
+        store.set_flags(response.flags);
 
         // This lacks the required properties and cannot be evaluated.
         let variant =
@@ -895,7 +873,7 @@ mod tests {
 
         let mut store = FeatureStore::new();
         let response: LocalEvaluationResponse = serde_json::from_str(data()).unwrap();
-        store.set_flags(response.flags, None).unwrap();
+        store.set_flags(response.flags);
 
         // This lacks the required properties and cannot be evaluated.
         let variant = store.evaluate_boolean_inner("boolean-flag", 1.00, &HashMap::new());
@@ -951,7 +929,7 @@ mod tests {
 
         let mut store = FeatureStore::new();
         let response: LocalEvaluationResponse = serde_json::from_str(data()).unwrap();
-        store.set_flags(response.flags, None).unwrap();
+        store.set_flags(response.flags);
 
         // This lacks the required properties and cannot be evaluated.
         let variant =

libs/pq_proto/src/lib.rs

@@ -198,7 +198,7 @@ impl fmt::Display for CancelKeyData {
 
         // This format is more compact and might work better for logs.
         f.debug_tuple("CancelKeyData")
-            .field(&format_args!("{id:x}"))
+            .field(&format_args!("{:x}", id))
             .finish()
     }
 }
@@ -291,7 +291,8 @@ impl FeMessage {
         let len = (&buf[1..5]).read_u32::<BigEndian>().unwrap();
         if len < 4 {
             return Err(ProtocolError::Protocol(format!(
-                "invalid message length {len}"
+                "invalid message length {}",
+                len
             )));
         }
 
@@ -366,7 +367,8 @@ impl FeStartupPacket {
         #[allow(clippy::manual_range_contains)]
         if len < 8 || len > MAX_STARTUP_PACKET_LENGTH {
             return Err(ProtocolError::Protocol(format!(
-                "invalid startup packet message length {len}"
+                "invalid startup packet message length {}",
+                len
             )));
         }
 

libs/proxy/postgres-protocol2/Cargo.toml

@@ -5,7 +5,7 @@ edition = "2024"
 license = "MIT/Apache-2.0"
 
 [dependencies]
-base64.workspace = true
+base64 = "0.20"
 byteorder.workspace = true
 bytes.workspace = true
 fallible-iterator.workspace = true

libs/proxy/postgres-protocol2/src/authentication/sasl.rs

@@ -3,8 +3,6 @@
 use std::fmt::Write;
 use std::{io, iter, mem, str};
 
-use base64::Engine as _;
-use base64::prelude::BASE64_STANDARD;
 use hmac::{Hmac, Mac};
 use rand::{self, Rng};
 use sha2::digest::FixedOutput;
@@ -228,7 +226,7 @@ impl ScramSha256 {
 
         let (client_key, server_key) = match password {
             Credentials::Password(password) => {
-                let salt = match BASE64_STANDARD.decode(parsed.salt) {
+                let salt = match base64::decode(parsed.salt) {
                     Ok(salt) => salt,
                     Err(e) => return Err(io::Error::new(io::ErrorKind::InvalidInput, e)),
                 };
@@ -257,7 +255,7 @@ impl ScramSha256 {
         let mut cbind_input = vec![];
         cbind_input.extend(channel_binding.gs2_header().as_bytes());
         cbind_input.extend(channel_binding.cbind_data());
-        let cbind_input = BASE64_STANDARD.encode(&cbind_input);
+        let cbind_input = base64::encode(&cbind_input);
 
         self.message.clear();
         write!(&mut self.message, "c={},r={}", cbind_input, parsed.nonce).unwrap();
@@ -274,12 +272,7 @@ impl ScramSha256 {
             *proof ^= signature;
         }
 
-        write!(
-            &mut self.message,
-            ",p={}",
-            BASE64_STANDARD.encode(client_proof)
-        )
-        .unwrap();
+        write!(&mut self.message, ",p={}", base64::encode(client_proof)).unwrap();
 
         self.state = State::Finish {
             server_key,
@@ -308,12 +301,12 @@ impl ScramSha256 {
 
         let verifier = match parsed {
             ServerFinalMessage::Error(e) => {
-                return Err(io::Error::other(format!("SCRAM error: {e}")));
+                return Err(io::Error::other(format!("SCRAM error: {}", e)));
             }
             ServerFinalMessage::Verifier(verifier) => verifier,
         };
 
-        let verifier = match BASE64_STANDARD.decode(verifier) {
+        let verifier = match base64::decode(verifier) {
             Ok(verifier) => verifier,
             Err(e) => return Err(io::Error::new(io::ErrorKind::InvalidInput, e)),
         };
@@ -343,8 +336,10 @@ impl<'a> Parser<'a> {
         match self.it.next() {
             Some((_, c)) if c == target => Ok(()),
             Some((i, c)) => {
-                let m =
-                    format!("unexpected character at byte {i}: expected `{target}` but got `{c}");
+                let m = format!(
+                    "unexpected character at byte {}: expected `{}` but got `{}",
+                    i, target, c
+                );
                 Err(io::Error::new(io::ErrorKind::InvalidInput, m))
             }
             None => Err(io::Error::new(
@@ -410,7 +405,7 @@ impl<'a> Parser<'a> {
         match self.it.peek() {
             Some(&(i, _)) => Err(io::Error::new(
                 io::ErrorKind::InvalidInput,
-                format!("unexpected trailing data at byte {i}"),
+                format!("unexpected trailing data at byte {}", i),
             )),
             None => Ok(()),
         }

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

@@ -211,7 +211,7 @@ impl Message {
                 tag => {
                     return Err(io::Error::new(
                         io::ErrorKind::InvalidInput,
-                        format!("unknown authentication tag `{tag}`"),
+                        format!("unknown authentication tag `{}`", tag),
                     ));
                 }
             },
@@ -238,7 +238,7 @@ impl Message {
             tag => {
                 return Err(io::Error::new(
                     io::ErrorKind::InvalidInput,
-                    format!("unknown message tag `{tag}`"),
+                    format!("unknown message tag `{}`", tag),
                 ));
             }
         };

libs/proxy/postgres-protocol2/src/password/mod.rs

@@ -6,8 +6,6 @@
 //! side. This is good because it ensures the cleartext password won't
 //! end up in logs pg_stat displays, etc.
 
-use base64::Engine as _;
-use base64::prelude::BASE64_STANDARD;
 use hmac::{Hmac, Mac};
 use rand::RngCore;
 use sha2::digest::FixedOutput;
@@ -85,8 +83,8 @@ pub(crate) async fn scram_sha_256_salt(
     format!(
         "SCRAM-SHA-256${}:{}${}:{}",
         SCRAM_DEFAULT_ITERATIONS,
-        BASE64_STANDARD.encode(salt),
-        BASE64_STANDARD.encode(stored_key),
-        BASE64_STANDARD.encode(server_key)
+        base64::encode(salt),
+        base64::encode(stored_key),
+        base64::encode(server_key)
     )
 }

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

@@ -46,7 +46,7 @@ impl fmt::Display for Type {
     fn fmt(&self, fmt: &mut fmt::Formatter<'_>) -> fmt::Result {
         match self.schema() {
             "public" | "pg_catalog" => {}
-            schema => write!(fmt, "{schema}.")?,
+            schema => write!(fmt, "{}.", schema)?,
         }
         fmt.write_str(self.name())
     }

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

@@ -1,13 +1,12 @@
 use tokio::net::TcpStream;
 
 use crate::client::SocketConfig;
-use crate::config::{Host, SslMode};
+use crate::config::Host;
 use crate::tls::MakeTlsConnect;
-use crate::{Error, cancel_query_raw, connect_socket};
+use crate::{Error, cancel_query_raw, connect_socket, connect_tls};
 
 pub(crate) async fn cancel_query<T>(
     config: SocketConfig,
-    ssl_mode: SslMode,
     tls: T,
     process_id: i32,
     secret_key: i32,
@@ -30,5 +29,6 @@ where
     )
     .await?;
 
-    cancel_query_raw::cancel_query_raw(socket, ssl_mode, tls, process_id, secret_key).await
+    let stream = connect_tls::connect_tls(socket, config.ssl_mode, tls).await?;
+    cancel_query_raw::cancel_query_raw(stream, process_id, secret_key).await
 }

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

@@ -2,23 +2,16 @@ use bytes::BytesMut;
 use postgres_protocol2::message::frontend;
 use tokio::io::{AsyncRead, AsyncWrite, AsyncWriteExt};
 
-use crate::config::SslMode;
-use crate::tls::TlsConnect;
-use crate::{Error, connect_tls};
+use crate::Error;
 
-pub async fn cancel_query_raw<S, T>(
-    stream: S,
-    mode: SslMode,
-    tls: T,
+pub async fn cancel_query_raw<S>(
+    mut stream: S,
     process_id: i32,
     secret_key: i32,
 ) -> Result<(), Error>
 where
     S: AsyncRead + AsyncWrite + Unpin,
-    T: TlsConnect<S>,
 {
-    let mut stream = connect_tls::connect_tls(stream, mode, tls).await?;
-
     let mut buf = BytesMut::new();
     frontend::cancel_request(process_id, secret_key, &mut buf);
 

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

@@ -3,21 +3,21 @@ use tokio::io::{AsyncRead, AsyncWrite};
 use tokio::net::TcpStream;
 
 use crate::client::SocketConfig;
-use crate::config::SslMode;
-use crate::tls::{MakeTlsConnect, TlsConnect};
+use crate::tls::MakeTlsConnect;
 use crate::{Error, cancel_query, cancel_query_raw};
 
-/// A cancellation token that allows easy cancellation of a query.
+/// The capability to request cancellation of in-progress queries on a
+/// connection.
 #[derive(Clone)]
 pub struct CancelToken {
     pub socket_config: SocketConfig,
     pub raw: RawCancelToken,
 }
 
-/// A raw cancellation token that allows cancellation of a query, given a fresh connection to postgres.
+/// The capability to request cancellation of in-progress queries on a
+/// connection.
 #[derive(Debug, Clone, Serialize, Deserialize)]
 pub struct RawCancelToken {
-    pub ssl_mode: SslMode,
     pub process_id: i32,
     pub secret_key: i32,
 }
@@ -41,7 +41,6 @@ impl CancelToken {
     {
         cancel_query::cancel_query(
             self.socket_config.clone(),
-            self.raw.ssl_mode,
             tls,
             self.raw.process_id,
             self.raw.secret_key,
@@ -53,18 +52,10 @@ impl CancelToken {
 impl RawCancelToken {
     /// Like `cancel_query`, but uses a stream which is already connected to the server rather than opening a new
     /// connection itself.
-    pub async fn cancel_query_raw<S, T>(&self, stream: S, tls: T) -> Result<(), Error>
+    pub async fn cancel_query_raw<S>(&self, stream: S) -> Result<(), Error>
     where
         S: AsyncRead + AsyncWrite + Unpin,
-        T: TlsConnect<S>,
     {
-        cancel_query_raw::cancel_query_raw(
-            stream,
-            self.ssl_mode,
-            tls,
-            self.process_id,
-            self.secret_key,
-        )
-        .await
+        cancel_query_raw::cancel_query_raw(stream, self.process_id, self.secret_key).await
     }
 }

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

@@ -167,6 +167,7 @@ pub struct SocketConfig {
     pub host: Host,
     pub port: u16,
     pub connect_timeout: Option<Duration>,
+    pub ssl_mode: SslMode,
 }
 
 /// An asynchronous PostgreSQL client.
@@ -178,7 +179,6 @@ pub struct Client {
     cached_typeinfo: CachedTypeInfo,
 
     socket_config: SocketConfig,
-    ssl_mode: SslMode,
     process_id: i32,
     secret_key: i32,
 }
@@ -188,7 +188,6 @@ impl Client {
         sender: mpsc::UnboundedSender<FrontendMessage>,
         receiver: mpsc::Receiver<BackendMessages>,
         socket_config: SocketConfig,
-        ssl_mode: SslMode,
         process_id: i32,
         secret_key: i32,
     ) -> Client {
@@ -206,7 +205,6 @@ impl Client {
             cached_typeinfo: Default::default(),
 
             socket_config,
-            ssl_mode,
             process_id,
             secret_key,
         }
@@ -334,7 +332,6 @@ impl Client {
         CancelToken {
             socket_config: self.socket_config.clone(),
             raw: RawCancelToken {
-                ssl_mode: self.ssl_mode,
                 process_id: self.process_id,
                 secret_key: self.secret_key,
             },

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

@@ -57,6 +57,7 @@ where
         host: host.clone(),
         port,
         connect_timeout: config.connect_timeout,
+        ssl_mode: config.ssl_mode,
     };
 
     let (client_tx, conn_rx) = mpsc::unbounded_channel();
@@ -65,7 +66,6 @@ where
         client_tx,
         client_rx,
         socket_config,
-        config.ssl_mode,
         process_id,
         secret_key,
     );