WIP: Add pg_tracing extension

.dockerignore

@@ -19,7 +19,6 @@
 !pageserver/
 !pgxn/
 !proxy/
-!object_storage/
 !storage_scrubber/
 !safekeeper/
 !storage_broker/

.github/actionlint.yml

@@ -6,7 +6,6 @@ self-hosted-runner:
     - small
     - small-metal
     - small-arm64
-    - unit-perf
     - us-east-2
 config-variables:
   - AWS_ECR_REGION

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

@@ -70,7 +70,6 @@ runs:
 
     - name: Install Allure
       shell: bash -euxo pipefail {0}
-      working-directory: /tmp
       run: |
         if ! which allure; then
           ALLURE_ZIP=allure-${ALLURE_VERSION}.zip

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

@@ -113,6 +113,8 @@ runs:
         TEST_OUTPUT: /tmp/test_output
         BUILD_TYPE: ${{ inputs.build_type }}
         COMPATIBILITY_SNAPSHOT_DIR: /tmp/compatibility_snapshot_pg${{ inputs.pg_version }}
+        ALLOW_BACKWARD_COMPATIBILITY_BREAKAGE: contains(github.event.pull_request.labels.*.name, 'backward compatibility breakage')
+        ALLOW_FORWARD_COMPATIBILITY_BREAKAGE: contains(github.event.pull_request.labels.*.name, 'forward compatibility breakage')
         RERUN_FAILED: ${{ inputs.rerun_failed }}
         PG_VERSION: ${{ inputs.pg_version }}
         SANITIZERS: ${{ inputs.sanitizers }}

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

@@ -272,13 +272,10 @@ jobs:
           # run pageserver tests with different settings
           for get_vectored_concurrent_io in sequential sidecar-task; do
             for io_engine in std-fs tokio-epoll-uring ; do
-                for io_mode in buffered direct direct-rw ; do
-                  NEON_PAGESERVER_UNIT_TEST_GET_VECTORED_CONCURRENT_IO=$get_vectored_concurrent_io \
-                  NEON_PAGESERVER_UNIT_TEST_VIRTUAL_FILE_IOENGINE=$io_engine \
-                  NEON_PAGESERVER_UNIT_TEST_VIRTUAL_FILE_IOMODE=$io_mode \
-                  ${cov_prefix} \
-                  cargo nextest run $CARGO_FLAGS $CARGO_FEATURES  -E 'package(pageserver)'
-              done
+              NEON_PAGESERVER_UNIT_TEST_GET_VECTORED_CONCURRENT_IO=$get_vectored_concurrent_io \
+                NEON_PAGESERVER_UNIT_TEST_VIRTUAL_FILE_IOENGINE=$io_engine \
+                ${cov_prefix} \
+                cargo nextest run $CARGO_FLAGS $CARGO_FEATURES  -E 'package(pageserver)'
             done
           done
 
@@ -349,7 +346,7 @@ jobs:
       contents: read
       statuses: write
     needs: [ build-neon ]
-    runs-on: ${{ fromJSON(format('["self-hosted", "{0}"]', inputs.arch == 'arm64' && 'large-arm64' || 'large-metal')) }}
+    runs-on: ${{ fromJSON(format('["self-hosted", "{0}"]', inputs.arch == 'arm64' && 'large-arm64' || 'large')) }}
     container:
       image: ${{ inputs.build-tools-image }}
       credentials:
@@ -395,7 +392,6 @@ jobs:
           BUILD_TAG: ${{ inputs.build-tag }}
           PAGESERVER_VIRTUAL_FILE_IO_ENGINE: tokio-epoll-uring
           PAGESERVER_GET_VECTORED_CONCURRENT_IO: sidecar-task
-          PAGESERVER_VIRTUAL_FILE_IO_MODE: direct
           USE_LFC: ${{ matrix.lfc_state == 'with-lfc' && 'true' || 'false' }}
 
       # Temporary disable this step until we figure out why it's so flaky

.github/workflows/_create-release-pr.yml

@@ -53,13 +53,10 @@ jobs:
             || inputs.component-name == 'Compute' && 'release-compute'
           }}
       run: |
-        now_date=$(date -u +'%Y-%m-%d')
-        now_time=$(date -u +'%H-%M-%Z')
-        {
-          echo "title=${COMPONENT_NAME} release ${now_date}"
-          echo "rc-branch=rc/${RELEASE_BRANCH}/${now_date}_${now_time}"
-          echo "release-branch=${RELEASE_BRANCH}"
-        } | tee -a ${GITHUB_OUTPUT}
+        today=$(date +'%Y-%m-%d')
+        echo "title=${COMPONENT_NAME} release ${today}" | tee -a ${GITHUB_OUTPUT}
+        echo "rc-branch=rc/${RELEASE_BRANCH}/${today}"  | tee -a ${GITHUB_OUTPUT}
+        echo "release-branch=${RELEASE_BRANCH}"         | tee -a ${GITHUB_OUTPUT}
 
     - name: Configure git
       run: |

.github/workflows/_meta.yml

@@ -165,5 +165,5 @@ jobs:
           GH_TOKEN: ${{ secrets.GITHUB_TOKEN }}
           CURRENT_SHA: ${{ github.sha }}
         run: |
-          RELEASE_PR_RUN_ID=$(gh api "/repos/${GITHUB_REPOSITORY}/actions/runs?head_sha=$CURRENT_SHA" | jq '[.workflow_runs[] | select(.name == "Build and Test") | select(.head_branch | test("^rc/release.*$"; "s"))] | first | .id // ("Failed to find Build and Test run from  RC PR!" | halt_error(1))')
+          RELEASE_PR_RUN_ID=$(gh api "/repos/${GITHUB_REPOSITORY}/actions/runs?head_sha=$CURRENT_SHA" | jq '[.workflow_runs[] | select(.name == "Build and Test") | select(.head_branch | test("^rc/release(-(proxy|compute))?/[0-9]{4}-[0-9]{2}-[0-9]{2}$"; "s"))] | first | .id // ("Failed to find Build and Test run from  RC PR!" | halt_error(1))')
           echo "release-pr-run-id=$RELEASE_PR_RUN_ID" | tee -a $GITHUB_OUTPUT

.github/workflows/build_and_test.yml

@@ -284,7 +284,7 @@ jobs:
       statuses: write
       contents: write
       pull-requests: write
-    runs-on: [ self-hosted, unit-perf ]
+    runs-on: [ self-hosted, small-metal ]
     container:
       image: ${{ needs.build-build-tools-image.outputs.image }}-bookworm
       credentials:
@@ -323,8 +323,6 @@ jobs:
           PERF_TEST_RESULT_CONNSTR: "${{ secrets.PERF_TEST_RESULT_CONNSTR }}"
           TEST_RESULT_CONNSTR: "${{ secrets.REGRESS_TEST_RESULT_CONNSTR_NEW }}"
           PAGESERVER_VIRTUAL_FILE_IO_ENGINE: tokio-epoll-uring
-          PAGESERVER_GET_VECTORED_CONCURRENT_IO: sidecar-task
-          PAGESERVER_VIRTUAL_FILE_IO_MODE: direct
           SYNC_BETWEEN_TESTS: true
       # XXX: no coverage data handling here, since benchmarks are run on release builds,
       # while coverage is currently collected for the debug ones
@@ -1273,7 +1271,7 @@ jobs:
           exit 1
 
   deploy:
-    needs: [ check-permissions, push-neon-image-dev, push-compute-image-dev, push-neon-image-prod, push-compute-image-prod, meta, trigger-custom-extensions-build-and-wait ]
+    needs: [ check-permissions, push-neon-image-dev, push-compute-image-dev, push-neon-image-prod, push-compute-image-prod, meta, build-and-test-locally, trigger-custom-extensions-build-and-wait ]
     # `!failure() && !cancelled()` is required because the workflow depends on the job that can be skipped: `push-neon-image-prod` and `push-compute-image-prod`
     if: ${{ contains(fromJSON('["push-main", "storage-release", "proxy-release", "compute-release"]'), needs.meta.outputs.run-kind) && !failure() && !cancelled() }}
     permissions:

.github/workflows/fast-forward.yml

@@ -27,17 +27,15 @@ jobs:
       - name: Fast forwarding
         uses: sequoia-pgp/fast-forward@ea7628bedcb0b0b96e94383ada458d812fca4979
         # See https://docs.github.com/en/graphql/reference/enums#mergestatestatus
-        if: ${{ contains(fromJSON('["clean", "unstable"]'), github.event.pull_request.mergeable_state) }}
+        if: ${{ github.event.pull_request.mergeable_state  == 'clean' }}
         with:
           merge: true
           comment: on-error
           github_token: ${{ secrets.CI_ACCESS_TOKEN }}
 
       - name: Comment if mergeable_state is not clean
-        if: ${{ !contains(fromJSON('["clean", "unstable"]'), github.event.pull_request.mergeable_state) }}
-        env:
-          GH_TOKEN: ${{ secrets.CI_ACCESS_TOKEN }}
+        if: ${{ github.event.pull_request.mergeable_state  != 'clean' }}
         run: |
           gh pr comment ${{ github.event.pull_request.number }} \
             --repo "${GITHUB_REPOSITORY}" \
-            --body "Not trying to forward pull-request, because \`mergeable_state\` is \`${{ github.event.pull_request.mergeable_state }}\`, not \`clean\` or \`unstable\`."
+            --body "Not trying to forward pull-request, because \`mergeable_state\` is \`${{ github.event.pull_request.mergeable_state }}\`, not \`clean\`."

.github/workflows/pg-clients.yml

@@ -30,7 +30,7 @@ permissions:
   statuses: write # require for posting a status update
 
 env:
-  DEFAULT_PG_VERSION: 17
+  DEFAULT_PG_VERSION: 16
   PLATFORM: neon-captest-new
   AWS_DEFAULT_REGION: eu-central-1
 
@@ -42,8 +42,6 @@ jobs:
       github-event-name: ${{ github.event_name }}
 
   build-build-tools-image:
-    permissions:
-      packages: write
     needs: [ check-permissions ]
     uses: ./.github/workflows/build-build-tools-image.yml
     secrets: inherit

.github/workflows/random-ops-test.yml

@@ -1,93 +0,0 @@
-name: Random Operations Test
-
-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:  '23 */2 * * *' # runs every 2 hours
-  workflow_dispatch:
-    inputs:
-      random_seed:
-        type: number
-        description: 'The random seed'
-        required: false
-        default: 0
-      num_operations:
-        type: number
-        description: "The number of operations to test"
-        default: 250
-
-defaults:
-  run:
-    shell: bash -euxo pipefail {0}
-
-permissions: {}
-
-env:
-  DEFAULT_PG_VERSION: 16
-  PLATFORM: neon-captest-new
-  AWS_DEFAULT_REGION: eu-central-1
-
-jobs:
-  run-random-rests:
-    env:
-      POSTGRES_DISTRIB_DIR: /tmp/neon/pg_install
-    runs-on: small
-    permissions:
-      id-token: write
-      statuses: write
-
-    strategy:
-      fail-fast: false
-      matrix:
-        pg-version: [16, 17]
-
-    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: Download Neon artifact
-        uses: ./.github/actions/download
-        with:
-          name: neon-${{ runner.os }}-${{ runner.arch }}-release-artifact
-          path: /tmp/neon/
-          prefix: latest
-          aws-oicd-role-arn: ${{ vars.DEV_AWS_OIDC_ROLE_ARN }}
-
-      - name: Run tests
-        uses: ./.github/actions/run-python-test-set
-        with:
-          build_type: remote
-          test_selection: random_ops
-          run_in_parallel: false
-          extra_params: -m remote_cluster
-          pg_version: ${{ matrix.pg-version }}
-          aws-oicd-role-arn: ${{ vars.DEV_AWS_OIDC_ROLE_ARN }}
-        env:
-          NEON_API_KEY: ${{ secrets.NEON_STAGING_API_KEY }}
-          RANDOM_SEED: ${{ inputs.random_seed }}
-          NUM_OPERATIONS: ${{ inputs.num_operations }}
-
-      - name: Create Allure report
-        if: ${{ !cancelled() }}
-        id: create-allure-report
-        uses: ./.github/actions/allure-report-generate
-        with:
-          store-test-results-into-db: true
-          aws-oicd-role-arn: ${{ vars.DEV_AWS_OIDC_ROLE_ARN }}
-        env:
-          REGRESS_TEST_RESULT_CONNSTR_NEW: ${{ secrets.REGRESS_TEST_RESULT_CONNSTR_NEW }}

.gitignore

@@ -1,4 +1,3 @@
-/artifact_cache
 /pg_install
 /target
 /tmp_check

Cargo.lock

@@ -1416,7 +1416,6 @@ name = "control_plane"
 version = "0.1.0"
 dependencies = [
  "anyhow",
- "base64 0.13.1",
  "camino",
  "clap",
  "comfy-table",
@@ -1426,12 +1425,10 @@ dependencies = [
  "humantime",
  "humantime-serde",
  "hyper 0.14.30",
- "jsonwebtoken",
  "nix 0.27.1",
  "once_cell",
  "pageserver_api",
  "pageserver_client",
- "pem",
  "postgres_backend",
  "postgres_connection",
  "regex",
@@ -1440,8 +1437,6 @@ dependencies = [
  "scopeguard",
  "serde",
  "serde_json",
- "sha2",
- "spki 0.7.3",
  "storage_broker",
  "thiserror 1.0.69",
  "tokio",
@@ -2822,7 +2817,6 @@ dependencies = [
  "hyper 0.14.30",
  "itertools 0.10.5",
  "jemalloc_pprof",
- "jsonwebtoken",
  "metrics",
  "once_cell",
  "pprof",
@@ -2843,7 +2837,6 @@ dependencies = [
  "utils",
  "uuid",
  "workspace_hack",
- "x509-cert",
 ]
 
 [[package]]
@@ -3998,33 +3991,6 @@ dependencies = [
  "memchr",
 ]
 
-[[package]]
-name = "object_storage"
-version = "0.0.1"
-dependencies = [
- "anyhow",
- "axum",
- "axum-extra",
- "camino",
- "camino-tempfile",
- "futures",
- "http-body-util",
- "itertools 0.10.5",
- "jsonwebtoken",
- "prometheus",
- "rand 0.8.5",
- "remote_storage",
- "serde",
- "serde_json",
- "test-log",
- "tokio",
- "tokio-util",
- "tower 0.5.2",
- "tracing",
- "utils",
- "workspace_hack",
-]
-
 [[package]]
 name = "once_cell"
 version = "1.20.2"
@@ -4275,7 +4241,6 @@ dependencies = [
  "hyper 0.14.30",
  "indoc",
  "itertools 0.10.5",
- "jsonwebtoken",
  "md5",
  "metrics",
  "nix 0.27.1",
@@ -4285,7 +4250,6 @@ dependencies = [
  "pageserver_api",
  "pageserver_client",
  "pageserver_compaction",
- "pem",
  "pin-project-lite",
  "postgres-protocol",
  "postgres-types",
@@ -4353,7 +4317,6 @@ dependencies = [
  "humantime-serde",
  "itertools 0.10.5",
  "nix 0.27.1",
- "once_cell",
  "postgres_backend",
  "postgres_ffi",
  "rand 0.8.5",
@@ -4730,7 +4693,7 @@ dependencies = [
 [[package]]
 name = "postgres-protocol"
 version = "0.6.6"
-source = "git+https://github.com/neondatabase/rust-postgres.git?branch=neon#f3cf448febde5fd298071d54d568a9c875a7a62b"
+source = "git+https://github.com/neondatabase/rust-postgres.git?branch=neon#1f21e7959a96a34dcfbfce1b14b73286cdadffe9"
 dependencies = [
  "base64 0.22.1",
  "byteorder",
@@ -4764,7 +4727,7 @@ dependencies = [
 [[package]]
 name = "postgres-types"
 version = "0.2.6"
-source = "git+https://github.com/neondatabase/rust-postgres.git?branch=neon#f3cf448febde5fd298071d54d568a9c875a7a62b"
+source = "git+https://github.com/neondatabase/rust-postgres.git?branch=neon#1f21e7959a96a34dcfbfce1b14b73286cdadffe9"
 dependencies = [
  "bytes",
  "chrono",
@@ -5694,9 +5657,9 @@ dependencies = [
 
 [[package]]
 name = "ring"
-version = "0.17.14"
+version = "0.17.13"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "a4689e6c2294d81e88dc6261c768b63bc4fcdb852be6d1352498b114f61383b7"
+checksum = "70ac5d832aa16abd7d1def883a8545280c20a60f523a370aa3a9617c2b8550ee"
 dependencies = [
  "cc",
  "cfg-if",
@@ -5997,12 +5960,10 @@ dependencies = [
  "humantime",
  "hyper 0.14.30",
  "itertools 0.10.5",
- "jsonwebtoken",
  "metrics",
  "once_cell",
  "pageserver_api",
  "parking_lot 0.12.1",
- "pem",
  "postgres-protocol",
  "postgres_backend",
  "postgres_ffi",
@@ -6964,28 +6925,6 @@ dependencies = [
  "syn 2.0.100",
 ]
 
-[[package]]
-name = "test-log"
-version = "0.2.17"
-source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "e7f46083d221181166e5b6f6b1e5f1d499f3a76888826e6cb1d057554157cd0f"
-dependencies = [
- "env_logger",
- "test-log-macros",
- "tracing-subscriber",
-]
-
-[[package]]
-name = "test-log-macros"
-version = "0.2.17"
-source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "888d0c3c6db53c0fdab160d2ed5e12ba745383d3e85813f2ea0f2b1475ab553f"
-dependencies = [
- "proc-macro2",
- "quote",
- "syn 2.0.100",
-]
-
 [[package]]
 name = "thiserror"
 version = "1.0.69"
@@ -7233,7 +7172,7 @@ dependencies = [
 [[package]]
 name = "tokio-postgres"
 version = "0.7.10"
-source = "git+https://github.com/neondatabase/rust-postgres.git?branch=neon#f3cf448febde5fd298071d54d568a9c875a7a62b"
+source = "git+https://github.com/neondatabase/rust-postgres.git?branch=neon#1f21e7959a96a34dcfbfce1b14b73286cdadffe9"
 dependencies = [
  "async-trait",
  "byteorder",
@@ -7883,7 +7822,6 @@ dependencies = [
  "metrics",
  "nix 0.27.1",
  "once_cell",
- "pem",
  "pin-project-lite",
  "postgres_connection",
  "pprof",

Cargo.toml

@@ -40,7 +40,6 @@ members = [
     "libs/proxy/postgres-protocol2",
     "libs/proxy/postgres-types2",
     "libs/proxy/tokio-postgres2",
-    "object_storage",
 ]
 
 [workspace.package]
@@ -141,7 +140,6 @@ parking_lot = "0.12"
 parquet = { version = "53", default-features = false, features = ["zstd"] }
 parquet_derive = "53"
 pbkdf2 = { version = "0.12.1", features = ["simple", "std"] }
-pem = "3.0.3"
 pin-project-lite = "0.2"
 pprof = { version = "0.14", features = ["criterion", "flamegraph", "frame-pointer", "prost-codec"] }
 procfs = "0.16"
@@ -175,7 +173,6 @@ signal-hook = "0.3"
 smallvec = "1.11"
 smol_str = { version = "0.2.0", features = ["serde"] }
 socket2 = "0.5"
-spki = "0.7.3"
 strum = "0.26"
 strum_macros = "0.26"
 "subtle"  = "2.5.0"
@@ -211,7 +208,6 @@ tracing-opentelemetry = "0.28"
 tracing-serde = "0.2.0"
 tracing-subscriber = { version = "0.3", default-features = false, features = ["smallvec", "fmt", "tracing-log", "std", "env-filter", "json"] }
 try-lock = "0.2.5"
-test-log = { version = "0.2.17", default-features = false, features = ["log"] }
 twox-hash = { version = "1.6.3", default-features = false }
 typed-json = "0.1"
 url = "2.2"

Dockerfile

@@ -89,7 +89,6 @@ RUN set -e \
       --bin storage_broker  \
       --bin storage_controller  \
       --bin proxy  \
-      --bin object_storage \
       --bin neon_local \
       --bin storage_scrubber \
       --locked --release
@@ -122,7 +121,6 @@ COPY --from=build --chown=neon:neon /home/nonroot/target/release/safekeeper
 COPY --from=build --chown=neon:neon /home/nonroot/target/release/storage_broker      /usr/local/bin
 COPY --from=build --chown=neon:neon /home/nonroot/target/release/storage_controller  /usr/local/bin
 COPY --from=build --chown=neon:neon /home/nonroot/target/release/proxy               /usr/local/bin
-COPY --from=build --chown=neon:neon /home/nonroot/target/release/object_storage      /usr/local/bin
 COPY --from=build --chown=neon:neon /home/nonroot/target/release/neon_local          /usr/local/bin
 COPY --from=build --chown=neon:neon /home/nonroot/target/release/storage_scrubber    /usr/local/bin
 

README.md

@@ -270,7 +270,7 @@ By default, this runs both debug and release modes, and all supported postgres v
 testing locally, it is convenient to run just one set of permutations, like this:
 
 ```sh
-DEFAULT_PG_VERSION=17 BUILD_TYPE=release ./scripts/pytest
+DEFAULT_PG_VERSION=16 BUILD_TYPE=release ./scripts/pytest
 ```
 
 ## Flamegraphs

clippy.toml

@@ -12,5 +12,3 @@ disallowed-macros = [
     # cannot disallow this, because clippy finds used from tokio macros
     #"tokio::pin",
 ]
-
-allow-unwrap-in-tests = true

compute/compute-node.Dockerfile

@@ -1388,6 +1388,38 @@ RUN make -j $(getconf _NPROCESSORS_ONLN) && \
     make -j $(getconf _NPROCESSORS_ONLN) install && \
     echo 'trusted = true' >> /usr/local/pgsql/share/extension/pg_partman.control
 
+#########################################################################################
+# Layer "pg_tracing"
+# compile pg_tracing extension
+#
+#########################################################################################
+FROM build-deps AS pg_tracing-src
+ARG PG_VERSION
+WORKDIR /ext-src
+RUN case "${PG_VERSION:?}" in \
+      "v14" | "v15") \
+        echo "pg_tracing not supported on this PostgreSQL version." && exit 0 \
+	;; \
+      *) \
+      ;; \
+    esac && \
+    wget https://github.com/DataDog/pg_tracing/archive/refs/tags/v0.1.3.tar.gz -O pg_tracing.tar.gz && \
+    echo "d0a7cca7279bb29601ba6c4c1aaeb3a44d71e6afa3b78aae1e3b7269e688f907 pg_tracing.tar.gz" | sha256sum --check && \
+    mkdir pg_tracing-src && cd pg_tracing-src && tar xzf ../pg_tracing.tar.gz --strip-components=1 -C .
+
+FROM pg-build AS pg_tracing-build
+COPY --from=pg_tracing-src /ext-src/ /ext-src/
+WORKDIR /ext-src/pg_tracing-src
+RUN case "${PG_VERSION:?}" in \
+      "v14" | "v15") \
+        echo "pg_tracing not supported on this PostgreSQL version." && exit 0 \
+        ;; \
+      *) \
+      ;; \
+    esac && \
+    make -j $(getconf _NPROCESSORS_ONLN) && \
+    make -j $(getconf _NPROCESSORS_ONLN) install
+
 #########################################################################################
 #
 # Layer "pg_mooncake"
@@ -1617,6 +1649,7 @@ COPY --from=pg_semver-build /usr/local/pgsql/ /usr/local/pgsql/
 COPY --from=wal2json-build /usr/local/pgsql /usr/local/pgsql
 COPY --from=pg_ivm-build /usr/local/pgsql/ /usr/local/pgsql/
 COPY --from=pg_partman-build /usr/local/pgsql/ /usr/local/pgsql/
+COPY --from=pg_tracing-build /usr/local/pgsql/ /usr/local/pgsql/
 COPY --from=pg_mooncake-build /usr/local/pgsql/ /usr/local/pgsql/
 COPY --from=pg_duckdb-build /usr/local/pgsql/ /usr/local/pgsql/
 COPY --from=pg_repack-build /usr/local/pgsql/ /usr/local/pgsql/
@@ -1793,6 +1826,7 @@ COPY --from=pg_semver-src /ext-src/ /ext-src/
 #COPY --from=wal2json-src /ext-src/ /ext-src/
 COPY --from=pg_ivm-src /ext-src/ /ext-src/
 COPY --from=pg_partman-src /ext-src/ /ext-src/
+COPY --from=pg_tracing-src /ext-src/ /ext-src/
 #COPY --from=pg_mooncake-src /ext-src/ /ext-src/
 COPY --from=pg_repack-src /ext-src/ /ext-src/
 COPY --from=pg_repack-build /usr/local/pgsql/ /usr/local/pgsql/

compute/patches/pg_anon.patch

@@ -0,0 +1,265 @@
+commit 00aa659afc9c7336ab81036edec3017168aabf40
+Author: Heikki Linnakangas <heikki@neon.tech>
+Date:   Tue Nov 12 16:59:19 2024 +0200
+
+    Temporarily disable test that depends on timezone
+
+diff --git a/tests/expected/generalization.out b/tests/expected/generalization.out
+index 23ef5fa..9e60deb 100644
+--- a/ext-src/pg_anon-src/tests/expected/generalization.out
++++ b/ext-src/pg_anon-src/tests/expected/generalization.out
+@@ -284,12 +284,9 @@ SELECT anon.generalize_tstzrange('19041107','century');
+  ["Tue Jan 01 00:00:00 1901 PST","Mon Jan 01 00:00:00 2001 PST")
+ (1 row)
+ 
+-SELECT anon.generalize_tstzrange('19041107','millennium');
+-                      generalize_tstzrange                       
+------------------------------------------------------------------
+- ["Thu Jan 01 00:00:00 1001 PST","Mon Jan 01 00:00:00 2001 PST")
+-(1 row)
+-
++-- temporarily disabled, see:
++-- https://gitlab.com/dalibo/postgresql_anonymizer/-/commit/199f0a392b37c59d92ae441fb8f037e094a11a52#note_2148017485
++--SELECT anon.generalize_tstzrange('19041107','millennium');
+ -- generalize_daterange
+ SELECT anon.generalize_daterange('19041107');
+   generalize_daterange   
+diff --git a/tests/sql/generalization.sql b/tests/sql/generalization.sql
+index b868344..b4fc977 100644
+--- a/ext-src/pg_anon-src/tests/sql/generalization.sql
++++ b/ext-src/pg_anon-src/tests/sql/generalization.sql
+@@ -61,7 +61,9 @@ SELECT anon.generalize_tstzrange('19041107','month');
+ SELECT anon.generalize_tstzrange('19041107','year');
+ SELECT anon.generalize_tstzrange('19041107','decade');
+ SELECT anon.generalize_tstzrange('19041107','century');
+-SELECT anon.generalize_tstzrange('19041107','millennium');
++-- temporarily disabled, see:
++-- https://gitlab.com/dalibo/postgresql_anonymizer/-/commit/199f0a392b37c59d92ae441fb8f037e094a11a52#note_2148017485
++--SELECT anon.generalize_tstzrange('19041107','millennium');
+ 
+ -- generalize_daterange
+ SELECT anon.generalize_daterange('19041107');
+
+commit 7dd414ee75f2875cffb1d6ba474df1f135a6fc6f
+Author: Alexey Masterov <alexeymasterov@neon.tech>
+Date:   Fri May 31 06:34:26 2024 +0000
+
+    These alternative expected files were added to consider the neon features
+
+diff --git a/ext-src/pg_anon-src/tests/expected/permissions_masked_role_1.out b/ext-src/pg_anon-src/tests/expected/permissions_masked_role_1.out
+new file mode 100644
+index 0000000..2539cfd
+--- /dev/null
++++ b/ext-src/pg_anon-src/tests/expected/permissions_masked_role_1.out
+@@ -0,0 +1,101 @@
++BEGIN;
++CREATE EXTENSION anon CASCADE;
++NOTICE:  installing required extension "pgcrypto"
++SELECT anon.init();
++ init 
++------
++ t
++(1 row)
++
++CREATE ROLE mallory_the_masked_user;
++SECURITY LABEL FOR anon ON ROLE mallory_the_masked_user IS 'MASKED';
++CREATE TABLE t1(i INT);
++ALTER TABLE t1 ADD COLUMN t TEXT;
++SECURITY LABEL FOR anon ON COLUMN t1.t
++IS 'MASKED WITH VALUE NULL';
++INSERT INTO t1 VALUES (1,'test');
++--
++-- We're checking the owner's permissions
++--
++-- see
++-- https://postgresql-anonymizer.readthedocs.io/en/latest/SECURITY/#permissions
++--
++SET ROLE mallory_the_masked_user;
++SELECT anon.pseudo_first_name(0) IS NOT NULL;
++ ?column? 
++----------
++ t
++(1 row)
++
++-- SHOULD FAIL
++DO $$
++BEGIN
++  PERFORM anon.init();
++  EXCEPTION WHEN insufficient_privilege
++  THEN RAISE NOTICE 'insufficient_privilege';
++END$$;
++NOTICE:  insufficient_privilege
++-- SHOULD FAIL
++DO $$
++BEGIN
++  PERFORM anon.anonymize_table('t1');
++  EXCEPTION WHEN insufficient_privilege
++  THEN RAISE NOTICE 'insufficient_privilege';
++END$$;
++NOTICE:  insufficient_privilege
++-- SHOULD FAIL
++SAVEPOINT fail_start_engine;
++SELECT anon.start_dynamic_masking();
++ERROR:  Only supersusers can start the dynamic masking engine.
++CONTEXT:  PL/pgSQL function anon.start_dynamic_masking(boolean) line 18 at RAISE
++ROLLBACK TO fail_start_engine;
++RESET ROLE;
++SELECT anon.start_dynamic_masking();
++ start_dynamic_masking 
++-----------------------
++ t
++(1 row)
++
++SET ROLE mallory_the_masked_user;
++SELECT * FROM mask.t1;
++ i | t 
++---+---
++ 1 | 
++(1 row)
++
++-- SHOULD FAIL
++DO $$
++BEGIN
++  SELECT * FROM public.t1;
++  EXCEPTION WHEN insufficient_privilege
++  THEN RAISE NOTICE 'insufficient_privilege';
++END$$;
++NOTICE:  insufficient_privilege
++-- SHOULD FAIL
++SAVEPOINT fail_stop_engine;
++SELECT anon.stop_dynamic_masking();
++ERROR:  Only supersusers can stop the dynamic masking engine.
++CONTEXT:  PL/pgSQL function anon.stop_dynamic_masking() line 18 at RAISE
++ROLLBACK TO fail_stop_engine;
++RESET ROLE;
++SELECT anon.stop_dynamic_masking();
++NOTICE:  The previous priviledges of 'mallory_the_masked_user' are not restored. You need to grant them manually.
++ stop_dynamic_masking 
++----------------------
++ t
++(1 row)
++
++SET ROLE mallory_the_masked_user;
++SELECT COUNT(*)=1 FROM anon.pg_masking_rules;
++ ?column? 
++----------
++ t
++(1 row)
++
++-- SHOULD FAIL
++SAVEPOINT fail_seclabel_on_role;
++SECURITY LABEL FOR anon ON ROLE mallory_the_masked_user IS NULL;
++ERROR:  permission denied
++DETAIL:  The current user must have the CREATEROLE attribute.
++ROLLBACK TO fail_seclabel_on_role;
++ROLLBACK;
+diff --git a/ext-src/pg_anon-src/tests/expected/permissions_owner_1.out b/ext-src/pg_anon-src/tests/expected/permissions_owner_1.out
+new file mode 100644
+index 0000000..8b090fe
+--- /dev/null
++++ b/ext-src/pg_anon-src/tests/expected/permissions_owner_1.out
+@@ -0,0 +1,104 @@
++BEGIN;
++CREATE EXTENSION anon CASCADE;
++NOTICE:  installing required extension "pgcrypto"
++SELECT anon.init();
++ init 
++------
++ t
++(1 row)
++
++CREATE ROLE oscar_the_owner;
++ALTER DATABASE :DBNAME OWNER TO oscar_the_owner;
++CREATE ROLE mallory_the_masked_user;
++SECURITY LABEL FOR anon ON ROLE mallory_the_masked_user IS 'MASKED';
++--
++-- We're checking the owner's permissions
++--
++-- see
++-- https://postgresql-anonymizer.readthedocs.io/en/latest/SECURITY/#permissions
++--
++SET ROLE oscar_the_owner;
++SELECT anon.pseudo_first_name(0) IS NOT NULL;
++ ?column? 
++----------
++ t
++(1 row)
++
++-- SHOULD FAIL
++DO $$
++BEGIN
++  PERFORM anon.init();
++  EXCEPTION WHEN insufficient_privilege
++  THEN RAISE NOTICE 'insufficient_privilege';
++END$$;
++NOTICE:  insufficient_privilege
++CREATE TABLE t1(i INT);
++ALTER TABLE t1 ADD COLUMN t TEXT;
++SECURITY LABEL FOR anon ON COLUMN t1.t
++IS 'MASKED WITH VALUE NULL';
++INSERT INTO t1 VALUES (1,'test');
++SELECT anon.anonymize_table('t1');
++ anonymize_table 
++-----------------
++ t
++(1 row)
++
++SELECT * FROM t1;
++ i | t 
++---+---
++ 1 | 
++(1 row)
++
++UPDATE t1 SET t='test' WHERE i=1;
++-- SHOULD FAIL
++SAVEPOINT fail_start_engine;
++SELECT anon.start_dynamic_masking();
++ start_dynamic_masking 
++-----------------------
++ t
++(1 row)
++
++ROLLBACK TO fail_start_engine;
++RESET ROLE;
++SELECT anon.start_dynamic_masking();
++ start_dynamic_masking 
++-----------------------
++ t
++(1 row)
++
++SET ROLE oscar_the_owner;
++SELECT * FROM t1;
++ i |  t   
++---+------
++ 1 | test
++(1 row)
++
++--SELECT * FROM mask.t1;
++-- SHOULD FAIL
++SAVEPOINT fail_stop_engine;
++SELECT anon.stop_dynamic_masking();
++ERROR:  permission denied for schema mask
++CONTEXT:  SQL statement "DROP VIEW mask.t1;"
++PL/pgSQL function anon.mask_drop_view(oid) line 3 at EXECUTE
++SQL statement "SELECT anon.mask_drop_view(oid)
++  FROM pg_catalog.pg_class
++  WHERE relnamespace=quote_ident(pg_catalog.current_setting('anon.sourceschema'))::REGNAMESPACE
++  AND relkind IN ('r','p','f')"
++PL/pgSQL function anon.stop_dynamic_masking() line 22 at PERFORM
++ROLLBACK TO fail_stop_engine;
++RESET ROLE;
++SELECT anon.stop_dynamic_masking();
++NOTICE:  The previous priviledges of 'mallory_the_masked_user' are not restored. You need to grant them manually.
++ stop_dynamic_masking 
++----------------------
++ t
++(1 row)
++
++SET ROLE oscar_the_owner;
++-- SHOULD FAIL
++SAVEPOINT fail_seclabel_on_role;
++SECURITY LABEL FOR anon ON ROLE mallory_the_masked_user IS NULL;
++ERROR:  permission denied
++DETAIL:  The current user must have the CREATEROLE attribute.
++ROLLBACK TO fail_seclabel_on_role;
++ROLLBACK;

compute/patches/pgvector.patch

@@ -15,7 +15,7 @@ index 7a4b88c..56678af 100644
  HEADERS = src/halfvec.h src/sparsevec.h src/vector.h
  
 diff --git a/src/hnswbuild.c b/src/hnswbuild.c
-index b667478..1298aa1 100644
+index b667478..dc95d89 100644
 --- a/src/hnswbuild.c
 +++ b/src/hnswbuild.c
 @@ -843,9 +843,17 @@ HnswParallelBuildMain(dsm_segment *seg, shm_toc *toc)
@@ -36,7 +36,7 @@ index b667478..1298aa1 100644
  	/* Close relations within worker */
  	index_close(indexRel, indexLockmode);
  	table_close(heapRel, heapLockmode);
-@@ -1100,13 +1108,25 @@ BuildIndex(Relation heap, Relation index, IndexInfo *indexInfo,
+@@ -1100,12 +1108,39 @@ BuildIndex(Relation heap, Relation index, IndexInfo *indexInfo,
  	SeedRandom(42);
  #endif
  
@@ -48,17 +48,32 @@ index b667478..1298aa1 100644
  
  	BuildGraph(buildstate, forkNum);
  
+-	if (RelationNeedsWAL(index) || forkNum == INIT_FORKNUM)
 +#ifdef NEON_SMGR
 +	smgr_finish_unlogged_build_phase_1(RelationGetSmgr(index));
 +#endif
 +
- 	if (RelationNeedsWAL(index) || forkNum == INIT_FORKNUM)
++	if (RelationNeedsWAL(index) || forkNum == INIT_FORKNUM) {
  		log_newpage_range(index, forkNum, 0, RelationGetNumberOfBlocksInFork(index, forkNum), true);
- 
++#ifdef NEON_SMGR
++		{
++#if PG_VERSION_NUM >= 160000
++			RelFileLocator rlocator = RelationGetSmgr(index)->smgr_rlocator.locator;
++#else
++			RelFileNode rlocator = RelationGetSmgr(index)->smgr_rnode.node;
++#endif
++			if (set_lwlsn_block_range_hook)
++				set_lwlsn_block_range_hook(XactLastRecEnd, rlocator,
++										   MAIN_FORKNUM, 0, RelationGetNumberOfBlocks(index));
++			if (set_lwlsn_relation_hook)
++				set_lwlsn_relation_hook(XactLastRecEnd, rlocator, MAIN_FORKNUM);
++		}
++#endif
++	}
++
 +#ifdef NEON_SMGR
 +	smgr_end_unlogged_build(RelationGetSmgr(index));
 +#endif
-+
+ 
  	FreeBuildState(buildstate);
  }
- 

compute/patches/rum.patch

@@ -1,5 +1,5 @@
 diff --git a/src/ruminsert.c b/src/ruminsert.c
-index 255e616..1c6edb7 100644
+index 255e616..7a2240f 100644
 --- a/src/ruminsert.c
 +++ b/src/ruminsert.c
 @@ -628,6 +628,10 @@ rumbuild(Relation heap, Relation index, struct IndexInfo *indexInfo)
@@ -24,12 +24,24 @@ index 255e616..1c6edb7 100644
  	/*
  	 * Write index to xlog
  	 */
-@@ -713,6 +721,10 @@ rumbuild(Relation heap, Relation index, struct IndexInfo *indexInfo)
+@@ -713,6 +721,22 @@ rumbuild(Relation heap, Relation index, struct IndexInfo *indexInfo)
  		UnlockReleaseBuffer(buffer);
  	}
  
 +#ifdef NEON_SMGR
-+	smgr_end_unlogged_build(index->rd_smgr);
++	{
++#if PG_VERSION_NUM >= 160000
++		RelFileLocator rlocator = RelationGetSmgr(index)->smgr_rlocator.locator;
++#else
++		RelFileNode rlocator = RelationGetSmgr(index)->smgr_rnode.node;
++#endif
++		if (set_lwlsn_block_range_hook)
++			set_lwlsn_block_range_hook(XactLastRecEnd, rlocator, MAIN_FORKNUM, 0, RelationGetNumberOfBlocks(index));
++		if (set_lwlsn_relation_hook)
++			set_lwlsn_relation_hook(XactLastRecEnd, rlocator, MAIN_FORKNUM);
++
++		smgr_end_unlogged_build(index->rd_smgr);
++	}
 +#endif
 +
  	/*

compute/vm-image-spec-bookworm.yaml

@@ -22,7 +22,7 @@ commands:
   - name: local_proxy
     user: postgres
     sysvInitAction: respawn
-    shell: 'RUST_LOG="info,proxy::serverless::sql_over_http=warn" /usr/local/bin/local_proxy --config-path /etc/local_proxy/config.json --pid-path /etc/local_proxy/pid --http 0.0.0.0:10432'
+    shell: '/usr/local/bin/local_proxy --config-path /etc/local_proxy/config.json --pid-path /etc/local_proxy/pid --http 0.0.0.0:10432'
   - name: postgres-exporter
     user: nobody
     sysvInitAction: respawn

compute/vm-image-spec-bullseye.yaml

@@ -22,7 +22,7 @@ commands:
   - name: local_proxy
     user: postgres
     sysvInitAction: respawn
-    shell: 'RUST_LOG="info,proxy::serverless::sql_over_http=warn" /usr/local/bin/local_proxy --config-path /etc/local_proxy/config.json --pid-path /etc/local_proxy/pid --http 0.0.0.0:10432'
+    shell: '/usr/local/bin/local_proxy --config-path /etc/local_proxy/config.json --pid-path /etc/local_proxy/pid --http 0.0.0.0:10432'
   - name: postgres-exporter
     user: nobody
     sysvInitAction: respawn

compute_tools/src/bin/compute_ctl.rs

@@ -29,12 +29,13 @@
 //! ```sh
 //! compute_ctl -D /var/db/postgres/compute \
 //!             -C 'postgresql://cloud_admin@localhost/postgres' \
-//!             -c /var/db/postgres/configs/config.json \
+//!             -S /var/db/postgres/specs/current.json \
 //!             -b /usr/local/bin/postgres \
 //!             -r http://pg-ext-s3-gateway \
 //! ```
 use std::ffi::OsString;
 use std::fs::File;
+use std::path::Path;
 use std::process::exit;
 use std::sync::mpsc;
 use std::thread;
@@ -42,7 +43,8 @@ use std::time::Duration;
 
 use anyhow::{Context, Result};
 use clap::Parser;
-use compute_api::responses::ComputeConfig;
+use compute_api::responses::ComputeCtlConfig;
+use compute_api::spec::ComputeSpec;
 use compute_tools::compute::{
     BUILD_TAG, ComputeNode, ComputeNodeParams, forward_termination_signal,
 };
@@ -57,13 +59,24 @@ use tracing::{error, info};
 use url::Url;
 use utils::failpoint_support;
 
+// Compatibility hack: if the control plane specified any remote-ext-config
+// use the default value for extension storage proxy gateway.
+// Remove this once the control plane is updated to pass the gateway URL
+fn parse_remote_ext_config(arg: &str) -> Result<String> {
+    if arg.starts_with("http") {
+        Ok(arg.trim_end_matches('/').to_string())
+    } else {
+        Ok("http://pg-ext-s3-gateway".to_string())
+    }
+}
+
 #[derive(Parser)]
 #[command(rename_all = "kebab-case")]
 struct Cli {
     #[arg(short = 'b', long, default_value = "postgres", env = "POSTGRES_PATH")]
     pub pgbin: String,
 
-    #[arg(short = 'r', long)]
+    #[arg(short = 'r', long, value_parser = parse_remote_ext_config)]
     pub remote_ext_config: Option<String>,
 
     /// The port to bind the external listening HTTP server to. Clients running
@@ -105,19 +118,16 @@ struct Cli {
     #[arg(long)]
     pub set_disk_quota_for_fs: Option<String>,
 
-    #[arg(short = 'c', long)]
-    pub config: Option<OsString>,
+    #[arg(short = 's', long = "spec", group = "spec")]
+    pub spec_json: Option<String>,
+
+    #[arg(short = 'S', long, group = "spec-path")]
+    pub spec_path: Option<OsString>,
 
     #[arg(short = 'i', long, group = "compute-id")]
     pub compute_id: String,
 
-    #[arg(
-        short = 'p',
-        long,
-        conflicts_with = "config",
-        value_name = "CONTROL_PLANE_API_BASE_URL",
-        requires = "compute-id"
-    )]
+    #[arg(short = 'p', long, conflicts_with_all = ["spec", "spec-path"], value_name = "CONTROL_PLANE_API_BASE_URL")]
     pub control_plane_uri: Option<String>,
 }
 
@@ -126,7 +136,7 @@ fn main() -> Result<()> {
 
     let scenario = failpoint_support::init();
 
-    // For historical reasons, the main thread that processes the config and launches postgres
+    // For historical reasons, the main thread that processes the spec and launches postgres
     // is synchronous, but we always have this tokio runtime available and we "enter" it so
     // that you can use tokio::spawn() and tokio::runtime::Handle::current().block_on(...)
     // from all parts of compute_ctl.
@@ -142,7 +152,7 @@ fn main() -> Result<()> {
 
     let connstr = Url::parse(&cli.connstr).context("cannot parse connstr as a URL")?;
 
-    let config = get_config(&cli)?;
+    let cli_spec = try_spec_from_cli(&cli)?;
 
     let compute_node = ComputeNode::new(
         ComputeNodeParams {
@@ -162,8 +172,10 @@ fn main() -> Result<()> {
             cgroup: cli.cgroup,
             #[cfg(target_os = "linux")]
             vm_monitor_addr: cli.vm_monitor_addr,
+            live_config_allowed: cli_spec.live_config_allowed,
         },
-        config,
+        cli_spec.spec,
+        cli_spec.compute_ctl_config,
     )?;
 
     let exit_code = compute_node.run()?;
@@ -188,17 +200,37 @@ async fn init() -> Result<()> {
     Ok(())
 }
 
-fn get_config(cli: &Cli) -> Result<ComputeConfig> {
-    // First, read the config from the path if provided
-    if let Some(ref config) = cli.config {
-        let file = File::open(config)?;
-        return Ok(serde_json::from_reader(&file)?);
+fn try_spec_from_cli(cli: &Cli) -> Result<CliSpecParams> {
+    // First, try to get cluster spec from the cli argument
+    if let Some(ref spec_json) = cli.spec_json {
+        info!("got spec from cli argument {}", spec_json);
+        return Ok(CliSpecParams {
+            spec: Some(serde_json::from_str(spec_json)?),
+            compute_ctl_config: ComputeCtlConfig::default(),
+            live_config_allowed: false,
+        });
     }
 
-    // If the config wasn't provided in the CLI arguments, then retrieve it from
-    // the control plane
-    match get_config_from_control_plane(cli.control_plane_uri.as_ref().unwrap(), &cli.compute_id) {
-        Ok(config) => Ok(config),
+    // Second, try to read it from the file if path is provided
+    if let Some(ref spec_path) = cli.spec_path {
+        let file = File::open(Path::new(spec_path))?;
+        return Ok(CliSpecParams {
+            spec: Some(serde_json::from_reader(file)?),
+            compute_ctl_config: ComputeCtlConfig::default(),
+            live_config_allowed: true,
+        });
+    }
+
+    if cli.control_plane_uri.is_none() {
+        panic!("must specify --control-plane-uri");
+    };
+
+    match get_spec_from_control_plane(cli.control_plane_uri.as_ref().unwrap(), &cli.compute_id) {
+        Ok(resp) => Ok(CliSpecParams {
+            spec: resp.0,
+            compute_ctl_config: resp.1,
+            live_config_allowed: true,
+        }),
         Err(e) => {
             error!(
                 "cannot get response from control plane: {}\n\
@@ -210,6 +242,14 @@ fn get_config(cli: &Cli) -> Result<ComputeConfig> {
     }
 }
 
+struct CliSpecParams {
+    /// If a spec was provided via CLI or file, the [`ComputeSpec`]
+    spec: Option<ComputeSpec>,
+    #[allow(dead_code)]
+    compute_ctl_config: ComputeCtlConfig,
+    live_config_allowed: bool,
+}
+
 fn deinit_and_exit(exit_code: Option<i32>) -> ! {
     // Shutdown trace pipeline gracefully, so that it has a chance to send any
     // pending traces before we exit. Shutting down OTEL tracing provider may

compute_tools/src/catalog.rs

@@ -98,15 +98,13 @@ pub async fn get_database_schema(
         .kill_on_drop(true)
         .spawn()?;
 
-    let stdout = cmd
-        .stdout
-        .take()
-        .ok_or_else(|| std::io::Error::other("Failed to capture stdout."))?;
+    let stdout = cmd.stdout.take().ok_or_else(|| {
+        std::io::Error::new(std::io::ErrorKind::Other, "Failed to capture stdout.")
+    })?;
 
-    let stderr = cmd
-        .stderr
-        .take()
-        .ok_or_else(|| std::io::Error::other("Failed to capture stderr."))?;
+    let stderr = cmd.stderr.take().ok_or_else(|| {
+        std::io::Error::new(std::io::ErrorKind::Other, "Failed to capture stderr.")
+    })?;
 
     let mut stdout_reader = FramedRead::new(stdout, BytesCodec::new());
     let stderr_reader = BufReader::new(stderr);
@@ -130,7 +128,8 @@ pub async fn get_database_schema(
                 }
             });
 
-            return Err(SchemaDumpError::IO(std::io::Error::other(
+            return Err(SchemaDumpError::IO(std::io::Error::new(
+                std::io::ErrorKind::Other,
                 "failed to start pg_dump",
             )));
         }

compute_tools/src/compute.rs

@@ -11,7 +11,7 @@ use std::{env, fs};
 use anyhow::{Context, Result};
 use chrono::{DateTime, Utc};
 use compute_api::privilege::Privilege;
-use compute_api::responses::{ComputeConfig, ComputeCtlConfig, ComputeMetrics, ComputeStatus};
+use compute_api::responses::{ComputeCtlConfig, ComputeMetrics, ComputeStatus};
 use compute_api::spec::{
     ComputeAudit, ComputeFeature, ComputeMode, ComputeSpec, ExtVersion, PgIdent,
 };
@@ -93,6 +93,20 @@ pub struct ComputeNodeParams {
 
     /// the address of extension storage proxy gateway
     pub ext_remote_storage: Option<String>,
+
+    /// We should only allow live re- / configuration of the compute node if
+    /// it uses 'pull model', i.e. it can go to control-plane and fetch
+    /// the latest configuration. Otherwise, there could be a case:
+    /// - we start compute with some spec provided as argument
+    /// - we push new spec and it does reconfiguration
+    /// - but then something happens and compute pod / VM is destroyed,
+    ///   so k8s controller starts it again with the **old** spec
+    ///
+    /// and the same for empty computes:
+    /// - we started compute without any spec
+    /// - we push spec and it does configuration
+    /// - but then it is restarted without any spec again
+    pub live_config_allowed: bool,
 }
 
 /// Compute node info shared across several `compute_ctl` threads.
@@ -303,7 +317,11 @@ struct StartVmMonitorResult {
 }
 
 impl ComputeNode {
-    pub fn new(params: ComputeNodeParams, config: ComputeConfig) -> Result<Self> {
+    pub fn new(
+        params: ComputeNodeParams,
+        cli_spec: Option<ComputeSpec>,
+        compute_ctl_config: ComputeCtlConfig,
+    ) -> Result<Self> {
         let connstr = params.connstr.as_str();
         let conn_conf = postgres::config::Config::from_str(connstr)
             .context("cannot build postgres config from connstr")?;
@@ -311,8 +329,8 @@ impl ComputeNode {
             .context("cannot build tokio postgres config from connstr")?;
 
         let mut new_state = ComputeState::new();
-        if let Some(spec) = config.spec {
-            let pspec = ParsedSpec::try_from(spec).map_err(|msg| anyhow::anyhow!(msg))?;
+        if let Some(cli_spec) = cli_spec {
+            let pspec = ParsedSpec::try_from(cli_spec).map_err(|msg| anyhow::anyhow!(msg))?;
             new_state.pspec = Some(pspec);
         }
 
@@ -323,7 +341,7 @@ impl ComputeNode {
             state: Mutex::new(new_state),
             state_changed: Condvar::new(),
             ext_download_progress: RwLock::new(HashMap::new()),
-            compute_ctl_config: config.compute_ctl_config,
+            compute_ctl_config,
         })
     }
 
@@ -519,14 +537,11 @@ impl ComputeNode {
 
         let pspec = compute_state.pspec.as_ref().expect("spec must be set");
         info!(
-            "starting compute for project {}, operation {}, tenant {}, timeline {}, project {}, branch {}, endpoint {}, features {:?}, spec.remote_extensions {:?}",
+            "starting compute for project {}, operation {}, tenant {}, timeline {}, features {:?}, spec.remote_extensions {:?}",
             pspec.spec.cluster.cluster_id.as_deref().unwrap_or("None"),
             pspec.spec.operation_uuid.as_deref().unwrap_or("None"),
             pspec.tenant_id,
             pspec.timeline_id,
-            pspec.spec.project_id.as_deref().unwrap_or("None"),
-            pspec.spec.branch_id.as_deref().unwrap_or("None"),
-            pspec.spec.endpoint_id.as_deref().unwrap_or("None"),
             pspec.spec.features,
             pspec.spec.remote_extensions,
         );
@@ -630,47 +645,31 @@ impl ComputeNode {
             });
         }
 
-        // Configure and start rsyslog for compliance audit logging
-        match pspec.spec.audit_log_level {
-            ComputeAudit::Hipaa | ComputeAudit::Extended | ComputeAudit::Full => {
-                let remote_endpoint =
-                    std::env::var("AUDIT_LOGGING_ENDPOINT").unwrap_or("".to_string());
-                if remote_endpoint.is_empty() {
-                    anyhow::bail!("AUDIT_LOGGING_ENDPOINT is empty");
-                }
-
-                let log_directory_path = Path::new(&self.params.pgdata).join("log");
-                let log_directory_path = log_directory_path.to_string_lossy().to_string();
-
-                // Add project_id,endpoint_id tag to identify the logs.
-                //
-                // These ids are passed from cplane,
-                // for backwards compatibility (old computes that don't have them),
-                // we set them to None.
-                // TODO: Clean up this code when all computes have them.
-                let tag: Option<String> = match (
-                    pspec.spec.project_id.as_deref(),
-                    pspec.spec.endpoint_id.as_deref(),
-                ) {
-                    (Some(project_id), Some(endpoint_id)) => {
-                        Some(format!("{project_id}/{endpoint_id}"))
-                    }
-                    (Some(project_id), None) => Some(format!("{project_id}/None")),
-                    (None, Some(endpoint_id)) => Some(format!("None,{endpoint_id}")),
-                    (None, None) => None,
-                };
-
-                configure_audit_rsyslog(log_directory_path.clone(), tag, &remote_endpoint)?;
-
-                // Launch a background task to clean up the audit logs
-                launch_pgaudit_gc(log_directory_path);
+        // Configure and start rsyslog for HIPAA if necessary
+        if let ComputeAudit::Hipaa = pspec.spec.audit_log_level {
+            let remote_endpoint = std::env::var("AUDIT_LOGGING_ENDPOINT").unwrap_or("".to_string());
+            if remote_endpoint.is_empty() {
+                anyhow::bail!("AUDIT_LOGGING_ENDPOINT is empty");
             }
-            _ => {}
+
+            let log_directory_path = Path::new(&self.params.pgdata).join("log");
+            let log_directory_path = log_directory_path.to_string_lossy().to_string();
+            configure_audit_rsyslog(log_directory_path.clone(), "hipaa", &remote_endpoint)?;
+
+            // Launch a background task to clean up the audit logs
+            launch_pgaudit_gc(log_directory_path);
         }
 
         // Configure and start rsyslog for Postgres logs export
-        let conf = PostgresLogsRsyslogConfig::new(pspec.spec.logs_export_host.as_deref());
-        configure_postgres_logs_export(conf)?;
+        if self.has_feature(ComputeFeature::PostgresLogsExport) {
+            if let Some(ref project_id) = pspec.spec.cluster.cluster_id {
+                let host = PostgresLogsRsyslogConfig::default_host(project_id);
+                let conf = PostgresLogsRsyslogConfig::new(Some(&host));
+                configure_postgres_logs_export(conf)?;
+            } else {
+                warn!("not configuring rsyslog for Postgres logs export: project ID is missing")
+            }
+        }
 
         // Launch remaining service threads
         let _monitor_handle = launch_monitor(self);
@@ -1574,10 +1573,6 @@ impl ComputeNode {
             });
         }
 
-        // Reconfigure rsyslog for Postgres logs export
-        let conf = PostgresLogsRsyslogConfig::new(spec.logs_export_host.as_deref());
-        configure_postgres_logs_export(conf)?;
-
         // Write new config
         let pgdata_path = Path::new(&self.params.pgdata);
         config::write_postgres_conf(

compute_tools/src/config.rs

@@ -7,7 +7,7 @@ use std::io::prelude::*;
 use std::path::Path;
 
 use compute_api::responses::TlsConfig;
-use compute_api::spec::{ComputeAudit, ComputeMode, ComputeSpec, GenericOption};
+use compute_api::spec::{ComputeAudit, ComputeFeature, ComputeMode, ComputeSpec, GenericOption};
 
 use crate::pg_helpers::{
     GenericOptionExt, GenericOptionsSearch, PgOptionsSerialize, escape_conf_value,
@@ -89,15 +89,6 @@ pub fn write_postgres_conf(
             escape_conf_value(&s.to_string())
         )?;
     }
-    if let Some(s) = &spec.project_id {
-        writeln!(file, "neon.project_id={}", escape_conf_value(s))?;
-    }
-    if let Some(s) = &spec.branch_id {
-        writeln!(file, "neon.branch_id={}", escape_conf_value(s))?;
-    }
-    if let Some(s) = &spec.endpoint_id {
-        writeln!(file, "neon.endpoint_id={}", escape_conf_value(s))?;
-    }
 
     // tls
     if let Some(tls_config) = tls_config {
@@ -178,7 +169,7 @@ pub fn write_postgres_conf(
     // and don't allow the user or the control plane admin to change them.
     match spec.audit_log_level {
         ComputeAudit::Disabled => {}
-        ComputeAudit::Log | ComputeAudit::Base => {
+        ComputeAudit::Log => {
             writeln!(file, "# Managed by compute_ctl base audit settings: start")?;
             writeln!(file, "pgaudit.log='ddl,role'")?;
             // Disable logging of catalog queries to reduce the noise
@@ -202,20 +193,16 @@ pub fn write_postgres_conf(
             }
             writeln!(file, "# Managed by compute_ctl base audit settings: end")?;
         }
-        ComputeAudit::Hipaa | ComputeAudit::Extended | ComputeAudit::Full => {
+        ComputeAudit::Hipaa => {
             writeln!(
                 file,
                 "# Managed by compute_ctl compliance audit settings: begin"
             )?;
-            // Enable logging of parameters.
-            // This is very verbose and may contain sensitive data.
-            if spec.audit_log_level == ComputeAudit::Full {
-                writeln!(file, "pgaudit.log_parameter=on")?;
-                writeln!(file, "pgaudit.log='all'")?;
-            } else {
-                writeln!(file, "pgaudit.log_parameter=off")?;
-                writeln!(file, "pgaudit.log='all, -misc'")?;
-            }
+            // This log level is very verbose
+            // but this is necessary for HIPAA compliance.
+            // Exclude 'misc' category, because it doesn't contain anythig relevant.
+            writeln!(file, "pgaudit.log='all, -misc'")?;
+            writeln!(file, "pgaudit.log_parameter=on")?;
             // Disable logging of catalog queries
             // The catalog doesn't contain sensitive data, so we don't need to audit it.
             writeln!(file, "pgaudit.log_catalog=off")?;
@@ -268,7 +255,7 @@ pub fn write_postgres_conf(
 
     // We need Postgres to send logs to rsyslog so that we can forward them
     // further to customers' log aggregation systems.
-    if spec.logs_export_host.is_some() {
+    if spec.features.contains(&ComputeFeature::PostgresLogsExport) {
         writeln!(file, "log_destination='stderr,syslog'")?;
     }
 

compute_tools/src/http/extract/mod.rs

@@ -6,5 +6,4 @@ pub(crate) mod request_id;
 pub(crate) use json::Json;
 pub(crate) use path::Path;
 pub(crate) use query::Query;
-#[allow(unused)]
 pub(crate) use request_id::RequestId;

compute_tools/src/http/middleware/authorize.rs

@@ -1,19 +1,24 @@
-use std::collections::HashSet;
+use std::{collections::HashSet, net::SocketAddr};
 
 use anyhow::{Result, anyhow};
-use axum::{RequestExt, body::Body};
+use axum::{RequestExt, body::Body, extract::ConnectInfo};
 use axum_extra::{
     TypedHeader,
     headers::{Authorization, authorization::Bearer},
 };
-use compute_api::requests::ComputeClaims;
 use futures::future::BoxFuture;
 use http::{Request, Response, StatusCode};
 use jsonwebtoken::{Algorithm, DecodingKey, TokenData, Validation, jwk::JwkSet};
+use serde::Deserialize;
 use tower_http::auth::AsyncAuthorizeRequest;
-use tracing::{debug, warn};
+use tracing::warn;
 
-use crate::http::JsonResponse;
+use crate::http::{JsonResponse, extract::RequestId};
+
+#[derive(Clone, Debug, Deserialize)]
+pub(in crate::http) struct Claims {
+    compute_id: String,
+}
 
 #[derive(Clone, Debug)]
 pub(in crate::http) struct Authorize {
@@ -52,6 +57,31 @@ impl AsyncAuthorizeRequest<Body> for Authorize {
         let validation = self.validation.clone();
 
         Box::pin(async move {
+            let request_id = request.extract_parts::<RequestId>().await.unwrap();
+
+            // TODO: Remove this stanza after teaching neon_local and the
+            // regression tests to use a JWT + JWKS.
+            //
+            // https://github.com/neondatabase/neon/issues/11316
+            if cfg!(feature = "testing") {
+                warn!(%request_id, "Skipping compute_ctl authorization check");
+
+                return Ok(request);
+            }
+
+            let connect_info = request
+                .extract_parts::<ConnectInfo<SocketAddr>>()
+                .await
+                .unwrap();
+
+            // In the event the request is coming from the loopback interface,
+            // allow all requests
+            if connect_info.ip().is_loopback() {
+                warn!(%request_id, "Bypassed authorization because request is coming from the loopback interface");
+
+                return Ok(request);
+            }
+
             let TypedHeader(Authorization(bearer)) = request
                 .extract_parts::<TypedHeader<Authorization<Bearer>>>()
                 .await
@@ -67,7 +97,7 @@ impl AsyncAuthorizeRequest<Body> for Authorize {
             if data.claims.compute_id != compute_id {
                 return Err(JsonResponse::error(
                     StatusCode::UNAUTHORIZED,
-                    "invalid compute ID in authorization token claims",
+                    "invalid claims in authorization token",
                 ));
             }
 
@@ -82,21 +112,13 @@ impl AsyncAuthorizeRequest<Body> for Authorize {
 
 impl Authorize {
     /// Verify the token using the JSON Web Key set and return the token data.
-    fn verify(
-        jwks: &JwkSet,
-        token: &str,
-        validation: &Validation,
-    ) -> Result<TokenData<ComputeClaims>> {
-        debug_assert!(!jwks.keys.is_empty());
-
-        debug!("verifying token {}", token);
-
+    fn verify(jwks: &JwkSet, token: &str, validation: &Validation) -> Result<TokenData<Claims>> {
         for jwk in jwks.keys.iter() {
             let decoding_key = match DecodingKey::from_jwk(jwk) {
                 Ok(key) => key,
                 Err(e) => {
                     warn!(
-                        "failed to construct decoding key from {}: {}",
+                        "Failed to construct decoding key from {}: {}",
                         jwk.common.key_id.as_ref().unwrap(),
                         e
                     );
@@ -105,11 +127,11 @@ impl Authorize {
                 }
             };
 
-            match jsonwebtoken::decode::<ComputeClaims>(token, &decoding_key, validation) {
+            match jsonwebtoken::decode::<Claims>(token, &decoding_key, validation) {
                 Ok(data) => return Ok(data),
                 Err(e) => {
                     warn!(
-                        "failed to decode authorization token using {}: {}",
+                        "Failed to decode authorization token using {}: {}",
                         jwk.common.key_id.as_ref().unwrap(),
                         e
                     );
@@ -119,6 +141,6 @@ impl Authorize {
             }
         }
 
-        Err(anyhow!("failed to verify authorization token"))
+        Err(anyhow!("Failed to verify authorization token"))
     }
 }

compute_tools/src/http/openapi_spec.yaml

@@ -306,6 +306,36 @@ paths:
               schema:
                 $ref: "#/components/schemas/GenericError"
 
+  /configure_telemetry:
+    post:
+      tags:
+        - Configure
+      summary: Configure rsyslog
+      description: |
+        This API endpoint configures rsyslog to forward Postgres logs
+        to a specified otel collector.
+      operationId: configureTelemetry
+      requestBody:
+        required: true
+        content:
+          application/json:
+            schema:
+              type: object
+              properties:
+                logs_export_host:
+                  type: string
+                  description: |
+                    Hostname and the port of the otel collector. Leave empty to disable logs forwarding.
+                    Example: config-shy-breeze-123-collector-monitoring.neon-telemetry.svc.cluster.local:54526
+      responses:
+        204:
+          description: "Telemetry configured successfully"
+        500:
+          content:
+            application/json:
+              schema:
+                $ref: "#/components/schemas/GenericError"
+
 components:
   securitySchemes:
     JWT:

compute_tools/src/http/routes/configure.rs

@@ -1,9 +1,11 @@
 use std::sync::Arc;
 
+use axum::body::Body;
 use axum::extract::State;
 use axum::response::Response;
-use compute_api::requests::ConfigurationRequest;
+use compute_api::requests::{ConfigurationRequest, ConfigureTelemetryRequest};
 use compute_api::responses::{ComputeStatus, ComputeStatusResponse};
+use compute_api::spec::ComputeFeature;
 use http::StatusCode;
 use tokio::task;
 use tracing::info;
@@ -11,6 +13,7 @@ use tracing::info;
 use crate::compute::{ComputeNode, ParsedSpec};
 use crate::http::JsonResponse;
 use crate::http::extract::Json;
+use crate::rsyslog::{PostgresLogsRsyslogConfig, configure_postgres_logs_export};
 
 // Accept spec in JSON format and request compute configuration. If anything
 // goes wrong after we set the compute status to `ConfigurationPending` and
@@ -22,6 +25,13 @@ pub(in crate::http) async fn configure(
     State(compute): State<Arc<ComputeNode>>,
     request: Json<ConfigurationRequest>,
 ) -> Response {
+    if !compute.params.live_config_allowed {
+        return JsonResponse::error(
+            StatusCode::PRECONDITION_FAILED,
+            "live configuration is not allowed for this compute node".to_string(),
+        );
+    }
+
     let pspec = match ParsedSpec::try_from(request.spec.clone()) {
         Ok(p) => p,
         Err(e) => return JsonResponse::error(StatusCode::BAD_REQUEST, e),
@@ -85,3 +95,25 @@ pub(in crate::http) async fn configure(
 
     JsonResponse::success(StatusCode::OK, body)
 }
+
+pub(in crate::http) async fn configure_telemetry(
+    State(compute): State<Arc<ComputeNode>>,
+    request: Json<ConfigureTelemetryRequest>,
+) -> Response {
+    if !compute.has_feature(ComputeFeature::PostgresLogsExport) {
+        return JsonResponse::error(
+            StatusCode::PRECONDITION_FAILED,
+            "Postgres logs export feature is not enabled".to_string(),
+        );
+    }
+
+    let conf = PostgresLogsRsyslogConfig::new(request.logs_export_host.as_deref());
+    if let Err(err) = configure_postgres_logs_export(conf) {
+        return JsonResponse::error(StatusCode::INTERNAL_SERVER_ERROR, err.to_string());
+    }
+
+    Response::builder()
+        .status(StatusCode::NO_CONTENT)
+        .body(Body::from(""))
+        .unwrap()
+}

compute_tools/src/http/server.rs

@@ -87,6 +87,7 @@ impl From<&Server> for Router<Arc<ComputeNode>> {
                 let authenticated_router = Router::<Arc<ComputeNode>>::new()
                     .route("/check_writability", post(check_writability::is_writable))
                     .route("/configure", post(configure::configure))
+                    .route("/configure_telemetry", post(configure::configure_telemetry))
                     .route("/database_schema", get(database_schema::get_schema_dump))
                     .route("/dbs_and_roles", get(dbs_and_roles::get_catalog_objects))
                     .route("/insights", get(insights::get_insights))

compute_tools/src/metrics.rs

@@ -19,13 +19,13 @@ pub(crate) static INSTALLED_EXTENSIONS: Lazy<UIntGaugeVec> = Lazy::new(|| {
 // but for all our APIs we defined a 'slug'/method/operationId in the OpenAPI spec.
 // And it's fair to call it a 'RPC' (Remote Procedure Call).
 pub enum CPlaneRequestRPC {
-    GetConfig,
+    GetSpec,
 }
 
 impl CPlaneRequestRPC {
     pub fn as_str(&self) -> &str {
         match self {
-            CPlaneRequestRPC::GetConfig => "GetConfig",
+            CPlaneRequestRPC::GetSpec => "GetSpec",
         }
     }
 }

compute_tools/src/rsyslog.rs

@@ -50,13 +50,13 @@ fn restart_rsyslog() -> Result<()> {
 
 pub fn configure_audit_rsyslog(
     log_directory: String,
-    tag: Option<String>,
+    tag: &str,
     remote_endpoint: &str,
 ) -> Result<()> {
     let config_content: String = format!(
         include_str!("config_template/compute_audit_rsyslog_template.conf"),
         log_directory = log_directory,
-        tag = tag.unwrap_or("".to_string()),
+        tag = tag,
         remote_endpoint = remote_endpoint
     );
 
@@ -119,9 +119,16 @@ impl<'a> PostgresLogsRsyslogConfig<'a> {
         };
         Ok(config_content)
     }
+
+    /// Returns the default host for otel collector that receives Postgres logs
+    pub fn default_host(project_id: &str) -> String {
+        format!(
+            "config-{}-collector.neon-telemetry.svc.cluster.local:10514",
+            project_id
+        )
+    }
 }
 
-/// Writes rsyslogd configuration for Postgres logs export and restarts rsyslog.
 pub fn configure_postgres_logs_export(conf: PostgresLogsRsyslogConfig) -> Result<()> {
     let new_config = conf.build()?;
     let current_config = PostgresLogsRsyslogConfig::current_config()?;
@@ -254,5 +261,16 @@ mod tests {
             let res = conf.build();
             assert!(res.is_err());
         }
+
+        {
+            // Verify config with default host
+            let host = PostgresLogsRsyslogConfig::default_host("shy-breeze-123");
+            let conf = PostgresLogsRsyslogConfig::new(Some(&host));
+            let res = conf.build();
+            assert!(res.is_ok());
+            let conf_str = res.unwrap();
+            assert!(conf_str.contains(r#"shy-breeze-123"#));
+            assert!(conf_str.contains(r#"port="10514""#));
+        }
     }
 }

compute_tools/src/spec.rs

@@ -3,8 +3,9 @@ use std::path::Path;
 
 use anyhow::{Result, anyhow, bail};
 use compute_api::responses::{
-    ComputeConfig, ControlPlaneComputeStatus, ControlPlaneConfigResponse,
+    ComputeCtlConfig, ControlPlaneComputeStatus, ControlPlaneSpecResponse,
 };
+use compute_api::spec::ComputeSpec;
 use reqwest::StatusCode;
 use tokio_postgres::Client;
 use tracing::{error, info, instrument};
@@ -20,7 +21,7 @@ use crate::params::PG_HBA_ALL_MD5;
 fn do_control_plane_request(
     uri: &str,
     jwt: &str,
-) -> Result<ControlPlaneConfigResponse, (bool, String, String)> {
+) -> Result<ControlPlaneSpecResponse, (bool, String, String)> {
     let resp = reqwest::blocking::Client::new()
         .get(uri)
         .header("Authorization", format!("Bearer {}", jwt))
@@ -28,14 +29,14 @@ fn do_control_plane_request(
         .map_err(|e| {
             (
                 true,
-                format!("could not perform request to control plane: {:?}", e),
+                format!("could not perform spec request to control plane: {:?}", e),
                 UNKNOWN_HTTP_STATUS.to_string(),
             )
         })?;
 
     let status = resp.status();
     match status {
-        StatusCode::OK => match resp.json::<ControlPlaneConfigResponse>() {
+        StatusCode::OK => match resp.json::<ControlPlaneSpecResponse>() {
             Ok(spec_resp) => Ok(spec_resp),
             Err(e) => Err((
                 true,
@@ -68,35 +69,40 @@ fn do_control_plane_request(
     }
 }
 
-/// Request config from the control-plane by compute_id. If
-/// `NEON_CONTROL_PLANE_TOKEN` env variable is set, it will be used for
-/// authorization.
-pub fn get_config_from_control_plane(base_uri: &str, compute_id: &str) -> Result<ComputeConfig> {
+/// Request spec from the control-plane by compute_id. If `NEON_CONTROL_PLANE_TOKEN`
+/// env variable is set, it will be used for authorization.
+pub fn get_spec_from_control_plane(
+    base_uri: &str,
+    compute_id: &str,
+) -> Result<(Option<ComputeSpec>, ComputeCtlConfig)> {
     let cp_uri = format!("{base_uri}/compute/api/v2/computes/{compute_id}/spec");
-    let jwt: String = std::env::var("NEON_CONTROL_PLANE_TOKEN").unwrap_or_default();
+    let jwt: String = match std::env::var("NEON_CONTROL_PLANE_TOKEN") {
+        Ok(v) => v,
+        Err(_) => "".to_string(),
+    };
     let mut attempt = 1;
 
-    info!("getting config from control plane: {}", cp_uri);
+    info!("getting spec from control plane: {}", cp_uri);
 
     // Do 3 attempts to get spec from the control plane using the following logic:
     // - network error -> then retry
     // - compute id is unknown or any other error -> bail out
     // - no spec for compute yet (Empty state) -> return Ok(None)
-    // - got config -> return Ok(Some(config))
+    // - got spec -> return Ok(Some(spec))
     while attempt < 4 {
         let result = match do_control_plane_request(&cp_uri, &jwt) {
-            Ok(config_resp) => {
+            Ok(spec_resp) => {
                 CPLANE_REQUESTS_TOTAL
                     .with_label_values(&[
-                        CPlaneRequestRPC::GetConfig.as_str(),
+                        CPlaneRequestRPC::GetSpec.as_str(),
                         &StatusCode::OK.to_string(),
                     ])
                     .inc();
-                match config_resp.status {
-                    ControlPlaneComputeStatus::Empty => Ok(config_resp.into()),
+                match spec_resp.status {
+                    ControlPlaneComputeStatus::Empty => Ok((None, spec_resp.compute_ctl_config)),
                     ControlPlaneComputeStatus::Attached => {
-                        if config_resp.spec.is_some() {
-                            Ok(config_resp.into())
+                        if let Some(spec) = spec_resp.spec {
+                            Ok((Some(spec), spec_resp.compute_ctl_config))
                         } else {
                             bail!("compute is attached, but spec is empty")
                         }
@@ -105,7 +111,7 @@ pub fn get_config_from_control_plane(base_uri: &str, compute_id: &str) -> Result
             }
             Err((retry, msg, status)) => {
                 CPLANE_REQUESTS_TOTAL
-                    .with_label_values(&[CPlaneRequestRPC::GetConfig.as_str(), &status])
+                    .with_label_values(&[CPlaneRequestRPC::GetSpec.as_str(), &status])
                     .inc();
                 if retry {
                     Err(anyhow!(msg))
@@ -116,7 +122,7 @@ pub fn get_config_from_control_plane(base_uri: &str, compute_id: &str) -> Result
         };
 
         if let Err(e) = &result {
-            error!("attempt {} to get config failed with: {}", attempt, e);
+            error!("attempt {} to get spec failed with: {}", attempt, e);
         } else {
             return result;
         }
@@ -127,13 +133,13 @@ pub fn get_config_from_control_plane(base_uri: &str, compute_id: &str) -> Result
 
     // All attempts failed, return error.
     Err(anyhow::anyhow!(
-        "Exhausted all attempts to retrieve the config from the control plane"
+        "Exhausted all attempts to retrieve the spec from the control plane"
     ))
 }
 
 /// Check `pg_hba.conf` and update if needed to allow external connections.
 pub fn update_pg_hba(pgdata_path: &Path) -> Result<()> {
-    // XXX: consider making it a part of config.json
+    // XXX: consider making it a part of spec.json
     let pghba_path = pgdata_path.join("pg_hba.conf");
 
     if config::line_in_file(&pghba_path, PG_HBA_ALL_MD5)? {
@@ -147,7 +153,7 @@ pub fn update_pg_hba(pgdata_path: &Path) -> Result<()> {
 
 /// Create a standby.signal file
 pub fn add_standby_signal(pgdata_path: &Path) -> Result<()> {
-    // XXX: consider making it a part of config.json
+    // XXX: consider making it a part of spec.json
     let signalfile = pgdata_path.join("standby.signal");
 
     if !signalfile.exists() {

compute_tools/src/spec_apply.rs

@@ -278,12 +278,12 @@ impl ComputeNode {
             // so that all config operations are audit logged.
             match spec.audit_log_level
             {
-                ComputeAudit::Hipaa | ComputeAudit::Extended | ComputeAudit::Full => {
+                ComputeAudit::Hipaa => {
                     phases.push(CreatePgauditExtension);
                     phases.push(CreatePgauditlogtofileExtension);
                     phases.push(DisablePostgresDBPgAudit);
                 }
-                ComputeAudit::Log | ComputeAudit::Base => {
+                ComputeAudit::Log => {
                     phases.push(CreatePgauditExtension);
                     phases.push(DisablePostgresDBPgAudit);
                 }

control_plane/Cargo.toml

@@ -6,16 +6,13 @@ license.workspace = true
 
 [dependencies]
 anyhow.workspace = true
-base64.workspace = true
 camino.workspace = true
 clap.workspace = true
 comfy-table.workspace = true
 futures.workspace = true
 humantime.workspace = true
-jsonwebtoken.workspace = true
 nix.workspace = true
 once_cell.workspace = true
-pem.workspace = true
 humantime-serde.workspace = true
 hyper0.workspace = true
 regex.workspace = true
@@ -23,8 +20,6 @@ reqwest = { workspace = true, features = ["blocking", "json"] }
 scopeguard.workspace = true
 serde.workspace = true
 serde_json.workspace = true
-sha2.workspace = true
-spki.workspace = true
 thiserror.workspace = true
 toml.workspace = true
 toml_edit.workspace = true

control_plane/src/bin/neon_local.rs

@@ -20,10 +20,8 @@ use compute_api::spec::ComputeMode;
 use control_plane::endpoint::ComputeControlPlane;
 use control_plane::local_env::{
     InitForceMode, LocalEnv, NeonBroker, NeonLocalInitConf, NeonLocalInitPageserverConf,
-    ObjectStorageConf, SafekeeperConf,
+    SafekeeperConf,
 };
-use control_plane::object_storage::OBJECT_STORAGE_DEFAULT_PORT;
-use control_plane::object_storage::ObjectStorage;
 use control_plane::pageserver::PageServerNode;
 use control_plane::safekeeper::SafekeeperNode;
 use control_plane::storage_controller::{
@@ -41,7 +39,7 @@ use pageserver_api::controller_api::{
 use pageserver_api::models::{
     ShardParameters, TenantConfigRequest, TimelineCreateRequest, TimelineInfo,
 };
-use pageserver_api::shard::{DEFAULT_STRIPE_SIZE, ShardCount, ShardStripeSize, TenantShardId};
+use pageserver_api::shard::{ShardCount, ShardStripeSize, TenantShardId};
 use postgres_backend::AuthType;
 use postgres_connection::parse_host_port;
 use safekeeper_api::membership::SafekeeperGeneration;
@@ -63,7 +61,7 @@ const DEFAULT_PAGESERVER_ID: NodeId = NodeId(1);
 const DEFAULT_BRANCH_NAME: &str = "main";
 project_git_version!(GIT_VERSION);
 
-const DEFAULT_PG_VERSION: u32 = 17;
+const DEFAULT_PG_VERSION: u32 = 16;
 
 const DEFAULT_PAGESERVER_CONTROL_PLANE_API: &str = "http://127.0.0.1:1234/upcall/v1/";
 
@@ -93,8 +91,6 @@ enum NeonLocalCmd {
     #[command(subcommand)]
     Safekeeper(SafekeeperCmd),
     #[command(subcommand)]
-    ObjectStorage(ObjectStorageCmd),
-    #[command(subcommand)]
     Endpoint(EndpointCmd),
     #[command(subcommand)]
     Mappings(MappingsCmd),
@@ -458,32 +454,6 @@ enum SafekeeperCmd {
     Restart(SafekeeperRestartCmdArgs),
 }
 
-#[derive(clap::Subcommand)]
-#[clap(about = "Manage object storage")]
-enum ObjectStorageCmd {
-    Start(ObjectStorageStartCmd),
-    Stop(ObjectStorageStopCmd),
-}
-
-#[derive(clap::Args)]
-#[clap(about = "Start object storage")]
-struct ObjectStorageStartCmd {
-    #[clap(short = 't', long, help = "timeout until we fail the command")]
-    #[arg(default_value = "10s")]
-    start_timeout: humantime::Duration,
-}
-
-#[derive(clap::Args)]
-#[clap(about = "Stop object storage")]
-struct ObjectStorageStopCmd {
-    #[arg(value_enum, default_value = "fast")]
-    #[clap(
-        short = 'm',
-        help = "If 'immediate', don't flush repository data at shutdown"
-    )]
-    stop_mode: StopMode,
-}
-
 #[derive(clap::Args)]
 #[clap(about = "Start local safekeeper")]
 struct SafekeeperStartCmdArgs {
@@ -552,7 +522,6 @@ enum EndpointCmd {
     Start(EndpointStartCmdArgs),
     Reconfigure(EndpointReconfigureCmdArgs),
     Stop(EndpointStopCmdArgs),
-    GenerateJwt(EndpointGenerateJwtCmdArgs),
 }
 
 #[derive(clap::Args)]
@@ -700,13 +669,6 @@ struct EndpointStopCmdArgs {
     mode: String,
 }
 
-#[derive(clap::Args)]
-#[clap(about = "Generate a JWT for an endpoint")]
-struct EndpointGenerateJwtCmdArgs {
-    #[clap(help = "Postgres endpoint id")]
-    endpoint_id: String,
-}
-
 #[derive(clap::Subcommand)]
 #[clap(about = "Manage neon_local branch name mappings")]
 enum MappingsCmd {
@@ -797,7 +759,6 @@ fn main() -> Result<()> {
             }
             NeonLocalCmd::StorageBroker(subcmd) => rt.block_on(handle_storage_broker(&subcmd, env)),
             NeonLocalCmd::Safekeeper(subcmd) => rt.block_on(handle_safekeeper(&subcmd, env)),
-            NeonLocalCmd::ObjectStorage(subcmd) => rt.block_on(handle_object_storage(&subcmd, env)),
             NeonLocalCmd::Endpoint(subcmd) => rt.block_on(handle_endpoint(&subcmd, env)),
             NeonLocalCmd::Mappings(subcmd) => handle_mappings(&subcmd, env),
         };
@@ -1014,9 +975,6 @@ fn handle_init(args: &InitCmdArgs) -> anyhow::Result<LocalEnv> {
                     }
                 })
                 .collect(),
-            object_storage: ObjectStorageConf {
-                port: OBJECT_STORAGE_DEFAULT_PORT,
-            },
             pg_distrib_dir: None,
             neon_distrib_dir: None,
             default_tenant_id: TenantId::from_array(std::array::from_fn(|_| 0)),
@@ -1125,7 +1083,7 @@ async fn handle_tenant(subcmd: &TenantCmd, env: &mut local_env::LocalEnv) -> any
                         stripe_size: args
                             .shard_stripe_size
                             .map(ShardStripeSize)
-                            .unwrap_or(DEFAULT_STRIPE_SIZE),
+                            .unwrap_or(ShardParameters::DEFAULT_STRIPE_SIZE),
                     },
                     placement_policy: args.placement_policy.clone(),
                     config: tenant_conf,
@@ -1438,7 +1396,7 @@ async fn handle_endpoint(subcmd: &EndpointCmd, env: &local_env::LocalEnv) -> Res
                     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,
+                    ShardParameters::DEFAULT_STRIPE_SIZE,
                 )
             } else {
                 // Look up the currently attached location of the tenant, and its striping metadata,
@@ -1536,16 +1494,6 @@ async fn handle_endpoint(subcmd: &EndpointCmd, env: &local_env::LocalEnv) -> Res
                 .with_context(|| format!("postgres endpoint {endpoint_id} is not found"))?;
             endpoint.stop(&args.mode, args.destroy)?;
         }
-        EndpointCmd::GenerateJwt(args) => {
-            let endpoint_id = &args.endpoint_id;
-            let endpoint = cplane
-                .endpoints
-                .get(endpoint_id)
-                .with_context(|| format!("postgres endpoint {endpoint_id} is not found"))?;
-            let jwt = endpoint.generate_jwt()?;
-
-            print!("{jwt}");
-        }
     }
 
     Ok(())
@@ -1735,41 +1683,6 @@ async fn handle_safekeeper(subcmd: &SafekeeperCmd, env: &local_env::LocalEnv) ->
     Ok(())
 }
 
-async fn handle_object_storage(subcmd: &ObjectStorageCmd, env: &local_env::LocalEnv) -> Result<()> {
-    use ObjectStorageCmd::*;
-    let storage = ObjectStorage::from_env(env);
-
-    // In tests like test_forward_compatibility or test_graceful_cluster_restart
-    // old neon binaries (without object_storage) are present
-    if !storage.bin.exists() {
-        eprintln!(
-            "{} binary not found. Ignore if this is a compatibility test",
-            storage.bin
-        );
-        return Ok(());
-    }
-
-    match subcmd {
-        Start(ObjectStorageStartCmd { start_timeout }) => {
-            if let Err(e) = storage.start(start_timeout).await {
-                eprintln!("object_storage start failed: {e}");
-                exit(1);
-            }
-        }
-        Stop(ObjectStorageStopCmd { stop_mode }) => {
-            let immediate = match stop_mode {
-                StopMode::Fast => false,
-                StopMode::Immediate => true,
-            };
-            if let Err(e) = storage.stop(immediate) {
-                eprintln!("proxy stop failed: {e}");
-                exit(1);
-            }
-        }
-    };
-    Ok(())
-}
-
 async fn handle_storage_broker(subcmd: &StorageBrokerCmd, env: &local_env::LocalEnv) -> Result<()> {
     match subcmd {
         StorageBrokerCmd::Start(args) => {
@@ -1864,13 +1777,6 @@ async fn handle_start_all_impl(
                     .map_err(|e| e.context(format!("start safekeeper {}", safekeeper.id)))
             });
         }
-
-        js.spawn(async move {
-            ObjectStorage::from_env(env)
-                .start(&retry_timeout)
-                .await
-                .map_err(|e| e.context("start object_storage"))
-        });
     })();
 
     let mut errors = Vec::new();
@@ -1968,11 +1874,6 @@ async fn try_stop_all(env: &local_env::LocalEnv, immediate: bool) {
         }
     }
 
-    let storage = ObjectStorage::from_env(env);
-    if let Err(e) = storage.stop(immediate) {
-        eprintln!("object_storage stop failed: {:#}", e);
-    }
-
     for ps_conf in &env.pageservers {
         let pageserver = PageServerNode::from_env(env, ps_conf);
         if let Err(e) = pageserver.stop(immediate) {

control_plane/src/endpoint.rs

@@ -29,7 +29,7 @@
 //!     compute.log               - log output of `compute_ctl` and `postgres`
 //!     endpoint.json             - serialized `EndpointConf` struct
 //!     postgresql.conf           - postgresql settings
-//!     config.json                 - passed to `compute_ctl`
+//!     spec.json                 - passed to `compute_ctl`
 //!     pgdata/
 //!         postgresql.conf       - copy of postgresql.conf created by `compute_ctl`
 //!         zenith.signal
@@ -42,30 +42,20 @@ use std::path::PathBuf;
 use std::process::Command;
 use std::str::FromStr;
 use std::sync::Arc;
-use std::time::{Duration, Instant};
+use std::time::{Duration, Instant, SystemTime, UNIX_EPOCH};
 
 use anyhow::{Context, Result, anyhow, bail};
-use compute_api::requests::{ComputeClaims, ConfigurationRequest};
-use compute_api::responses::{
-    ComputeConfig, ComputeCtlConfig, ComputeStatus, ComputeStatusResponse, TlsConfig,
-};
+use compute_api::requests::ConfigurationRequest;
+use compute_api::responses::{ComputeCtlConfig, ComputeStatus, ComputeStatusResponse};
 use compute_api::spec::{
     Cluster, ComputeAudit, ComputeFeature, ComputeMode, ComputeSpec, Database, PgIdent,
     RemoteExtSpec, Role,
 };
-use jsonwebtoken::jwk::{
-    AlgorithmParameters, CommonParameters, EllipticCurve, Jwk, JwkSet, KeyAlgorithm, KeyOperations,
-    OctetKeyPairParameters, OctetKeyPairType, PublicKeyUse,
-};
 use nix::sys::signal::{Signal, kill};
 use pageserver_api::shard::ShardStripeSize;
-use pem::Pem;
 use reqwest::header::CONTENT_TYPE;
 use safekeeper_api::membership::SafekeeperGeneration;
 use serde::{Deserialize, Serialize};
-use sha2::{Digest, Sha256};
-use spki::der::Decode;
-use spki::{SubjectPublicKeyInfo, SubjectPublicKeyInfoRef};
 use tracing::debug;
 use url::Host;
 use utils::id::{NodeId, TenantId, TimelineId};
@@ -90,7 +80,6 @@ pub struct EndpointConf {
     drop_subscriptions_before_start: bool,
     features: Vec<ComputeFeature>,
     cluster: Option<Cluster>,
-    compute_ctl_config: ComputeCtlConfig,
 }
 
 //
@@ -146,37 +135,6 @@ impl ComputeControlPlane {
             .unwrap_or(self.base_port)
     }
 
-    /// Create a JSON Web Key Set. This ideally matches the way we create a JWKS
-    /// from the production control plane.
-    fn create_jwks_from_pem(pem: &Pem) -> Result<JwkSet> {
-        let spki: SubjectPublicKeyInfoRef = SubjectPublicKeyInfo::from_der(pem.contents())?;
-        let public_key = spki.subject_public_key.raw_bytes();
-
-        let mut hasher = Sha256::new();
-        hasher.update(public_key);
-        let key_hash = hasher.finalize();
-
-        Ok(JwkSet {
-            keys: vec![Jwk {
-                common: CommonParameters {
-                    public_key_use: Some(PublicKeyUse::Signature),
-                    key_operations: Some(vec![KeyOperations::Verify]),
-                    key_algorithm: Some(KeyAlgorithm::EdDSA),
-                    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>,
-                    x509_sha256_fingerprint: None::<String>,
-                },
-                algorithm: AlgorithmParameters::OctetKeyPair(OctetKeyPairParameters {
-                    key_type: OctetKeyPairType::OctetKeyPair,
-                    curve: EllipticCurve::Ed25519,
-                    x: base64::encode_config(public_key, base64::URL_SAFE_NO_PAD),
-                }),
-            }],
-        })
-    }
-
     #[allow(clippy::too_many_arguments)]
     pub fn new_endpoint(
         &mut self,
@@ -194,10 +152,6 @@ impl ComputeControlPlane {
         let pg_port = pg_port.unwrap_or_else(|| self.get_port());
         let external_http_port = external_http_port.unwrap_or_else(|| self.get_port() + 1);
         let internal_http_port = internal_http_port.unwrap_or_else(|| external_http_port + 1);
-        let compute_ctl_config = ComputeCtlConfig {
-            jwks: Self::create_jwks_from_pem(&self.env.read_public_key()?)?,
-            tls: None::<TlsConfig>,
-        };
         let ep = Arc::new(Endpoint {
             endpoint_id: endpoint_id.to_owned(),
             pg_address: SocketAddr::new(IpAddr::from(Ipv4Addr::LOCALHOST), pg_port),
@@ -225,7 +179,6 @@ impl ComputeControlPlane {
             reconfigure_concurrency: 1,
             features: vec![],
             cluster: None,
-            compute_ctl_config: compute_ctl_config.clone(),
         });
 
         ep.create_endpoint_dir()?;
@@ -245,7 +198,6 @@ impl ComputeControlPlane {
                 reconfigure_concurrency: 1,
                 features: vec![],
                 cluster: None,
-                compute_ctl_config,
             })?,
         )?;
         std::fs::write(
@@ -288,6 +240,7 @@ impl ComputeControlPlane {
 
 ///////////////////////////////////////////////////////////////////////////////
 
+#[derive(Debug)]
 pub struct Endpoint {
     /// used as the directory name
     endpoint_id: String,
@@ -316,9 +269,6 @@ pub struct Endpoint {
     features: Vec<ComputeFeature>,
     // Cluster settings
     cluster: Option<Cluster>,
-
-    /// The compute_ctl config for the endpoint's compute.
-    compute_ctl_config: ComputeCtlConfig,
 }
 
 #[derive(PartialEq, Eq)]
@@ -381,7 +331,6 @@ impl Endpoint {
             drop_subscriptions_before_start: conf.drop_subscriptions_before_start,
             features: conf.features,
             cluster: conf.cluster,
-            compute_ctl_config: conf.compute_ctl_config,
         })
     }
 
@@ -629,13 +578,6 @@ impl Endpoint {
         Ok(safekeeper_connstrings)
     }
 
-    /// Generate a JWT with the correct claims.
-    pub fn generate_jwt(&self) -> Result<String> {
-        self.env.generate_auth_token(&ComputeClaims {
-            compute_id: self.endpoint_id.clone(),
-        })
-    }
-
     #[allow(clippy::too_many_arguments)]
     pub async fn start(
         &self,
@@ -677,97 +619,86 @@ impl Endpoint {
             remote_extensions = None;
         };
 
-        // Create config file
-        let config = {
-            let mut spec = ComputeSpec {
-                skip_pg_catalog_updates: self.skip_pg_catalog_updates,
-                format_version: 1.0,
-                operation_uuid: None,
-                features: self.features.clone(),
-                swap_size_bytes: None,
-                disk_quota_bytes: None,
-                disable_lfc_resizing: None,
-                cluster: Cluster {
-                    cluster_id: None, // project ID: not used
-                    name: None,       // project name: not used
-                    state: None,
-                    roles: if create_test_user {
-                        vec![Role {
-                            name: PgIdent::from_str("test").unwrap(),
-                            encrypted_password: None,
-                            options: None,
-                        }]
-                    } else {
-                        Vec::new()
-                    },
-                    databases: if create_test_user {
-                        vec![Database {
-                            name: PgIdent::from_str("neondb").unwrap(),
-                            owner: PgIdent::from_str("test").unwrap(),
-                            options: None,
-                            restrict_conn: false,
-                            invalid: false,
-                        }]
-                    } else {
-                        Vec::new()
-                    },
-                    settings: None,
-                    postgresql_conf: Some(postgresql_conf.clone()),
-                },
-                delta_operations: None,
-                tenant_id: Some(self.tenant_id),
-                timeline_id: Some(self.timeline_id),
-                project_id: None,
-                branch_id: None,
-                endpoint_id: Some(self.endpoint_id.clone()),
-                mode: self.mode,
-                pageserver_connstring: Some(pageserver_connstring),
-                safekeepers_generation: safekeepers_generation.map(|g| g.into_inner()),
-                safekeeper_connstrings,
-                storage_auth_token: auth_token.clone(),
-                remote_extensions,
-                pgbouncer_settings: None,
-                shard_stripe_size: Some(shard_stripe_size),
-                local_proxy_config: None,
-                reconfigure_concurrency: self.reconfigure_concurrency,
-                drop_subscriptions_before_start: self.drop_subscriptions_before_start,
-                audit_log_level: ComputeAudit::Disabled,
-                logs_export_host: None::<String>,
-            };
-
-            // this strange code is needed to support respec() in tests
-            if self.cluster.is_some() {
-                debug!("Cluster is already set in the endpoint spec, using it");
-                spec.cluster = self.cluster.clone().unwrap();
-
-                debug!("spec.cluster {:?}", spec.cluster);
-
-                // fill missing fields again
-                if create_test_user {
-                    spec.cluster.roles.push(Role {
+        // Create spec file
+        let mut spec = ComputeSpec {
+            skip_pg_catalog_updates: self.skip_pg_catalog_updates,
+            format_version: 1.0,
+            operation_uuid: None,
+            features: self.features.clone(),
+            swap_size_bytes: None,
+            disk_quota_bytes: None,
+            disable_lfc_resizing: None,
+            cluster: Cluster {
+                cluster_id: None, // project ID: not used
+                name: None,       // project name: not used
+                state: None,
+                roles: if create_test_user {
+                    vec![Role {
                         name: PgIdent::from_str("test").unwrap(),
                         encrypted_password: None,
                         options: None,
-                    });
-                    spec.cluster.databases.push(Database {
+                    }]
+                } else {
+                    Vec::new()
+                },
+                databases: if create_test_user {
+                    vec![Database {
                         name: PgIdent::from_str("neondb").unwrap(),
                         owner: PgIdent::from_str("test").unwrap(),
                         options: None,
                         restrict_conn: false,
                         invalid: false,
-                    });
-                }
-                spec.cluster.postgresql_conf = Some(postgresql_conf);
-            }
-
-            ComputeConfig {
-                spec: Some(spec),
-                compute_ctl_config: self.compute_ctl_config.clone(),
-            }
+                    }]
+                } else {
+                    Vec::new()
+                },
+                settings: None,
+                postgresql_conf: Some(postgresql_conf.clone()),
+            },
+            delta_operations: None,
+            tenant_id: Some(self.tenant_id),
+            timeline_id: Some(self.timeline_id),
+            mode: self.mode,
+            pageserver_connstring: Some(pageserver_connstring),
+            safekeepers_generation: safekeepers_generation.map(|g| g.into_inner()),
+            safekeeper_connstrings,
+            storage_auth_token: auth_token.clone(),
+            remote_extensions,
+            pgbouncer_settings: None,
+            shard_stripe_size: Some(shard_stripe_size),
+            local_proxy_config: None,
+            reconfigure_concurrency: self.reconfigure_concurrency,
+            drop_subscriptions_before_start: self.drop_subscriptions_before_start,
+            audit_log_level: ComputeAudit::Disabled,
         };
 
-        let config_path = self.endpoint_path().join("config.json");
-        std::fs::write(config_path, serde_json::to_string_pretty(&config)?)?;
+        // this strange code is needed to support respec() in tests
+        if self.cluster.is_some() {
+            debug!("Cluster is already set in the endpoint spec, using it");
+            spec.cluster = self.cluster.clone().unwrap();
+
+            debug!("spec.cluster {:?}", spec.cluster);
+
+            // fill missing fields again
+            if create_test_user {
+                spec.cluster.roles.push(Role {
+                    name: PgIdent::from_str("test").unwrap(),
+                    encrypted_password: None,
+                    options: None,
+                });
+                spec.cluster.databases.push(Database {
+                    name: PgIdent::from_str("neondb").unwrap(),
+                    owner: PgIdent::from_str("test").unwrap(),
+                    options: None,
+                    restrict_conn: false,
+                    invalid: false,
+                });
+            }
+            spec.cluster.postgresql_conf = Some(postgresql_conf);
+        }
+
+        let spec_path = self.endpoint_path().join("spec.json");
+        std::fs::write(spec_path, serde_json::to_string_pretty(&spec)?)?;
 
         // Open log file. We'll redirect the stdout and stderr of `compute_ctl` to it.
         let logfile = std::fs::OpenOptions::new()
@@ -793,8 +724,10 @@ impl Endpoint {
         ])
         .args(["--pgdata", self.pgdata().to_str().unwrap()])
         .args(["--connstr", &conn_str])
-        .arg("--config")
-        .arg(self.endpoint_path().join("config.json").as_os_str())
+        .args([
+            "--spec-path",
+            self.endpoint_path().join("spec.json").to_str().unwrap(),
+        ])
         .args([
             "--pgbin",
             self.env
@@ -805,7 +738,16 @@ impl Endpoint {
         ])
         // TODO: It would be nice if we generated compute IDs with the same
         // algorithm as the real control plane.
-        .args(["--compute-id", &self.endpoint_id])
+        .args([
+            "--compute-id",
+            &format!(
+                "compute-{}",
+                SystemTime::now()
+                    .duration_since(UNIX_EPOCH)
+                    .unwrap()
+                    .as_secs()
+            ),
+        ])
         .stdin(std::process::Stdio::null())
         .stderr(logfile.try_clone()?)
         .stdout(logfile);
@@ -903,7 +845,6 @@ impl Endpoint {
                     self.external_http_address.port()
                 ),
             )
-            .bearer_auth(self.generate_jwt()?)
             .send()
             .await?;
 
@@ -928,12 +869,10 @@ impl Endpoint {
         stripe_size: Option<ShardStripeSize>,
         safekeepers: Option<Vec<NodeId>>,
     ) -> Result<()> {
-        let (mut spec, compute_ctl_config) = {
-            let config_path = self.endpoint_path().join("config.json");
-            let file = std::fs::File::open(config_path)?;
-            let config: ComputeConfig = serde_json::from_reader(file)?;
-
-            (config.spec.unwrap(), config.compute_ctl_config)
+        let mut spec: ComputeSpec = {
+            let spec_path = self.endpoint_path().join("spec.json");
+            let file = std::fs::File::open(spec_path)?;
+            serde_json::from_reader(file)?
         };
 
         let postgresql_conf = self.read_postgresql_conf()?;
@@ -980,11 +919,10 @@ impl Endpoint {
                 self.external_http_address.port()
             ))
             .header(CONTENT_TYPE.as_str(), "application/json")
-            .bearer_auth(self.generate_jwt()?)
             .body(
                 serde_json::to_string(&ConfigurationRequest {
                     spec,
-                    compute_ctl_config,
+                    compute_ctl_config: ComputeCtlConfig::default(),
                 })
                 .unwrap(),
             )

control_plane/src/lib.rs

@@ -10,7 +10,6 @@ mod background_process;
 pub mod broker;
 pub mod endpoint;
 pub mod local_env;
-pub mod object_storage;
 pub mod pageserver;
 pub mod postgresql_conf;
 pub mod safekeeper;

control_plane/src/local_env.rs

@@ -12,18 +12,16 @@ use std::{env, fs};
 
 use anyhow::{Context, bail};
 use clap::ValueEnum;
-use pem::Pem;
 use postgres_backend::AuthType;
 use reqwest::Url;
 use serde::{Deserialize, Serialize};
-use utils::auth::encode_from_key_file;
+use utils::auth::{Claims, encode_from_key_file};
 use utils::id::{NodeId, TenantId, TenantTimelineId, TimelineId};
 
-use crate::object_storage::{OBJECT_STORAGE_REMOTE_STORAGE_DIR, ObjectStorage};
 use crate::pageserver::{PAGESERVER_REMOTE_STORAGE_DIR, PageServerNode};
 use crate::safekeeper::SafekeeperNode;
 
-pub const DEFAULT_PG_VERSION: u32 = 17;
+pub const DEFAULT_PG_VERSION: u32 = 16;
 
 //
 // This data structures represents neon_local CLI config
@@ -57,8 +55,6 @@ pub struct LocalEnv {
 
     // used to issue tokens during e.g pg start
     pub private_key_path: PathBuf,
-    /// Path to environment's public key
-    pub public_key_path: PathBuf,
 
     pub broker: NeonBroker,
 
@@ -72,8 +68,6 @@ pub struct LocalEnv {
 
     pub safekeepers: Vec<SafekeeperConf>,
 
-    pub object_storage: ObjectStorageConf,
-
     // Control plane upcall API for pageserver: if None, we will not run storage_controller  If set, this will
     // be propagated into each pageserver's configuration.
     pub control_plane_api: Url,
@@ -101,7 +95,6 @@ pub struct OnDiskConfig {
     pub neon_distrib_dir: PathBuf,
     pub default_tenant_id: Option<TenantId>,
     pub private_key_path: PathBuf,
-    pub public_key_path: PathBuf,
     pub broker: NeonBroker,
     pub storage_controller: NeonStorageControllerConf,
     #[serde(
@@ -110,7 +103,6 @@ pub struct OnDiskConfig {
     )]
     pub pageservers: Vec<PageServerConf>,
     pub safekeepers: Vec<SafekeeperConf>,
-    pub object_storage: ObjectStorageConf,
     pub control_plane_api: Option<Url>,
     pub control_plane_hooks_api: Option<Url>,
     pub control_plane_compute_hook_api: Option<Url>,
@@ -144,18 +136,11 @@ pub struct NeonLocalInitConf {
     pub storage_controller: Option<NeonStorageControllerConf>,
     pub pageservers: Vec<NeonLocalInitPageserverConf>,
     pub safekeepers: Vec<SafekeeperConf>,
-    pub object_storage: ObjectStorageConf,
     pub control_plane_api: Option<Url>,
     pub control_plane_hooks_api: Option<Url>,
     pub generate_local_ssl_certs: bool,
 }
 
-#[derive(Serialize, Default, Deserialize, PartialEq, Eq, Clone, Debug)]
-#[serde(default)]
-pub struct ObjectStorageConf {
-    pub port: u16,
-}
-
 /// Broker config for cluster internal communication.
 #[derive(Serialize, Deserialize, PartialEq, Eq, Clone, Debug)]
 #[serde(default)]
@@ -413,10 +398,6 @@ impl LocalEnv {
         self.pg_dir(pg_version, "lib")
     }
 
-    pub fn object_storage_bin(&self) -> PathBuf {
-        self.neon_distrib_dir.join("object_storage")
-    }
-
     pub fn pageserver_bin(&self) -> PathBuf {
         self.neon_distrib_dir.join("pageserver")
     }
@@ -450,10 +431,6 @@ impl LocalEnv {
         self.base_data_dir.join("safekeepers").join(data_dir_name)
     }
 
-    pub fn object_storage_data_dir(&self) -> PathBuf {
-        self.base_data_dir.join("object_storage")
-    }
-
     pub fn get_pageserver_conf(&self, id: NodeId) -> anyhow::Result<&PageServerConf> {
         if let Some(conf) = self.pageservers.iter().find(|node| node.id == id) {
             Ok(conf)
@@ -605,7 +582,6 @@ impl LocalEnv {
                 neon_distrib_dir,
                 default_tenant_id,
                 private_key_path,
-                public_key_path,
                 broker,
                 storage_controller,
                 pageservers,
@@ -615,7 +591,6 @@ impl LocalEnv {
                 control_plane_compute_hook_api: _,
                 branch_name_mappings,
                 generate_local_ssl_certs,
-                object_storage,
             } = on_disk_config;
             LocalEnv {
                 base_data_dir: repopath.to_owned(),
@@ -623,7 +598,6 @@ impl LocalEnv {
                 neon_distrib_dir,
                 default_tenant_id,
                 private_key_path,
-                public_key_path,
                 broker,
                 storage_controller,
                 pageservers,
@@ -632,7 +606,6 @@ impl LocalEnv {
                 control_plane_hooks_api,
                 branch_name_mappings,
                 generate_local_ssl_certs,
-                object_storage,
             }
         };
 
@@ -732,7 +705,6 @@ impl LocalEnv {
                 neon_distrib_dir: self.neon_distrib_dir.clone(),
                 default_tenant_id: self.default_tenant_id,
                 private_key_path: self.private_key_path.clone(),
-                public_key_path: self.public_key_path.clone(),
                 broker: self.broker.clone(),
                 storage_controller: self.storage_controller.clone(),
                 pageservers: vec![], // it's skip_serializing anyway
@@ -742,7 +714,6 @@ impl LocalEnv {
                 control_plane_compute_hook_api: None,
                 branch_name_mappings: self.branch_name_mappings.clone(),
                 generate_local_ssl_certs: self.generate_local_ssl_certs,
-                object_storage: self.object_storage.clone(),
             },
         )
     }
@@ -759,12 +730,12 @@ impl LocalEnv {
     }
 
     // this function is used only for testing purposes in CLI e g generate tokens during init
-    pub fn generate_auth_token<S: Serialize>(&self, claims: &S) -> anyhow::Result<String> {
-        let key = self.read_private_key()?;
-        encode_from_key_file(claims, &key)
+    pub fn generate_auth_token(&self, claims: &Claims) -> anyhow::Result<String> {
+        let private_key_path = self.get_private_key_path();
+        let key_data = fs::read(private_key_path)?;
+        encode_from_key_file(claims, &key_data)
     }
 
-    /// Get the path to the private key.
     pub fn get_private_key_path(&self) -> PathBuf {
         if self.private_key_path.is_absolute() {
             self.private_key_path.to_path_buf()
@@ -773,29 +744,6 @@ impl LocalEnv {
         }
     }
 
-    /// Get the path to the public key.
-    pub fn get_public_key_path(&self) -> PathBuf {
-        if self.public_key_path.is_absolute() {
-            self.public_key_path.to_path_buf()
-        } else {
-            self.base_data_dir.join(&self.public_key_path)
-        }
-    }
-
-    /// Read the contents of the private key file.
-    pub fn read_private_key(&self) -> anyhow::Result<Pem> {
-        let private_key_path = self.get_private_key_path();
-        let pem = pem::parse(fs::read(private_key_path)?)?;
-        Ok(pem)
-    }
-
-    /// Read the contents of the public key file.
-    pub fn read_public_key(&self) -> anyhow::Result<Pem> {
-        let public_key_path = self.get_public_key_path();
-        let pem = pem::parse(fs::read(public_key_path)?)?;
-        Ok(pem)
-    }
-
     /// Materialize the [`NeonLocalInitConf`] to disk. Called during [`neon_local init`].
     pub fn init(conf: NeonLocalInitConf, force: &InitForceMode) -> anyhow::Result<()> {
         let base_path = base_path();
@@ -849,7 +797,6 @@ impl LocalEnv {
             control_plane_api,
             generate_local_ssl_certs,
             control_plane_hooks_api,
-            object_storage,
         } = conf;
 
         // Find postgres binaries.
@@ -881,7 +828,6 @@ impl LocalEnv {
         )
         .context("generate auth keys")?;
         let private_key_path = PathBuf::from("auth_private_key.pem");
-        let public_key_path = PathBuf::from("auth_public_key.pem");
 
         // create the runtime type because the remaining initialization code below needs
         // a LocalEnv instance op operation
@@ -892,7 +838,6 @@ impl LocalEnv {
             neon_distrib_dir,
             default_tenant_id: Some(default_tenant_id),
             private_key_path,
-            public_key_path,
             broker,
             storage_controller: storage_controller.unwrap_or_default(),
             pageservers: pageservers.iter().map(Into::into).collect(),
@@ -901,7 +846,6 @@ impl LocalEnv {
             control_plane_hooks_api,
             branch_name_mappings: Default::default(),
             generate_local_ssl_certs,
-            object_storage,
         };
 
         if generate_local_ssl_certs {
@@ -929,13 +873,8 @@ impl LocalEnv {
                 .context("pageserver init failed")?;
         }
 
-        ObjectStorage::from_env(&env)
-            .init()
-            .context("object storage init failed")?;
-
         // setup remote remote location for default LocalFs remote storage
         std::fs::create_dir_all(env.base_data_dir.join(PAGESERVER_REMOTE_STORAGE_DIR))?;
-        std::fs::create_dir_all(env.base_data_dir.join(OBJECT_STORAGE_REMOTE_STORAGE_DIR))?;
 
         env.persist_config()
     }
@@ -981,7 +920,6 @@ fn generate_auth_keys(private_key_path: &Path, public_key_path: &Path) -> anyhow
             String::from_utf8_lossy(&keygen_output.stderr)
         );
     }
-
     // Extract the public key from the private key file
     //
     // openssl pkey -in auth_private_key.pem -pubout -out auth_public_key.pem
@@ -998,7 +936,6 @@ fn generate_auth_keys(private_key_path: &Path, public_key_path: &Path) -> anyhow
             String::from_utf8_lossy(&keygen_output.stderr)
         );
     }
-
     Ok(())
 }
 
@@ -1007,7 +944,7 @@ fn generate_ssl_ca_cert(cert_path: &Path, key_path: &Path) -> anyhow::Result<()>
     // -out rootCA.crt -keyout rootCA.key
     let keygen_output = Command::new("openssl")
         .args([
-            "req", "-x509", "-newkey", "ed25519", "-nodes", "-days", "36500",
+            "req", "-x509", "-newkey", "rsa:2048", "-nodes", "-days", "36500",
         ])
         .args(["-subj", "/CN=Neon Local CA"])
         .args(["-out", cert_path.to_str().unwrap()])
@@ -1037,7 +974,7 @@ fn generate_ssl_cert(
     // -subj "/CN=localhost" -addext "subjectAltName=DNS:localhost,IP:127.0.0.1"
     let keygen_output = Command::new("openssl")
         .args(["req", "-new", "-nodes"])
-        .args(["-newkey", "ed25519"])
+        .args(["-newkey", "rsa:2048"])
         .args(["-subj", "/CN=localhost"])
         .args(["-addext", "subjectAltName=DNS:localhost,IP:127.0.0.1"])
         .args(["-keyout", key_path.to_str().unwrap()])

control_plane/src/object_storage.rs

@@ -1,107 +0,0 @@
-use crate::background_process::{self, start_process, stop_process};
-use crate::local_env::LocalEnv;
-use anyhow::anyhow;
-use anyhow::{Context, Result};
-use camino::Utf8PathBuf;
-use std::io::Write;
-use std::time::Duration;
-
-/// Directory within .neon which will be used by default for LocalFs remote storage.
-pub const OBJECT_STORAGE_REMOTE_STORAGE_DIR: &str = "local_fs_remote_storage/object_storage";
-pub const OBJECT_STORAGE_DEFAULT_PORT: u16 = 9993;
-
-pub struct ObjectStorage {
-    pub bin: Utf8PathBuf,
-    pub data_dir: Utf8PathBuf,
-    pub pemfile: Utf8PathBuf,
-    pub port: u16,
-}
-
-impl ObjectStorage {
-    pub fn from_env(env: &LocalEnv) -> ObjectStorage {
-        ObjectStorage {
-            bin: Utf8PathBuf::from_path_buf(env.object_storage_bin()).unwrap(),
-            data_dir: Utf8PathBuf::from_path_buf(env.object_storage_data_dir()).unwrap(),
-            pemfile: Utf8PathBuf::from_path_buf(env.public_key_path.clone()).unwrap(),
-            port: env.object_storage.port,
-        }
-    }
-
-    fn config_path(&self) -> Utf8PathBuf {
-        self.data_dir.join("object_storage.json")
-    }
-
-    fn listen_addr(&self) -> Utf8PathBuf {
-        format!("127.0.0.1:{}", self.port).into()
-    }
-
-    pub fn init(&self) -> Result<()> {
-        println!("Initializing object storage in {:?}", self.data_dir);
-        let parent = self.data_dir.parent().unwrap();
-
-        #[derive(serde::Serialize)]
-        struct Cfg {
-            listen: Utf8PathBuf,
-            pemfile: Utf8PathBuf,
-            local_path: Utf8PathBuf,
-            r#type: String,
-        }
-        let cfg = Cfg {
-            listen: self.listen_addr(),
-            pemfile: parent.join(self.pemfile.clone()),
-            local_path: parent.join(OBJECT_STORAGE_REMOTE_STORAGE_DIR),
-            r#type: "LocalFs".to_string(),
-        };
-        std::fs::create_dir_all(self.config_path().parent().unwrap())?;
-        std::fs::write(self.config_path(), serde_json::to_string(&cfg)?)
-            .context("write object storage config")?;
-        Ok(())
-    }
-
-    pub async fn start(&self, retry_timeout: &Duration) -> Result<()> {
-        println!("Starting s3 proxy at {}", self.listen_addr());
-        std::io::stdout().flush().context("flush stdout")?;
-
-        let process_status_check = || async {
-            tokio::time::sleep(Duration::from_millis(500)).await;
-            let res = reqwest::Client::new()
-                .get(format!("http://{}/metrics", self.listen_addr()))
-                .send()
-                .await;
-            match res {
-                Ok(response) if response.status().is_success() => Ok(true),
-                Ok(_) => Err(anyhow!("Failed to query /metrics")),
-                Err(e) => Err(anyhow!("Failed to check node status: {e}")),
-            }
-        };
-
-        let res = start_process(
-            "object_storage",
-            &self.data_dir.clone().into_std_path_buf(),
-            &self.bin.clone().into_std_path_buf(),
-            vec![self.config_path().to_string()],
-            vec![("RUST_LOG".into(), "debug".into())],
-            background_process::InitialPidFile::Create(self.pid_file()),
-            retry_timeout,
-            process_status_check,
-        )
-        .await;
-        if res.is_err() {
-            eprintln!("Logs:\n{}", std::fs::read_to_string(self.log_file())?);
-        }
-
-        res
-    }
-
-    pub fn stop(&self, immediate: bool) -> anyhow::Result<()> {
-        stop_process(immediate, "object_storage", &self.pid_file())
-    }
-
-    fn log_file(&self) -> Utf8PathBuf {
-        self.data_dir.join("object_storage.log")
-    }
-
-    fn pid_file(&self) -> Utf8PathBuf {
-        self.data_dir.join("object_storage.pid")
-    }
-}

control_plane/src/pageserver.rs

@@ -413,11 +413,6 @@ impl PageServerNode {
                 .map(serde_json::from_str)
                 .transpose()
                 .context("Failed to parse 'compaction_algorithm' json")?,
-            compaction_shard_ancestor: settings
-                .remove("compaction_shard_ancestor")
-                .map(|x| x.parse::<bool>())
-                .transpose()
-                .context("Failed to parse 'compaction_shard_ancestor' as a bool")?,
             compaction_l0_first: settings
                 .remove("compaction_l0_first")
                 .map(|x| x.parse::<bool>())
@@ -540,11 +535,6 @@ impl PageServerNode {
                 .map(|x| x.parse::<bool>())
                 .transpose()
                 .context("Failed to parse 'gc_compaction_enabled' as bool")?,
-            gc_compaction_verification: settings
-                .remove("gc_compaction_verification")
-                .map(|x| x.parse::<bool>())
-                .transpose()
-                .context("Failed to parse 'gc_compaction_verification' as bool")?,
             gc_compaction_initial_threshold_kb: settings
                 .remove("gc_compaction_initial_threshold_kb")
                 .map(|x| x.parse::<u64>())

control_plane/src/storage_controller.rs

@@ -13,12 +13,9 @@ use pageserver_api::controller_api::{
     NodeConfigureRequest, NodeDescribeResponse, NodeRegisterRequest, TenantCreateRequest,
     TenantCreateResponse, TenantLocateResponse,
 };
-use pageserver_api::models::{
-    TenantConfig, TenantConfigRequest, TimelineCreateRequest, TimelineInfo,
-};
+use pageserver_api::models::{TenantConfigRequest, TimelineCreateRequest, TimelineInfo};
 use pageserver_api::shard::TenantShardId;
 use pageserver_client::mgmt_api::ResponseErrorMessageExt;
-use pem::Pem;
 use postgres_backend::AuthType;
 use reqwest::{Certificate, Method};
 use serde::de::DeserializeOwned;
@@ -35,8 +32,8 @@ use crate::local_env::{LocalEnv, NeonStorageControllerConf};
 
 pub struct StorageController {
     env: LocalEnv,
-    private_key: Option<Pem>,
-    public_key: Option<Pem>,
+    private_key: Option<Vec<u8>>,
+    public_key: Option<String>,
     client: reqwest::Client,
     config: NeonStorageControllerConf,
 
@@ -85,8 +82,7 @@ impl NeonStorageControllerStopArgs {
 pub struct AttachHookRequest {
     pub tenant_shard_id: TenantShardId,
     pub node_id: Option<NodeId>,
-    pub generation_override: Option<i32>, // only new tenants
-    pub config: Option<TenantConfig>,     // only new tenants
+    pub generation_override: Option<i32>,
 }
 
 #[derive(Serialize, Deserialize)]
@@ -117,9 +113,7 @@ impl StorageController {
             AuthType::Trust => (None, None),
             AuthType::NeonJWT => {
                 let private_key_path = env.get_private_key_path();
-                let private_key =
-                    pem::parse(fs::read(private_key_path).expect("failed to read private key"))
-                        .expect("failed to parse PEM file");
+                let private_key = fs::read(private_key_path).expect("failed to read private key");
 
                 // If pageserver auth is enabled, this implicitly enables auth for this service,
                 // using the same credentials.
@@ -141,13 +135,9 @@ impl StorageController {
                         .expect("Empty key dir")
                         .expect("Error reading key dir");
 
-                    pem::parse(std::fs::read_to_string(dent.path()).expect("Can't read public key"))
-                        .expect("Failed to parse PEM file")
+                    std::fs::read_to_string(dent.path()).expect("Can't read public key")
                 } else {
-                    pem::parse(
-                        std::fs::read_to_string(&public_key_path).expect("Can't read public key"),
-                    )
-                    .expect("Failed to parse PEM file")
+                    std::fs::read_to_string(&public_key_path).expect("Can't read public key")
                 };
                 (Some(private_key), Some(public_key))
             }
@@ -815,7 +805,6 @@ impl StorageController {
             tenant_shard_id,
             node_id: Some(pageserver_id),
             generation_override: None,
-            config: None,
         };
 
         let response = self

control_plane/storcon_cli/src/main.rs

@@ -941,7 +941,7 @@ async fn main() -> anyhow::Result<()> {
             let mut node_to_fill_descs = Vec::new();
 
             for desc in node_descs {
-                let to_drain = nodes.contains(&desc.id);
+                let to_drain = nodes.iter().any(|id| *id == desc.id);
                 if to_drain {
                     node_to_drain_descs.push(desc);
                 } else {

docker-compose/README.md

@@ -1,3 +1,4 @@
+
 # Example docker compose configuration
 
 The configuration in this directory is used for testing Neon docker images: it is
@@ -7,13 +8,3 @@ you can experiment with a miniature Neon system, use `cargo neon` rather than co
 This configuration does not start the storage controller, because the controller
 needs a way to reconfigure running computes, and no such thing exists in this setup.
 
-## Generating the JWKS for a compute
-
-```shell
-openssl genpkey -algorithm Ed25519 -out private-key.pem
-openssl pkey -in private-key.pem -pubout -out public-key.pem
-openssl pkey -pubin -inform pem -in public-key.pem -pubout -outform der -out public-key.der
-key="$(xxd -plain -cols 32 -s -32 public-key.der)"
-key_id="$(printf '%s' "$key" | sha256sum | awk '{ print $1 }' | basenc --base64url --wrap=0)"
-x="$(printf '%s' "$key" | basenc --base64url --wrap=0)"
-```

docker-compose/compute_wrapper/private-key.pem

@@ -1,3 +0,0 @@
------BEGIN PRIVATE KEY-----
-MC4CAQAwBQYDK2VwBCIEIOmnRbzt2AJ0d+S3aU1hiYOl/tXpvz1FmWBfwHYBgOma
------END PRIVATE KEY-----

docker-compose/compute_wrapper/public-key.pem

@@ -1,3 +0,0 @@
------BEGIN PUBLIC KEY-----
-MCowBQYDK2VwAyEADY0al/U0bgB3+9fUGk+3PKWnsck9OyxN5DjHIN6Xep0=
------END PUBLIC KEY-----

docker-compose/compute_wrapper/shell/compute.sh

@@ -11,8 +11,8 @@ generate_id() {
 
 PG_VERSION=${PG_VERSION:-14}
 
-CONFIG_FILE_ORG=/var/db/postgres/configs/config.json
-CONFIG_FILE=/tmp/config.json
+SPEC_FILE_ORG=/var/db/postgres/specs/spec.json
+SPEC_FILE=/tmp/spec.json
 
 echo "Waiting pageserver become ready."
 while ! nc -z pageserver 6400; do
@@ -20,7 +20,7 @@ while ! nc -z pageserver 6400; do
 done
 echo "Page server is ready."
 
-cp ${CONFIG_FILE_ORG} ${CONFIG_FILE}
+cp ${SPEC_FILE_ORG} ${SPEC_FILE}
 
  if [ -n "${TENANT_ID:-}" ] && [ -n "${TIMELINE_ID:-}" ]; then
    tenant_id=${TENANT_ID}
@@ -73,17 +73,17 @@ else
   ulid_extension=ulid
 fi
 echo "Adding pgx_ulid"
-shared_libraries=$(jq -r '.spec.cluster.settings[] | select(.name=="shared_preload_libraries").value' ${CONFIG_FILE})
-sed -i "s/${shared_libraries}/${shared_libraries},${ulid_extension}/" ${CONFIG_FILE}
+shared_libraries=$(jq -r '.cluster.settings[] | select(.name=="shared_preload_libraries").value' ${SPEC_FILE})
+sed -i "s/${shared_libraries}/${shared_libraries},${ulid_extension}/" ${SPEC_FILE}
 echo "Overwrite tenant id and timeline id in spec file"
-sed -i "s/TENANT_ID/${tenant_id}/" ${CONFIG_FILE}
-sed -i "s/TIMELINE_ID/${timeline_id}/" ${CONFIG_FILE}
+sed -i "s/TENANT_ID/${tenant_id}/" ${SPEC_FILE}
+sed -i "s/TIMELINE_ID/${timeline_id}/" ${SPEC_FILE}
 
-cat ${CONFIG_FILE}
+cat ${SPEC_FILE}
 
 echo "Start compute node"
 /usr/local/bin/compute_ctl --pgdata /var/db/postgres/compute \
      -C "postgresql://cloud_admin@localhost:55433/postgres"  \
      -b /usr/local/bin/postgres                              \
      --compute-id "compute-$RANDOM"                          \
-     --config "$CONFIG_FILE"
+     -S ${SPEC_FILE}

docker-compose/compute_wrapper/var/db/postgres/configs/config.json

@@ -1,160 +0,0 @@
-{
-    "spec": {
-        "format_version": 1.0,
-
-        "timestamp": "2022-10-12T18:00:00.000Z",
-        "operation_uuid": "0f657b36-4b0f-4a2d-9c2e-1dcd615e7d8c",
-
-        "cluster": {
-            "cluster_id": "docker_compose",
-            "name": "docker_compose_test",
-            "state": "restarted",
-            "roles": [
-                {
-                    "name": "cloud_admin",
-                    "encrypted_password": "b093c0d3b281ba6da1eacc608620abd8",
-                    "options": null
-                }
-            ],
-            "databases": [
-            ],
-            "settings": [
-                {
-                    "name": "fsync",
-                    "value": "off",
-                    "vartype": "bool"
-                },
-                {
-                    "name": "wal_level",
-                    "value": "logical",
-                    "vartype": "enum"
-                },
-                {
-                    "name": "wal_log_hints",
-                    "value": "on",
-                    "vartype": "bool"
-                },
-                {
-                    "name": "log_connections",
-                    "value": "on",
-                    "vartype": "bool"
-                },
-                {
-                    "name": "port",
-                    "value": "55433",
-                    "vartype": "integer"
-                },
-                {
-                    "name": "shared_buffers",
-                    "value": "1MB",
-                    "vartype": "string"
-                },
-                {
-                    "name": "max_connections",
-                    "value": "100",
-                    "vartype": "integer"
-                },
-                {
-                    "name": "listen_addresses",
-                    "value": "0.0.0.0",
-                    "vartype": "string"
-                },
-                {
-                    "name": "max_wal_senders",
-                    "value": "10",
-                    "vartype": "integer"
-                },
-                {
-                    "name": "max_replication_slots",
-                    "value": "10",
-                    "vartype": "integer"
-                },
-                {
-                    "name": "wal_sender_timeout",
-                    "value": "5s",
-                    "vartype": "string"
-                },
-                {
-                    "name": "wal_keep_size",
-                    "value": "0",
-                    "vartype": "integer"
-                },
-                {
-                    "name": "password_encryption",
-                    "value": "md5",
-                    "vartype": "enum"
-                },
-                {
-                    "name": "restart_after_crash",
-                    "value": "off",
-                    "vartype": "bool"
-                },
-                {
-                    "name": "synchronous_standby_names",
-                    "value": "walproposer",
-                    "vartype": "string"
-                },
-                {
-                    "name": "shared_preload_libraries",
-                    "value": "neon,pg_cron,timescaledb,pg_stat_statements",
-                    "vartype": "string"
-                },
-                {
-                    "name": "neon.safekeepers",
-                    "value": "safekeeper1:5454,safekeeper2:5454,safekeeper3:5454",
-                    "vartype": "string"
-                },
-                {
-                    "name": "neon.timeline_id",
-                    "value": "TIMELINE_ID",
-                    "vartype": "string"
-                },
-                {
-                    "name": "neon.tenant_id",
-                    "value": "TENANT_ID",
-                    "vartype": "string"
-                },
-                {
-                    "name": "neon.pageserver_connstring",
-                    "value": "host=pageserver port=6400",
-                    "vartype": "string"
-                },
-                {
-                    "name": "max_replication_write_lag",
-                    "value": "500MB",
-                    "vartype": "string"
-                },
-                {
-                    "name": "max_replication_flush_lag",
-                    "value": "10GB",
-                    "vartype": "string"
-                },
-                {
-                    "name": "cron.database",
-                    "value": "postgres",
-                    "vartype": "string"
-                }
-            ]
-        },
-
-        "delta_operations": [
-        ]
-    },
-    "compute_ctl_config": {
-        "jwks": {
-            "keys": [
-                {
-                    "use": "sig",
-                    "key_ops": [
-                        "verify"
-                    ],
-                    "alg": "EdDSA",
-                    "kid": "ZGIxMzAzOGY0YWQwODk2ODU1MTk1NzMxMDFkYmUyOWU2NzZkOWNjNjMyMGRkZGJjOWY0MjdjYWVmNzE1MjUyOAo=",
-                    "kty": "OKP",
-                    "crv": "Ed25519",
-                    "x": "MGQ4ZDFhOTdmNTM0NmUwMDc3ZmJkN2Q0MWE0ZmI3M2NhNWE3YjFjOTNkM2IyYzRkZTQzOGM3MjBkZTk3N2E5ZAo="
-                }
-            ]
-        }
-    }
-}

docker-compose/compute_wrapper/var/db/postgres/specs/spec.json

@@ -0,0 +1,141 @@
+{
+    "format_version": 1.0,
+
+    "timestamp": "2022-10-12T18:00:00.000Z",
+    "operation_uuid": "0f657b36-4b0f-4a2d-9c2e-1dcd615e7d8c",
+
+    "cluster": {
+        "cluster_id": "docker_compose",
+        "name": "docker_compose_test",
+        "state": "restarted",
+        "roles": [
+            {
+                "name": "cloud_admin",
+                "encrypted_password": "b093c0d3b281ba6da1eacc608620abd8",
+                "options": null
+            }
+        ],
+        "databases": [
+        ],
+        "settings": [
+            {
+                "name": "fsync",
+                "value": "off",
+                "vartype": "bool"
+            },
+            {
+                "name": "wal_level",
+                "value": "logical",
+                "vartype": "enum"
+            },
+            {
+                "name": "wal_log_hints",
+                "value": "on",
+                "vartype": "bool"
+            },
+            {
+                "name": "log_connections",
+                "value": "on",
+                "vartype": "bool"
+            },
+            {
+                "name": "port",
+                "value": "55433",
+                "vartype": "integer"
+            },
+            {
+                "name": "shared_buffers",
+                "value": "1MB",
+                "vartype": "string"
+            },
+            {
+                "name": "max_connections",
+                "value": "100",
+                "vartype": "integer"
+            },
+            {
+                "name": "listen_addresses",
+                "value": "0.0.0.0",
+                "vartype": "string"
+            },
+            {
+                "name": "max_wal_senders",
+                "value": "10",
+                "vartype": "integer"
+            },
+            {
+                "name": "max_replication_slots",
+                "value": "10",
+                "vartype": "integer"
+            },
+            {
+                "name": "wal_sender_timeout",
+                "value": "5s",
+                "vartype": "string"
+            },
+            {
+                "name": "wal_keep_size",
+                "value": "0",
+                "vartype": "integer"
+            },
+            {
+                "name": "password_encryption",
+                "value": "md5",
+                "vartype": "enum"
+            },
+            {
+                "name": "restart_after_crash",
+                "value": "off",
+                "vartype": "bool"
+            },
+            {
+                "name": "synchronous_standby_names",
+                "value": "walproposer",
+                "vartype": "string"
+            },
+            {
+                "name": "shared_preload_libraries",
+                "value": "neon,pg_cron,timescaledb,pg_stat_statements",
+                "vartype": "string"
+            },
+            {
+                "name": "neon.safekeepers",
+                "value": "safekeeper1:5454,safekeeper2:5454,safekeeper3:5454",
+                "vartype": "string"
+            },
+            {
+                "name": "neon.timeline_id",
+                "value": "TIMELINE_ID",
+                "vartype": "string"
+            },
+            {
+                "name": "neon.tenant_id",
+                "value": "TENANT_ID",
+                "vartype": "string"
+            },
+            {
+                "name": "neon.pageserver_connstring",
+                "value": "host=pageserver port=6400",
+                "vartype": "string"
+            },
+            {
+                "name": "max_replication_write_lag",
+                "value": "500MB",
+                "vartype": "string"
+            },
+            {
+                "name": "max_replication_flush_lag",
+                "value": "10GB",
+                "vartype": "string"
+            },
+            {
+                "name": "cron.database",
+                "value": "postgres",
+                "vartype": "string"
+            }
+        ]
+    },
+
+    "delta_operations": [
+    ]
+}

docker-compose/docker-compose.yml

@@ -159,7 +159,7 @@ services:
       #- RUST_BACKTRACE=1
     # Mount the test files directly, for faster editing cycle.
     volumes:
-      - ./compute_wrapper/var/db/postgres/configs/:/var/db/postgres/configs/
+      - ./compute_wrapper/var/db/postgres/specs/:/var/db/postgres/specs/
       - ./compute_wrapper/shell/:/shell/
     ports:
       - 55433:55433 # pg protocol handler

docker-compose/ext-src/pg_jsonschema-src/Makefile

@@ -1,8 +0,0 @@
-EXTENSION = pg_jsonschema
-DATA = pg_jsonschema--1.0.sql
-REGRESS = jsonschema_valid_api  jsonschema_edge_cases
-REGRESS_OPTS = --load-extension=pg_jsonschema
-
-PG_CONFIG ?= pg_config
-PGXS := $(shell $(PG_CONFIG) --pgxs)
-include $(PGXS)

docker-compose/ext-src/pg_jsonschema-src/expected/jsonschema_edge_cases.out

@@ -1,87 +0,0 @@
--- Schema with enums, nulls, extra properties disallowed
-SELECT jsonschema_is_valid('{
-  "type": "object",
-  "properties": {
-    "status": { "type": "string", "enum": ["active", "inactive", "pending"] },
-    "email": { "type": ["string", "null"], "format": "email" }
-  },
-  "required": ["status"],
-  "additionalProperties": false
-}'::json);
- jsonschema_is_valid 
----------------------
- t
-(1 row)
-
--- Valid enum and null email
-SELECT jsonschema_validation_errors(
-  '{
-    "type": "object",
-    "properties": {
-      "status": { "type": "string", "enum": ["active", "inactive", "pending"] },
-      "email": { "type": ["string", "null"], "format": "email" }
-    },
-    "required": ["status"],
-    "additionalProperties": false
-  }'::json,
-  '{"status": "active", "email": null}'::json
-);
- jsonschema_validation_errors 
-------------------------------
- {}
-(1 row)
-
--- Invalid enum value
-SELECT jsonschema_validation_errors(
-  '{
-    "type": "object",
-    "properties": {
-      "status": { "type": "string", "enum": ["active", "inactive", "pending"] },
-      "email": { "type": ["string", "null"], "format": "email" }
-    },
-    "required": ["status"],
-    "additionalProperties": false
-  }'::json,
-  '{"status": "disabled", "email": null}'::json
-);
-                     jsonschema_validation_errors                     
-----------------------------------------------------------------------
- {"\"disabled\" is not one of [\"active\",\"inactive\",\"pending\"]"}
-(1 row)
-
--- Invalid email format (assuming format is validated)
-SELECT jsonschema_validation_errors(
-  '{
-    "type": "object",
-    "properties": {
-      "status": { "type": "string", "enum": ["active", "inactive", "pending"] },
-      "email": { "type": ["string", "null"], "format": "email" }
-    },
-    "required": ["status"],
-    "additionalProperties": false
-  }'::json,
-  '{"status": "active", "email": "not-an-email"}'::json
-);
-      jsonschema_validation_errors       
------------------------------------------
- {"\"not-an-email\" is not a \"email\""}
-(1 row)
-
--- Extra property not allowed
-SELECT jsonschema_validation_errors(
-  '{
-    "type": "object",
-    "properties": {
-      "status": { "type": "string", "enum": ["active", "inactive", "pending"] },
-      "email": { "type": ["string", "null"], "format": "email" }
-    },
-    "required": ["status"],
-    "additionalProperties": false
-  }'::json,
-  '{"status": "active", "extra": "should not be here"}'::json
-);
-                    jsonschema_validation_errors                    
---------------------------------------------------------------------
- {"Additional properties are not allowed ('extra' was unexpected)"}
-(1 row)
-

docker-compose/ext-src/pg_jsonschema-src/expected/jsonschema_valid_api.out

@@ -1,65 +0,0 @@
--- Define schema
-SELECT jsonschema_is_valid('{
-  "type": "object",
-  "properties": {
-    "username": { "type": "string" },
-    "age": { "type": "integer" }
-  },
-  "required": ["username"]
-}'::json);
- jsonschema_is_valid 
----------------------
- t
-(1 row)
-
--- Valid instance
-SELECT jsonschema_validation_errors(
-  '{
-    "type": "object",
-    "properties": {
-      "username": { "type": "string" },
-      "age": { "type": "integer" }
-    },
-    "required": ["username"]
-  }'::json,
-  '{"username": "alice", "age": 25}'::json
-);
- jsonschema_validation_errors 
-------------------------------
- {}
-(1 row)
-
--- Invalid instance: missing required "username"
-SELECT jsonschema_validation_errors(
-  '{
-    "type": "object",
-    "properties": {
-      "username": { "type": "string" },
-      "age": { "type": "integer" }
-    },
-    "required": ["username"]
-  }'::json,
-  '{"age": 25}'::json
-);
-      jsonschema_validation_errors       
------------------------------------------
- {"\"username\" is a required property"}
-(1 row)
-
--- Invalid instance: wrong type for "age"
-SELECT jsonschema_validation_errors(
-  '{
-    "type": "object",
-    "properties": {
-      "username": { "type": "string" },
-      "age": { "type": "integer" }
-    },
-    "required": ["username"]
-  }'::json,
-  '{"username": "bob", "age": "twenty"}'::json
-);
-       jsonschema_validation_errors        
--------------------------------------------
- {"\"twenty\" is not of type \"integer\""}
-(1 row)
-

docker-compose/ext-src/pg_jsonschema-src/sql/jsonschema_edge_cases.sql

@@ -1,66 +0,0 @@
--- Schema with enums, nulls, extra properties disallowed
-SELECT jsonschema_is_valid('{
-  "type": "object",
-  "properties": {
-    "status": { "type": "string", "enum": ["active", "inactive", "pending"] },
-    "email": { "type": ["string", "null"], "format": "email" }
-  },
-  "required": ["status"],
-  "additionalProperties": false
-}'::json);
-
--- Valid enum and null email
-SELECT jsonschema_validation_errors(
-  '{
-    "type": "object",
-    "properties": {
-      "status": { "type": "string", "enum": ["active", "inactive", "pending"] },
-      "email": { "type": ["string", "null"], "format": "email" }
-    },
-    "required": ["status"],
-    "additionalProperties": false
-  }'::json,
-  '{"status": "active", "email": null}'::json
-);
-
--- Invalid enum value
-SELECT jsonschema_validation_errors(
-  '{
-    "type": "object",
-    "properties": {
-      "status": { "type": "string", "enum": ["active", "inactive", "pending"] },
-      "email": { "type": ["string", "null"], "format": "email" }
-    },
-    "required": ["status"],
-    "additionalProperties": false
-  }'::json,
-  '{"status": "disabled", "email": null}'::json
-);
-
--- Invalid email format (assuming format is validated)
-SELECT jsonschema_validation_errors(
-  '{
-    "type": "object",
-    "properties": {
-      "status": { "type": "string", "enum": ["active", "inactive", "pending"] },
-      "email": { "type": ["string", "null"], "format": "email" }
-    },
-    "required": ["status"],
-    "additionalProperties": false
-  }'::json,
-  '{"status": "active", "email": "not-an-email"}'::json
-);
-
--- Extra property not allowed
-SELECT jsonschema_validation_errors(
-  '{
-    "type": "object",
-    "properties": {
-      "status": { "type": "string", "enum": ["active", "inactive", "pending"] },
-      "email": { "type": ["string", "null"], "format": "email" }
-    },
-    "required": ["status"],
-    "additionalProperties": false
-  }'::json,
-  '{"status": "active", "extra": "should not be here"}'::json
-);

docker-compose/ext-src/pg_jsonschema-src/sql/jsonschema_valid_api.sql

@@ -1,48 +0,0 @@
--- Define schema
-SELECT jsonschema_is_valid('{
-  "type": "object",
-  "properties": {
-    "username": { "type": "string" },
-    "age": { "type": "integer" }
-  },
-  "required": ["username"]
-}'::json);
-
--- Valid instance
-SELECT jsonschema_validation_errors(
-  '{
-    "type": "object",
-    "properties": {
-      "username": { "type": "string" },
-      "age": { "type": "integer" }
-    },
-    "required": ["username"]
-  }'::json,
-  '{"username": "alice", "age": 25}'::json
-);
-
--- Invalid instance: missing required "username"
-SELECT jsonschema_validation_errors(
-  '{
-    "type": "object",
-    "properties": {
-      "username": { "type": "string" },
-      "age": { "type": "integer" }
-    },
-    "required": ["username"]
-  }'::json,
-  '{"age": 25}'::json
-);
-
--- Invalid instance: wrong type for "age"
-SELECT jsonschema_validation_errors(
-  '{
-    "type": "object",
-    "properties": {
-      "username": { "type": "string" },
-      "age": { "type": "integer" }
-    },
-    "required": ["username"]
-  }'::json,
-  '{"username": "bob", "age": "twenty"}'::json
-);

docker-compose/ext-src/pg_session_jwt-src/Makefile

@@ -1,9 +0,0 @@
-EXTENSION = pg_session_jwt
-
-REGRESS = basic_functions
-REGRESS_OPTS = --load-extension=$(EXTENSION)
-export PGOPTIONS = -c pg_session_jwt.jwk={"crv":"Ed25519","kty":"OKP","x":"R_Abz-63zJ00l-IraL5fQhwkhGVZCSooQFV5ntC3C7M"}
-
-PG_CONFIG ?= pg_config
-PGXS := $(shell $(PG_CONFIG) --pgxs)
-include $(PGXS)

docker-compose/ext-src/pg_session_jwt-src/expected/basic_functions.out

@@ -1,35 +0,0 @@
--- Basic functionality tests for pg_session_jwt
--- Test auth.init() function
-SELECT auth.init();
- init 
-------
- 
-(1 row)
-
--- Test an invalid JWT
-SELECT auth.jwt_session_init('INVALID-JWT');
-ERROR:  invalid JWT encoding
--- Test creating a session with an expired JWT
-SELECT auth.jwt_session_init('eyJhbGciOiJFZERTQSJ9.eyJleHAiOjE3NDI1NjQ0MzIsImlhdCI6MTc0MjU2NDI1MiwianRpIjo0MjQyNDIsInN1YiI6InVzZXIxMjMifQ.A6FwKuaSduHB9O7Gz37g0uoD_U9qVS0JNtT7YABGVgB7HUD1AMFc9DeyhNntWBqncg8k5brv-hrNTuUh5JYMAw');
-ERROR:  Token used after it has expired
--- Test creating a session with a valid JWT
-SELECT auth.jwt_session_init('eyJhbGciOiJFZERTQSJ9.eyJleHAiOjQ4OTYxNjQyNTIsImlhdCI6MTc0MjU2NDI1MiwianRpIjo0MzQzNDMsInN1YiI6InVzZXIxMjMifQ.2TXVgjb6JSUq6_adlvp-m_SdOxZSyGS30RS9TLB0xu2N83dMSs2NybwE1NMU8Fb0tcAZR_ET7M2rSxbTrphfCg');
- jwt_session_init 
-------------------
- 
-(1 row)
-
--- Test auth.session() function
-SELECT auth.session();
-                                 session                                 
--------------------------------------------------------------------------
- {"exp": 4896164252, "iat": 1742564252, "jti": 434343, "sub": "user123"}
-(1 row)
-
--- Test auth.user_id() function
-SELECT auth.user_id() AS user_id;
- user_id 
----------
- user123
-(1 row)
-

docker-compose/ext-src/pg_session_jwt-src/sql/basic_functions.sql

@@ -1,19 +0,0 @@
--- Basic functionality tests for pg_session_jwt
-
--- Test auth.init() function
-SELECT auth.init();
-
--- Test an invalid JWT
-SELECT auth.jwt_session_init('INVALID-JWT');
-
--- Test creating a session with an expired JWT
-SELECT auth.jwt_session_init('eyJhbGciOiJFZERTQSJ9.eyJleHAiOjE3NDI1NjQ0MzIsImlhdCI6MTc0MjU2NDI1MiwianRpIjo0MjQyNDIsInN1YiI6InVzZXIxMjMifQ.A6FwKuaSduHB9O7Gz37g0uoD_U9qVS0JNtT7YABGVgB7HUD1AMFc9DeyhNntWBqncg8k5brv-hrNTuUh5JYMAw');
-
--- Test creating a session with a valid JWT
-SELECT auth.jwt_session_init('eyJhbGciOiJFZERTQSJ9.eyJleHAiOjQ4OTYxNjQyNTIsImlhdCI6MTc0MjU2NDI1MiwianRpIjo0MzQzNDMsInN1YiI6InVzZXIxMjMifQ.2TXVgjb6JSUq6_adlvp-m_SdOxZSyGS30RS9TLB0xu2N83dMSs2NybwE1NMU8Fb0tcAZR_ET7M2rSxbTrphfCg');
-
--- Test auth.session() function
-SELECT auth.session();
-
--- Test auth.user_id() function
-SELECT auth.user_id() AS user_id;

docs/storage_controller.md

@@ -151,7 +151,7 @@ Example body:
 ```
 {
   "tenant_id": "1f359dd625e519a1a4e8d7509690f6fc",
-  "stripe_size": 2048,
+  "stripe_size": 32768,
   "shards": [
       {"node_id": 344, "shard_number": 0},
       {"node_id": 722, "shard_number": 1},

libs/compute_api/src/requests.rs

@@ -5,14 +5,6 @@ use crate::privilege::Privilege;
 use crate::responses::ComputeCtlConfig;
 use crate::spec::{ComputeSpec, ExtVersion, PgIdent};
 
-/// When making requests to the `compute_ctl` external HTTP server, the client
-/// must specify a set of claims in `Authorization` header JWTs such that
-/// `compute_ctl` can authorize the request.
-#[derive(Clone, Debug, Deserialize, Serialize)]
-pub struct ComputeClaims {
-    pub compute_id: String,
-}
-
 /// Request of the /configure API
 ///
 /// We now pass only `spec` in the configuration request, but later we can
@@ -38,3 +30,9 @@ pub struct SetRoleGrantsRequest {
     pub privileges: Vec<Privilege>,
     pub role: PgIdent,
 }
+
+/// Request of the /configure_telemetry API
+#[derive(Debug, Deserialize, Serialize)]
+pub struct ConfigureTelemetryRequest {
+    pub logs_export_host: Option<String>,
+}

libs/compute_api/src/responses.rs

@@ -14,32 +14,6 @@ pub struct GenericAPIError {
     pub error: String,
 }
 
-/// All configuration parameters necessary for a compute. When
-/// [`ComputeConfig::spec`] is provided, it means that the compute is attached
-/// to a tenant. [`ComputeConfig::compute_ctl_config`] will always be provided
-/// and contains parameters necessary for operating `compute_ctl` independently
-/// of whether a tenant is attached to the compute or not.
-///
-/// This also happens to be the body of `compute_ctl`'s /configure request.
-#[derive(Debug, Deserialize, Serialize)]
-pub struct ComputeConfig {
-    /// The compute spec
-    pub spec: Option<ComputeSpec>,
-
-    /// The compute_ctl configuration
-    #[allow(dead_code)]
-    pub compute_ctl_config: ComputeCtlConfig,
-}
-
-impl From<ControlPlaneConfigResponse> for ComputeConfig {
-    fn from(value: ControlPlaneConfigResponse) -> Self {
-        Self {
-            spec: value.spec,
-            compute_ctl_config: value.compute_ctl_config,
-        }
-    }
-}
-
 #[derive(Debug, Clone, Serialize)]
 pub struct ExtensionInstallResponse {
     pub extension: PgIdent,
@@ -160,7 +134,7 @@ pub struct CatalogObjects {
     pub databases: Vec<Database>,
 }
 
-#[derive(Clone, Debug, Deserialize, Eq, PartialEq, Serialize)]
+#[derive(Clone, Debug, Deserialize, Serialize)]
 pub struct ComputeCtlConfig {
     /// Set of JSON web keys that the compute can use to authenticate
     /// communication from the control plane.
@@ -179,7 +153,7 @@ impl Default for ComputeCtlConfig {
     }
 }
 
-#[derive(Clone, Debug, Deserialize, Eq, PartialEq, Serialize)]
+#[derive(Clone, Debug, Deserialize, Serialize)]
 pub struct TlsConfig {
     pub key_path: String,
     pub cert_path: String,
@@ -187,7 +161,7 @@ pub struct TlsConfig {
 
 /// Response of the `/computes/{compute_id}/spec` control-plane API.
 #[derive(Deserialize, Debug)]
-pub struct ControlPlaneConfigResponse {
+pub struct ControlPlaneSpecResponse {
     pub spec: Option<ComputeSpec>,
     pub status: ControlPlaneComputeStatus,
     pub compute_ctl_config: ComputeCtlConfig,

libs/compute_api/src/spec.rs

@@ -1,8 +1,8 @@
-//! The ComputeSpec contains all the information needed to start up
-//! the right version of PostgreSQL, and connect it to the storage nodes.
-//! It can be passed as part of the `config.json`, or the control plane can
-//! provide it by calling the compute_ctl's `/compute_ctl` endpoint, or
-//! compute_ctl can fetch it by calling the control plane's API.
+//! `ComputeSpec` represents the contents of the spec.json file.
+//!
+//! The spec.json file is used to pass information to 'compute_ctl'. It contains
+//! all the information needed to start up the right version of PostgreSQL,
+//! and connect it to the storage nodes.
 use std::collections::HashMap;
 
 use indexmap::IndexMap;
@@ -104,12 +104,6 @@ pub struct ComputeSpec {
     pub timeline_id: Option<TimelineId>,
     pub pageserver_connstring: Option<String>,
 
-    // More neon ids that we expose to the compute_ctl
-    // and to postgres as neon extension GUCs.
-    pub project_id: Option<String>,
-    pub branch_id: Option<String>,
-    pub endpoint_id: Option<String>,
-
     /// Safekeeper membership config generation. It is put in
     /// neon.safekeepers GUC and serves two purposes:
     /// 1) Non zero value forces walproposer to use membership configurations.
@@ -165,13 +159,15 @@ pub struct ComputeSpec {
     #[serde(default)] // Default false
     pub drop_subscriptions_before_start: bool,
 
-    /// Log level for compute audit logging
+    /// Log level for audit logging:
+    ///
+    /// Disabled - no audit logging. This is the default.
+    /// log - log masked statements to the postgres log using pgaudit extension
+    /// hipaa - log unmasked statements to the file using pgaudit and pgauditlogtofile extension
+    ///
+    /// Extensions should be present in shared_preload_libraries
     #[serde(default)]
     pub audit_log_level: ComputeAudit,
-
-    /// Hostname and the port of the otel collector. Leave empty to disable Postgres logs forwarding.
-    /// Example: config-shy-breeze-123-collector-monitoring.neon-telemetry.svc.cluster.local:10514
-    pub logs_export_host: Option<String>,
 }
 
 /// Feature flag to signal `compute_ctl` to enable certain experimental functionality.
@@ -183,6 +179,9 @@ pub enum ComputeFeature {
     /// track short-lived connections as user activity.
     ActivityMonitorExperimental,
 
+    /// Allow to configure rsyslog for Postgres logs export
+    PostgresLogsExport,
+
     /// This is a special feature flag that is used to represent unknown feature flags.
     /// Basically all unknown to enum flags are represented as this one. See unit test
     /// `parse_unknown_features()` for more details.
@@ -289,25 +288,14 @@ impl ComputeMode {
 }
 
 /// Log level for audit logging
+/// Disabled, log, hipaa
+/// Default is Disabled
 #[derive(Clone, Debug, Default, Eq, PartialEq, Deserialize, Serialize)]
 pub enum ComputeAudit {
     #[default]
     Disabled,
-    // Deprecated, use Base instead
     Log,
-    // (pgaudit.log = 'ddl', pgaudit.log_parameter='off')
-    // logged to the standard postgresql log stream
-    Base,
-    // Deprecated, use Full or Extended instead
     Hipaa,
-    // (pgaudit.log = 'all, -misc', pgaudit.log_parameter='off')
-    // logged to separate files collected by rsyslog
-    // into dedicated log storage with strict access
-    Extended,
-    // (pgaudit.log='all', pgaudit.log_parameter='on'),
-    // logged to separate files collected by rsyslog
-    // into dedicated log storage with strict access.
-    Full,
 }
 
 #[derive(Clone, Debug, Default, Deserialize, Serialize, PartialEq, Eq)]

libs/http-utils/Cargo.toml

@@ -14,7 +14,6 @@ futures.workspace = true
 hyper0.workspace = true
 itertools.workspace = true
 jemalloc_pprof.workspace = true
-jsonwebtoken.workspace = true
 once_cell.workspace = true
 pprof.workspace = true
 regex.workspace = true
@@ -31,7 +30,6 @@ tokio.workspace = true
 tracing.workspace = true
 url.workspace = true
 uuid.workspace = true
-x509-cert.workspace = true
 
 # to use tokio channels as streams, this is faster to compile than async_stream
 # why is it only here? no other crate should use it, streams are rarely needed.

libs/http-utils/src/endpoint.rs

@@ -8,7 +8,6 @@ use bytes::{Bytes, BytesMut};
 use hyper::header::{AUTHORIZATION, CONTENT_DISPOSITION, CONTENT_TYPE, HeaderName};
 use hyper::http::HeaderValue;
 use hyper::{Body, Method, Request, Response};
-use jsonwebtoken::TokenData;
 use metrics::{Encoder, IntCounter, TextEncoder, register_int_counter};
 use once_cell::sync::Lazy;
 use pprof::ProfilerGuardBuilder;
@@ -619,7 +618,7 @@ pub fn auth_middleware<B: hyper::body::HttpBody + Send + Sync + 'static>(
                     })?;
                     let token = parse_token(header_value)?;
 
-                    let data: TokenData<Claims> = auth.decode(token).map_err(|err| {
+                    let data = auth.decode(token).map_err(|err| {
                         warn!("Authentication error: {err}");
                         // Rely on From<AuthError> for ApiError impl
                         err

libs/http-utils/src/server.rs

@@ -4,8 +4,6 @@ use futures::StreamExt;
 use futures::stream::FuturesUnordered;
 use hyper0::Body;
 use hyper0::server::conn::Http;
-use metrics::{IntCounterVec, register_int_counter_vec};
-use once_cell::sync::Lazy;
 use routerify::{RequestService, RequestServiceBuilder};
 use tokio::io::{AsyncRead, AsyncWrite};
 use tokio_rustls::TlsAcceptor;
@@ -28,24 +26,6 @@ pub struct Server {
     tls_acceptor: Option<TlsAcceptor>,
 }
 
-static CONNECTION_STARTED_COUNT: Lazy<IntCounterVec> = Lazy::new(|| {
-    register_int_counter_vec!(
-        "http_server_connection_started_total",
-        "Number of established http/https connections",
-        &["scheme"]
-    )
-    .expect("failed to define a metric")
-});
-
-static CONNECTION_ERROR_COUNT: Lazy<IntCounterVec> = Lazy::new(|| {
-    register_int_counter_vec!(
-        "http_server_connection_errors_total",
-        "Number of occured connection errors by type",
-        &["type"]
-    )
-    .expect("failed to define a metric")
-});
-
 impl Server {
     pub fn new(
         request_service: Arc<RequestServiceBuilder<Body, ApiError>>,
@@ -80,15 +60,6 @@ impl Server {
             false
         }
 
-        let tcp_error_cnt = CONNECTION_ERROR_COUNT.with_label_values(&["tcp"]);
-        let tls_error_cnt = CONNECTION_ERROR_COUNT.with_label_values(&["tls"]);
-        let http_error_cnt = CONNECTION_ERROR_COUNT.with_label_values(&["http"]);
-        let https_error_cnt = CONNECTION_ERROR_COUNT.with_label_values(&["https"]);
-        let panic_error_cnt = CONNECTION_ERROR_COUNT.with_label_values(&["panic"]);
-
-        let http_connection_cnt = CONNECTION_STARTED_COUNT.with_label_values(&["http"]);
-        let https_connection_cnt = CONNECTION_STARTED_COUNT.with_label_values(&["https"]);
-
         let mut connections = FuturesUnordered::new();
         loop {
             tokio::select! {
@@ -96,7 +67,6 @@ impl Server {
                     let (tcp_stream, remote_addr) = match stream {
                         Ok(stream) => stream,
                         Err(err) => {
-                            tcp_error_cnt.inc();
                             if !suppress_io_error(&err) {
                                 info!("Failed to accept TCP connection: {err:#}");
                             }
@@ -108,18 +78,11 @@ impl Server {
                     let tls_acceptor = self.tls_acceptor.clone();
                     let cancel = cancel.clone();
 
-                    let tls_error_cnt = tls_error_cnt.clone();
-                    let http_error_cnt = http_error_cnt.clone();
-                    let https_error_cnt = https_error_cnt.clone();
-                    let http_connection_cnt = http_connection_cnt.clone();
-                    let https_connection_cnt = https_connection_cnt.clone();
-
                     connections.push(tokio::spawn(
                         async move {
                             match tls_acceptor {
                                 Some(tls_acceptor) => {
                                     // Handle HTTPS connection.
-                                    https_connection_cnt.inc();
                                     let tls_stream = tokio::select! {
                                         tls_stream = tls_acceptor.accept(tcp_stream) => tls_stream,
                                         _ = cancel.cancelled() => return,
@@ -127,7 +90,6 @@ impl Server {
                                     let tls_stream = match tls_stream {
                                         Ok(tls_stream) => tls_stream,
                                         Err(err) => {
-                                            tls_error_cnt.inc();
                                             if !suppress_io_error(&err) {
                                                 info!(%remote_addr, "Failed to accept TLS connection: {err:#}");
                                             }
@@ -135,7 +97,6 @@ impl Server {
                                         }
                                     };
                                     if let Err(err) = Self::serve_connection(tls_stream, service, cancel).await {
-                                        https_error_cnt.inc();
                                         if !suppress_hyper_error(&err) {
                                             info!(%remote_addr, "Failed to serve HTTPS connection: {err:#}");
                                         }
@@ -143,9 +104,7 @@ impl Server {
                                 }
                                 None => {
                                     // Handle HTTP connection.
-                                    http_connection_cnt.inc();
                                     if let Err(err) = Self::serve_connection(tcp_stream, service, cancel).await {
-                                        http_error_cnt.inc();
                                         if !suppress_hyper_error(&err) {
                                             info!(%remote_addr, "Failed to serve HTTP connection: {err:#}");
                                         }
@@ -156,7 +115,6 @@ impl Server {
                  }
                 Some(conn) = connections.next() => {
                     if let Err(err) = conn {
-                        panic_error_cnt.inc();
                         error!("Connection panicked: {err:#}");
                     }
                 }
@@ -164,7 +122,6 @@ impl Server {
                     // Wait for graceful shutdown of all connections.
                     while let Some(conn) = connections.next().await {
                         if let Err(err) = conn {
-                            panic_error_cnt.inc();
                             error!("Connection panicked: {err:#}");
                         }
                     }

libs/http-utils/src/tls_certs.rs

@@ -3,14 +3,11 @@ use std::{sync::Arc, time::Duration};
 use anyhow::Context;
 use arc_swap::ArcSwap;
 use camino::Utf8Path;
-use metrics::{IntCounterVec, UIntGaugeVec, register_int_counter_vec, register_uint_gauge_vec};
-use once_cell::sync::Lazy;
 use rustls::{
-    pki_types::{CertificateDer, PrivateKeyDer, UnixTime},
+    pki_types::{CertificateDer, PrivateKeyDer},
     server::{ClientHello, ResolvesServerCert},
     sign::CertifiedKey,
 };
-use x509_cert::der::Reader;
 
 pub async fn load_cert_chain(filename: &Utf8Path) -> anyhow::Result<Vec<CertificateDer<'static>>> {
     let cert_data = tokio::fs::read(filename)
@@ -56,76 +53,6 @@ pub async fn load_certified_key(
     Ok(certified_key)
 }
 
-/// rustls's CertifiedKey with extra parsed fields used for metrics.
-struct ParsedCertifiedKey {
-    certified_key: CertifiedKey,
-    expiration_time: UnixTime,
-}
-
-/// Parse expiration time from an X509 certificate.
-fn parse_expiration_time(cert: &CertificateDer<'_>) -> anyhow::Result<UnixTime> {
-    let parsed_cert = x509_cert::der::SliceReader::new(cert)
-        .context("Failed to parse cerficiate")?
-        .decode::<x509_cert::Certificate>()
-        .context("Failed to parse cerficiate")?;
-
-    Ok(UnixTime::since_unix_epoch(
-        parsed_cert
-            .tbs_certificate
-            .validity
-            .not_after
-            .to_unix_duration(),
-    ))
-}
-
-async fn load_and_parse_certified_key(
-    key_filename: &Utf8Path,
-    cert_filename: &Utf8Path,
-) -> anyhow::Result<ParsedCertifiedKey> {
-    let certified_key = load_certified_key(key_filename, cert_filename).await?;
-    let expiration_time = parse_expiration_time(certified_key.end_entity_cert()?)?;
-    Ok(ParsedCertifiedKey {
-        certified_key,
-        expiration_time,
-    })
-}
-
-static CERT_EXPIRATION_TIME: Lazy<UIntGaugeVec> = Lazy::new(|| {
-    register_uint_gauge_vec!(
-        "tls_certs_expiration_time_seconds",
-        "Expiration time of the loaded certificate since unix epoch in seconds",
-        &["resolver_name"]
-    )
-    .expect("failed to define a metric")
-});
-
-static CERT_RELOAD_STARTED_COUNTER: Lazy<IntCounterVec> = Lazy::new(|| {
-    register_int_counter_vec!(
-        "tls_certs_reload_started_total",
-        "Number of certificate reload loop iterations started",
-        &["resolver_name"]
-    )
-    .expect("failed to define a metric")
-});
-
-static CERT_RELOAD_UPDATED_COUNTER: Lazy<IntCounterVec> = Lazy::new(|| {
-    register_int_counter_vec!(
-        "tls_certs_reload_updated_total",
-        "Number of times the certificate was updated to the new one",
-        &["resolver_name"]
-    )
-    .expect("failed to define a metric")
-});
-
-static CERT_RELOAD_FAILED_COUNTER: Lazy<IntCounterVec> = Lazy::new(|| {
-    register_int_counter_vec!(
-        "tls_certs_reload_failed_total",
-        "Number of times the certificate reload failed",
-        &["resolver_name"]
-    )
-    .expect("failed to define a metric")
-});
-
 /// Implementation of [`rustls::server::ResolvesServerCert`] which reloads certificates from
 /// the disk periodically.
 #[derive(Debug)]
@@ -136,28 +63,16 @@ pub struct ReloadingCertificateResolver {
 impl ReloadingCertificateResolver {
     /// Creates a new Resolver by loading certificate and private key from FS and
     /// creating tokio::task to reload them with provided reload_period.
-    /// resolver_name is used as metric's label.
     pub async fn new(
-        resolver_name: &str,
         key_filename: &Utf8Path,
         cert_filename: &Utf8Path,
         reload_period: Duration,
     ) -> anyhow::Result<Arc<Self>> {
-        // Create metrics for current resolver.
-        let cert_expiration_time = CERT_EXPIRATION_TIME.with_label_values(&[resolver_name]);
-        let cert_reload_started_counter =
-            CERT_RELOAD_STARTED_COUNTER.with_label_values(&[resolver_name]);
-        let cert_reload_updated_counter =
-            CERT_RELOAD_UPDATED_COUNTER.with_label_values(&[resolver_name]);
-        let cert_reload_failed_counter =
-            CERT_RELOAD_FAILED_COUNTER.with_label_values(&[resolver_name]);
-
-        let parsed_key = load_and_parse_certified_key(key_filename, cert_filename).await?;
-
         let this = Arc::new(Self {
-            certified_key: ArcSwap::from_pointee(parsed_key.certified_key),
+            certified_key: ArcSwap::from_pointee(
+                load_certified_key(key_filename, cert_filename).await?,
+            ),
         });
-        cert_expiration_time.set(parsed_key.expiration_time.as_secs());
 
         tokio::spawn({
             let weak_this = Arc::downgrade(&this);
@@ -173,22 +88,17 @@ impl ReloadingCertificateResolver {
                         Some(this) => this,
                         None => break, // Resolver has been destroyed, exit.
                     };
-                    cert_reload_started_counter.inc();
-
-                    match load_and_parse_certified_key(&key_filename, &cert_filename).await {
-                        Ok(parsed_key) => {
-                            if parsed_key.certified_key.cert == this.certified_key.load().cert {
+                    match load_certified_key(&key_filename, &cert_filename).await {
+                        Ok(new_certified_key) => {
+                            if new_certified_key.cert == this.certified_key.load().cert {
                                 tracing::debug!("Certificate has not changed since last reloading");
                             } else {
                                 tracing::info!("Certificate has been reloaded");
-                                this.certified_key.store(Arc::new(parsed_key.certified_key));
-                                cert_expiration_time.set(parsed_key.expiration_time.as_secs());
-                                cert_reload_updated_counter.inc();
+                                this.certified_key.store(Arc::new(new_certified_key));
                             }
                             last_reload_failed = false;
                         }
                         Err(err) => {
-                            cert_reload_failed_counter.inc();
                             // Note: Reloading certs may fail if it conflicts with the script updating
                             // the files at the same time. Warn only if the error is persistent.
                             if last_reload_failed {

libs/pageserver_api/Cargo.toml

@@ -35,7 +35,6 @@ nix = {workspace = true, optional = true}
 reqwest.workspace = true
 rand.workspace = true
 tracing-utils.workspace = true
-once_cell.workspace = true
 
 [dev-dependencies]
 bincode.workspace = true

libs/pageserver_api/src/config.rs

@@ -180,7 +180,6 @@ pub struct ConfigToml {
     #[serde(skip_serializing_if = "Option::is_none")]
     pub generate_unarchival_heatmap: Option<bool>,
     pub tracing: Option<Tracing>,
-    pub enable_tls_page_service_api: bool,
 }
 
 #[derive(Debug, Clone, PartialEq, Eq, serde::Serialize, serde::Deserialize)]
@@ -207,10 +206,6 @@ pub struct PageServicePipeliningConfigPipelined {
     /// Causes runtime errors if larger than max get_vectored batch size.
     pub max_batch_size: NonZeroUsize,
     pub execution: PageServiceProtocolPipelinedExecutionStrategy,
-    // The default below is such that new versions of the software can start
-    // with the old configuration.
-    #[serde(default)]
-    pub batching: PageServiceProtocolPipelinedBatchingStrategy,
 }
 
 #[derive(Debug, Clone, Copy, PartialEq, Eq, serde::Serialize, serde::Deserialize)]
@@ -220,19 +215,6 @@ pub enum PageServiceProtocolPipelinedExecutionStrategy {
     Tasks,
 }
 
-#[derive(Default, Debug, Clone, Copy, PartialEq, Eq, serde::Serialize, serde::Deserialize)]
-#[serde(rename_all = "kebab-case")]
-pub enum PageServiceProtocolPipelinedBatchingStrategy {
-    /// All get page requests in a batch will be at the same LSN
-    #[default]
-    UniformLsn,
-    /// Get page requests in a batch may be at different LSN
-    ///
-    /// One key cannot be present more than once at different LSNs in
-    /// the same batch.
-    ScatteredLsn,
-}
-
 #[derive(Debug, Clone, PartialEq, Eq, serde::Serialize, serde::Deserialize)]
 #[serde(tag = "mode", rename_all = "kebab-case")]
 pub enum GetVectoredConcurrentIo {
@@ -379,8 +361,6 @@ pub struct TenantConfigToml {
     /// size exceeds `compaction_upper_limit * checkpoint_distance`.
     pub compaction_upper_limit: usize,
     pub compaction_algorithm: crate::models::CompactionAlgorithmSettings,
-    /// If true, enable shard ancestor compaction (enabled by default).
-    pub compaction_shard_ancestor: bool,
     /// If true, compact down L0 across all tenant timelines before doing regular compaction. L0
     /// compaction must be responsive to avoid read amp during heavy ingestion. Defaults to true.
     pub compaction_l0_first: bool,
@@ -471,8 +451,6 @@ pub struct TenantConfigToml {
     // gc-compaction related configs
     /// Enable automatic gc-compaction trigger on this tenant.
     pub gc_compaction_enabled: bool,
-    /// Enable verification of gc-compaction results.
-    pub gc_compaction_verification: bool,
     /// The initial threshold for gc-compaction in KB. Once the total size of layers below the gc-horizon is above this threshold,
     /// gc-compaction will be triggered.
     pub gc_compaction_initial_threshold_kb: u64,
@@ -634,12 +612,9 @@ impl Default for ConfigToml {
             page_service_pipelining: if !cfg!(test) {
                 PageServicePipeliningConfig::Serial
             } else {
-                // Do not turn this into the default until scattered reads have been
-                // validated and rolled-out fully.
                 PageServicePipeliningConfig::Pipelined(PageServicePipeliningConfigPipelined {
                     max_batch_size: NonZeroUsize::new(32).unwrap(),
                     execution: PageServiceProtocolPipelinedExecutionStrategy::ConcurrentFutures,
-                    batching: PageServiceProtocolPipelinedBatchingStrategy::ScatteredLsn,
                 })
             },
             get_vectored_concurrent_io: if !cfg!(test) {
@@ -656,7 +631,6 @@ impl Default for ConfigToml {
             load_previous_heatmap: None,
             generate_unarchival_heatmap: None,
             tracing: None,
-            enable_tls_page_service_api: false,
         }
     }
 }
@@ -679,13 +653,12 @@ pub mod tenant_conf_defaults {
 
     pub const DEFAULT_COMPACTION_PERIOD: &str = "20 s";
     pub const DEFAULT_COMPACTION_THRESHOLD: usize = 10;
-    pub const DEFAULT_COMPACTION_SHARD_ANCESTOR: bool = true;
 
     // This value needs to be tuned to avoid OOM. We have 3/4*CPUs threads for L0 compaction, that's
-    // 3/4*8=6 on most of our pageservers. Compacting 10 layers requires a maximum of
-    // DEFAULT_CHECKPOINT_DISTANCE*10 memory, that's 2560MB. So with this config, we can get a maximum peak
-    // compaction usage of 15360MB.
-    pub const DEFAULT_COMPACTION_UPPER_LIMIT: usize = 10;
+    // 3/4*16=9 on most of our pageservers. Compacting 20 layers requires about 1 GB memory (could
+    // be reduced later by optimizing L0 hole calculation to avoid loading all keys into memory). So
+    // with this config, we can get a maximum peak compaction usage of 9 GB.
+    pub const DEFAULT_COMPACTION_UPPER_LIMIT: usize = 20;
     // Enable L0 compaction pass and semaphore by default. L0 compaction must be responsive to avoid
     // read amp.
     pub const DEFAULT_COMPACTION_L0_FIRST: bool = true;
@@ -702,11 +675,8 @@ pub mod tenant_conf_defaults {
     // Relevant: https://github.com/neondatabase/neon/issues/3394
     pub const DEFAULT_GC_PERIOD: &str = "1 hr";
     pub const DEFAULT_IMAGE_CREATION_THRESHOLD: usize = 3;
-    // Currently, any value other than 0 will trigger image layer creation preemption immediately with L0 backpressure
-    // without looking at the exact number of L0 layers.
-    // It was expected to have the following behavior:
-    // > If there are more than threshold * compaction_threshold (that is 3 * 10 in the default config) L0 layers, image
-    // > layer creation will end immediately. Set to 0 to disable.
+    // If there are more than threshold * compaction_threshold (that is 3 * 10 in the default config) L0 layers, image
+    // layer creation will end immediately. Set to 0 to disable.
     pub const DEFAULT_IMAGE_CREATION_PREEMPT_THRESHOLD: usize = 3;
     pub const DEFAULT_PITR_INTERVAL: &str = "7 days";
     pub const DEFAULT_WALRECEIVER_CONNECT_TIMEOUT: &str = "10 seconds";
@@ -720,7 +690,6 @@ pub mod tenant_conf_defaults {
     // image layers should be created.
     pub const DEFAULT_IMAGE_LAYER_CREATION_CHECK_THRESHOLD: u8 = 2;
     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;
 }
@@ -740,7 +709,6 @@ impl Default for TenantConfigToml {
             compaction_algorithm: crate::models::CompactionAlgorithmSettings {
                 kind: DEFAULT_COMPACTION_ALGORITHM,
             },
-            compaction_shard_ancestor: DEFAULT_COMPACTION_SHARD_ANCESTOR,
             compaction_l0_first: DEFAULT_COMPACTION_L0_FIRST,
             compaction_l0_semaphore: DEFAULT_COMPACTION_L0_SEMAPHORE,
             l0_flush_delay_threshold: None,
@@ -776,7 +744,6 @@ impl Default for TenantConfigToml {
             wal_receiver_protocol_override: None,
             rel_size_v2_enabled: false,
             gc_compaction_enabled: DEFAULT_GC_COMPACTION_ENABLED,
-            gc_compaction_verification: DEFAULT_GC_COMPACTION_VERIFICATION,
             gc_compaction_initial_threshold_kb: DEFAULT_GC_COMPACTION_INITIAL_THRESHOLD_KB,
             gc_compaction_ratio_percent: DEFAULT_GC_COMPACTION_RATIO_PERCENT,
             sampling_ratio: None,

libs/pageserver_api/src/controller_api.rs

@@ -7,8 +7,7 @@ use std::time::{Duration, Instant};
 /// API (`/control/v1` prefix).  Implemented by the server
 /// in [`storage_controller::http`]
 use serde::{Deserialize, Serialize};
-use utils::id::{NodeId, TenantId, TimelineId};
-use utils::lsn::Lsn;
+use utils::id::{NodeId, TenantId};
 
 use crate::models::{PageserverUtilization, ShardParameters, TenantConfig};
 use crate::shard::{ShardStripeSize, TenantShardId};
@@ -500,15 +499,6 @@ pub struct SafekeeperSchedulingPolicyRequest {
     pub scheduling_policy: SkSchedulingPolicy,
 }
 
-/// Import request for safekeeper timelines.
-#[derive(Serialize, Deserialize, Clone)]
-pub struct TimelineImportRequest {
-    pub tenant_id: TenantId,
-    pub timeline_id: TimelineId,
-    pub start_lsn: Lsn,
-    pub sk_set: Vec<NodeId>,
-}
-
 #[cfg(test)]
 mod test {
     use serde_json;

libs/pageserver_api/src/key.rs

@@ -927,7 +927,7 @@ impl Key {
 
     /// Guaranteed to return `Ok()` if [`Self::is_rel_block_key`] returns `true` for `key`.
     #[inline(always)]
-    pub fn to_rel_block(self) -> Result<(RelTag, BlockNumber), ToRelBlockError> {
+    pub fn to_rel_block(self) -> anyhow::Result<(RelTag, BlockNumber)> {
         Ok(match self.field1 {
             0x00 => (
                 RelTag {
@@ -938,7 +938,7 @@ impl Key {
                 },
                 self.field6,
             ),
-            _ => return Err(ToRelBlockError(self.field1)),
+            _ => anyhow::bail!("unexpected value kind 0x{:02x}", self.field1),
         })
     }
 }
@@ -951,17 +951,6 @@ impl std::str::FromStr for Key {
     }
 }
 
-#[derive(Debug)]
-pub struct ToRelBlockError(u8);
-
-impl fmt::Display for ToRelBlockError {
-    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
-        write!(f, "unexpected value kind 0x{:02x}", self.0)
-    }
-}
-
-impl std::error::Error for ToRelBlockError {}
-
 #[cfg(test)]
 mod tests {
     use std::str::FromStr;

libs/pageserver_api/src/keyspace.rs

@@ -613,7 +613,8 @@ mod tests {
     use rand::{RngCore, SeedableRng};
 
     use super::*;
-    use crate::shard::{DEFAULT_STRIPE_SIZE, ShardCount, ShardNumber, ShardStripeSize};
+    use crate::models::ShardParameters;
+    use crate::shard::{ShardCount, ShardNumber};
 
     // Helper function to create a key range.
     //
@@ -963,8 +964,12 @@ mod tests {
     }
     #[test]
     fn sharded_range_relation_gap() {
-        let shard_identity =
-            ShardIdentity::new(ShardNumber(0), ShardCount::new(4), DEFAULT_STRIPE_SIZE).unwrap();
+        let shard_identity = ShardIdentity::new(
+            ShardNumber(0),
+            ShardCount::new(4),
+            ShardParameters::DEFAULT_STRIPE_SIZE,
+        )
+        .unwrap();
 
         let range = ShardedRange::new(
             Range {
@@ -980,8 +985,12 @@ mod tests {
 
     #[test]
     fn shard_identity_keyspaces_single_key() {
-        let shard_identity =
-            ShardIdentity::new(ShardNumber(1), ShardCount::new(4), DEFAULT_STRIPE_SIZE).unwrap();
+        let shard_identity = ShardIdentity::new(
+            ShardNumber(1),
+            ShardCount::new(4),
+            ShardParameters::DEFAULT_STRIPE_SIZE,
+        )
+        .unwrap();
 
         let range = ShardedRange::new(
             Range {
@@ -1025,8 +1034,12 @@ mod tests {
 
     #[test]
     fn shard_identity_keyspaces_forkno_gap() {
-        let shard_identity =
-            ShardIdentity::new(ShardNumber(1), ShardCount::new(4), DEFAULT_STRIPE_SIZE).unwrap();
+        let shard_identity = ShardIdentity::new(
+            ShardNumber(1),
+            ShardCount::new(4),
+            ShardParameters::DEFAULT_STRIPE_SIZE,
+        )
+        .unwrap();
 
         let range = ShardedRange::new(
             Range {
@@ -1048,7 +1061,7 @@ mod tests {
             let shard_identity = ShardIdentity::new(
                 ShardNumber(shard_number),
                 ShardCount::new(4),
-                DEFAULT_STRIPE_SIZE,
+                ShardParameters::DEFAULT_STRIPE_SIZE,
             )
             .unwrap();
 
@@ -1131,44 +1144,37 @@ mod tests {
     /// for a single tenant.
     #[test]
     fn sharded_range_fragment_simple() {
-        const SHARD_COUNT: u8 = 4;
-        const STRIPE_SIZE: u32 = DEFAULT_STRIPE_SIZE.0;
-
         let shard_identity = ShardIdentity::new(
             ShardNumber(0),
-            ShardCount::new(SHARD_COUNT),
-            ShardStripeSize(STRIPE_SIZE),
+            ShardCount::new(4),
+            ShardParameters::DEFAULT_STRIPE_SIZE,
         )
         .unwrap();
 
         // A range which we happen to know covers exactly one stripe which belongs to this shard
         let input_start = Key::from_hex("000000067f00000001000000ae0000000000").unwrap();
-        let mut input_end = input_start;
-        input_end.field6 += STRIPE_SIZE; // field6 is block number
+        let input_end = Key::from_hex("000000067f00000001000000ae0000008000").unwrap();
 
         // Ask for stripe_size blocks, we get the whole stripe
         assert_eq!(
-            do_fragment(input_start, input_end, &shard_identity, STRIPE_SIZE),
-            (STRIPE_SIZE, vec![(STRIPE_SIZE, input_start..input_end)])
+            do_fragment(input_start, input_end, &shard_identity, 32768),
+            (32768, vec![(32768, input_start..input_end)])
         );
 
         // Ask for more, we still get the whole stripe
         assert_eq!(
-            do_fragment(input_start, input_end, &shard_identity, 10 * STRIPE_SIZE),
-            (STRIPE_SIZE, vec![(STRIPE_SIZE, input_start..input_end)])
+            do_fragment(input_start, input_end, &shard_identity, 10000000),
+            (32768, vec![(32768, input_start..input_end)])
         );
 
         // Ask for target_nblocks of half the stripe size, we get two halves
         assert_eq!(
-            do_fragment(input_start, input_end, &shard_identity, STRIPE_SIZE / 2),
+            do_fragment(input_start, input_end, &shard_identity, 16384),
             (
-                STRIPE_SIZE,
+                32768,
                 vec![
-                    (
-                        STRIPE_SIZE / 2,
-                        input_start..input_start.add(STRIPE_SIZE / 2)
-                    ),
-                    (STRIPE_SIZE / 2, input_start.add(STRIPE_SIZE / 2)..input_end)
+                    (16384, input_start..input_start.add(16384)),
+                    (16384, input_start.add(16384)..input_end)
                 ]
             )
         );
@@ -1176,53 +1182,40 @@ mod tests {
 
     #[test]
     fn sharded_range_fragment_multi_stripe() {
-        const SHARD_COUNT: u8 = 4;
-        const STRIPE_SIZE: u32 = DEFAULT_STRIPE_SIZE.0;
-        const RANGE_SIZE: u32 = SHARD_COUNT as u32 * STRIPE_SIZE;
-
         let shard_identity = ShardIdentity::new(
             ShardNumber(0),
-            ShardCount::new(SHARD_COUNT),
-            ShardStripeSize(STRIPE_SIZE),
+            ShardCount::new(4),
+            ShardParameters::DEFAULT_STRIPE_SIZE,
         )
         .unwrap();
 
         // A range which covers multiple stripes, exactly one of which belongs to the current shard.
         let input_start = Key::from_hex("000000067f00000001000000ae0000000000").unwrap();
-        let mut input_end = input_start;
-        input_end.field6 += RANGE_SIZE; // field6 is block number
-
+        let input_end = Key::from_hex("000000067f00000001000000ae0000020000").unwrap();
         // Ask for all the blocks, get a fragment that covers the whole range but reports
         // its size to be just the blocks belonging to our shard.
         assert_eq!(
-            do_fragment(input_start, input_end, &shard_identity, RANGE_SIZE),
-            (STRIPE_SIZE, vec![(STRIPE_SIZE, input_start..input_end)])
+            do_fragment(input_start, input_end, &shard_identity, 131072),
+            (32768, vec![(32768, input_start..input_end)])
         );
 
-        // Ask for a sub-stripe quantity that results in 3 fragments.
-        let limit = STRIPE_SIZE / 3 + 1;
+        // Ask for a sub-stripe quantity
         assert_eq!(
-            do_fragment(input_start, input_end, &shard_identity, limit),
+            do_fragment(input_start, input_end, &shard_identity, 16000),
             (
-                STRIPE_SIZE,
+                32768,
                 vec![
-                    (limit, input_start..input_start.add(limit)),
-                    (limit, input_start.add(limit)..input_start.add(2 * limit)),
-                    (
-                        STRIPE_SIZE - 2 * limit,
-                        input_start.add(2 * limit)..input_end
-                    ),
+                    (16000, input_start..input_start.add(16000)),
+                    (16000, input_start.add(16000)..input_start.add(32000)),
+                    (768, input_start.add(32000)..input_end),
                 ]
             )
         );
 
         // Try on a range that starts slightly after our owned stripe
         assert_eq!(
-            do_fragment(input_start.add(1), input_end, &shard_identity, RANGE_SIZE),
-            (
-                STRIPE_SIZE - 1,
-                vec![(STRIPE_SIZE - 1, input_start.add(1)..input_end)]
-            )
+            do_fragment(input_start.add(1), input_end, &shard_identity, 131072),
+            (32767, vec![(32767, input_start.add(1)..input_end)])
         );
     }
 
@@ -1230,40 +1223,32 @@ mod tests {
     /// a previous relation.
     #[test]
     fn sharded_range_fragment_starting_from_logical_size() {
-        const SHARD_COUNT: u8 = 4;
-        const STRIPE_SIZE: u32 = DEFAULT_STRIPE_SIZE.0;
-        const RANGE_SIZE: u32 = SHARD_COUNT as u32 * STRIPE_SIZE;
-
         let input_start = Key::from_hex("000000067f00000001000000ae00ffffffff").unwrap();
-        let mut input_end = Key::from_hex("000000067f00000001000000ae0100000000").unwrap();
-        input_end.field6 += RANGE_SIZE; // field6 is block number
+        let input_end = Key::from_hex("000000067f00000001000000ae0100008000").unwrap();
 
         // Shard 0 owns the first stripe in the relation, and the preceding logical size is shard local too
         let shard_identity = ShardIdentity::new(
             ShardNumber(0),
-            ShardCount::new(SHARD_COUNT),
-            ShardStripeSize(STRIPE_SIZE),
+            ShardCount::new(4),
+            ShardParameters::DEFAULT_STRIPE_SIZE,
         )
         .unwrap();
         assert_eq!(
-            do_fragment(input_start, input_end, &shard_identity, 2 * STRIPE_SIZE),
-            (
-                STRIPE_SIZE + 1,
-                vec![(STRIPE_SIZE + 1, input_start..input_end)]
-            )
+            do_fragment(input_start, input_end, &shard_identity, 0x10000),
+            (0x8001, vec![(0x8001, input_start..input_end)])
         );
 
         // Shard 1 does not own the first stripe in the relation, but it does own the logical size (all shards
         // store all logical sizes)
         let shard_identity = ShardIdentity::new(
             ShardNumber(1),
-            ShardCount::new(SHARD_COUNT),
-            ShardStripeSize(STRIPE_SIZE),
+            ShardCount::new(4),
+            ShardParameters::DEFAULT_STRIPE_SIZE,
         )
         .unwrap();
         assert_eq!(
-            do_fragment(input_start, input_end, &shard_identity, 2 * STRIPE_SIZE),
-            (1, vec![(1, input_start..input_end)])
+            do_fragment(input_start, input_end, &shard_identity, 0x10000),
+            (0x1, vec![(0x1, input_start..input_end)])
         );
     }
 
@@ -1299,8 +1284,12 @@ mod tests {
         );
 
         // Same, but using a sharded identity
-        let shard_identity =
-            ShardIdentity::new(ShardNumber(0), ShardCount::new(4), DEFAULT_STRIPE_SIZE).unwrap();
+        let shard_identity = ShardIdentity::new(
+            ShardNumber(0),
+            ShardCount::new(4),
+            ShardParameters::DEFAULT_STRIPE_SIZE,
+        )
+        .unwrap();
         assert_eq!(
             do_fragment(input_start, input_end, &shard_identity, 0x8000),
             (u32::MAX, vec![(u32::MAX, input_start..input_end),])
@@ -1342,7 +1331,7 @@ mod tests {
                 ShardIdentity::new(
                     ShardNumber((prng.next_u32() % shard_count) as u8),
                     ShardCount::new(shard_count as u8),
-                    DEFAULT_STRIPE_SIZE,
+                    ShardParameters::DEFAULT_STRIPE_SIZE,
                 )
                 .unwrap()
             };

libs/pageserver_api/src/models.rs

@@ -26,7 +26,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};
+use crate::shard::{ShardCount, ShardStripeSize, TenantShardId};
 
 /// The state of a tenant in this pageserver.
 ///
@@ -80,22 +80,10 @@ pub enum TenantState {
     ///
     /// Transitions out of this state are possible through `set_broken()`.
     Stopping {
-        /// The barrier can be used to wait for shutdown to complete. The first caller to set
-        /// Some(Barrier) is responsible for driving shutdown to completion. Subsequent callers
-        /// will wait for the first caller's existing barrier.
-        ///
-        /// None is set when an attach is cancelled, to signal to shutdown that the attach has in
-        /// fact cancelled:
-        ///
-        /// 1. `shutdown` sees `TenantState::Attaching`, and cancels the tenant.
-        /// 2. `attach` sets `TenantState::Stopping(None)` and exits.
-        /// 3. `set_stopping` waits for `TenantState::Stopping(None)` and sets
-        ///    `TenantState::Stopping(Some)` to claim the barrier as the shutdown owner.
-        //
         // Because of https://github.com/serde-rs/serde/issues/2105 this has to be a named field,
         // otherwise it will not be skipped during deserialization
         #[serde(skip)]
-        progress: Option<completion::Barrier>,
+        progress: completion::Barrier,
     },
     /// The tenant is recognized by the pageserver, but can no longer be used for
     /// any operations.
@@ -438,6 +426,8 @@ pub struct ShardParameters {
 }
 
 impl ShardParameters {
+    pub const DEFAULT_STRIPE_SIZE: ShardStripeSize = ShardStripeSize(256 * 1024 / 8);
+
     pub fn is_unsharded(&self) -> bool {
         self.count.is_unsharded()
     }
@@ -447,7 +437,7 @@ impl Default for ShardParameters {
     fn default() -> Self {
         Self {
             count: ShardCount::new(0),
-            stripe_size: DEFAULT_STRIPE_SIZE,
+            stripe_size: Self::DEFAULT_STRIPE_SIZE,
         }
     }
 }
@@ -526,8 +516,6 @@ pub struct TenantConfigPatch {
     #[serde(skip_serializing_if = "FieldPatch::is_noop")]
     pub compaction_algorithm: FieldPatch<CompactionAlgorithmSettings>,
     #[serde(skip_serializing_if = "FieldPatch::is_noop")]
-    pub compaction_shard_ancestor: FieldPatch<bool>,
-    #[serde(skip_serializing_if = "FieldPatch::is_noop")]
     pub compaction_l0_first: FieldPatch<bool>,
     #[serde(skip_serializing_if = "FieldPatch::is_noop")]
     pub compaction_l0_semaphore: FieldPatch<bool>,
@@ -578,8 +566,6 @@ pub struct TenantConfigPatch {
     #[serde(skip_serializing_if = "FieldPatch::is_noop")]
     pub gc_compaction_enabled: FieldPatch<bool>,
     #[serde(skip_serializing_if = "FieldPatch::is_noop")]
-    pub gc_compaction_verification: FieldPatch<bool>,
-    #[serde(skip_serializing_if = "FieldPatch::is_noop")]
     pub gc_compaction_initial_threshold_kb: FieldPatch<u64>,
     #[serde(skip_serializing_if = "FieldPatch::is_noop")]
     pub gc_compaction_ratio_percent: FieldPatch<u64>,
@@ -617,9 +603,6 @@ pub struct TenantConfig {
     #[serde(skip_serializing_if = "Option::is_none")]
     pub compaction_algorithm: Option<CompactionAlgorithmSettings>,
 
-    #[serde(skip_serializing_if = "Option::is_none")]
-    pub compaction_shard_ancestor: Option<bool>,
-
     #[serde(skip_serializing_if = "Option::is_none")]
     pub compaction_l0_first: Option<bool>,
 
@@ -703,9 +686,6 @@ pub struct TenantConfig {
     #[serde(skip_serializing_if = "Option::is_none")]
     pub gc_compaction_enabled: Option<bool>,
 
-    #[serde(skip_serializing_if = "Option::is_none")]
-    pub gc_compaction_verification: Option<bool>,
-
     #[serde(skip_serializing_if = "Option::is_none")]
     pub gc_compaction_initial_threshold_kb: Option<u64>,
 
@@ -729,7 +709,6 @@ impl TenantConfig {
             mut compaction_threshold,
             mut compaction_upper_limit,
             mut compaction_algorithm,
-            mut compaction_shard_ancestor,
             mut compaction_l0_first,
             mut compaction_l0_semaphore,
             mut l0_flush_delay_threshold,
@@ -755,7 +734,6 @@ impl TenantConfig {
             mut wal_receiver_protocol_override,
             mut rel_size_v2_enabled,
             mut gc_compaction_enabled,
-            mut gc_compaction_verification,
             mut gc_compaction_initial_threshold_kb,
             mut gc_compaction_ratio_percent,
             mut sampling_ratio,
@@ -778,9 +756,6 @@ impl TenantConfig {
             .compaction_upper_limit
             .apply(&mut compaction_upper_limit);
         patch.compaction_algorithm.apply(&mut compaction_algorithm);
-        patch
-            .compaction_shard_ancestor
-            .apply(&mut compaction_shard_ancestor);
         patch.compaction_l0_first.apply(&mut compaction_l0_first);
         patch
             .compaction_l0_semaphore
@@ -850,9 +825,6 @@ impl TenantConfig {
         patch
             .gc_compaction_enabled
             .apply(&mut gc_compaction_enabled);
-        patch
-            .gc_compaction_verification
-            .apply(&mut gc_compaction_verification);
         patch
             .gc_compaction_initial_threshold_kb
             .apply(&mut gc_compaction_initial_threshold_kb);
@@ -869,7 +841,6 @@ impl TenantConfig {
             compaction_threshold,
             compaction_upper_limit,
             compaction_algorithm,
-            compaction_shard_ancestor,
             compaction_l0_first,
             compaction_l0_semaphore,
             l0_flush_delay_threshold,
@@ -895,7 +866,6 @@ impl TenantConfig {
             wal_receiver_protocol_override,
             rel_size_v2_enabled,
             gc_compaction_enabled,
-            gc_compaction_verification,
             gc_compaction_initial_threshold_kb,
             gc_compaction_ratio_percent,
             sampling_ratio,
@@ -930,9 +900,6 @@ impl TenantConfig {
                 .as_ref()
                 .unwrap_or(&global_conf.compaction_algorithm)
                 .clone(),
-            compaction_shard_ancestor: self
-                .compaction_shard_ancestor
-                .unwrap_or(global_conf.compaction_shard_ancestor),
             compaction_l0_first: self
                 .compaction_l0_first
                 .unwrap_or(global_conf.compaction_l0_first),
@@ -997,9 +964,6 @@ impl TenantConfig {
             gc_compaction_enabled: self
                 .gc_compaction_enabled
                 .unwrap_or(global_conf.gc_compaction_enabled),
-            gc_compaction_verification: self
-                .gc_compaction_verification
-                .unwrap_or(global_conf.gc_compaction_verification),
             gc_compaction_initial_threshold_kb: self
                 .gc_compaction_initial_threshold_kb
                 .unwrap_or(global_conf.gc_compaction_initial_threshold_kb),
@@ -1704,7 +1668,6 @@ pub struct SecondaryProgress {
 pub struct TenantScanRemoteStorageShard {
     pub tenant_shard_id: TenantShardId,
     pub generation: Option<u32>,
-    pub stripe_size: Option<ShardStripeSize>,
 }
 
 #[derive(Serialize, Deserialize, Debug, Default)]
@@ -1817,34 +1780,8 @@ pub mod virtual_file {
     }
 
     impl IoMode {
-        pub fn preferred() -> Self {
-            // The default behavior when running Rust unit tests without any further
-            // flags is to use the newest behavior if available on the platform (Direct).
-            // The CI uses the following environment variable to unit tests for all
-            // different modes.
-            // NB: the Python regression & perf tests have their own defaults management
-            // that writes pageserver.toml; they do not use this variable.
-            if cfg!(test) {
-                use once_cell::sync::Lazy;
-                static CACHED: Lazy<IoMode> = Lazy::new(|| {
-                    utils::env::var_serde_json_string(
-                        "NEON_PAGESERVER_UNIT_TEST_VIRTUAL_FILE_IO_MODE",
-                    )
-                    .unwrap_or({
-                        #[cfg(target_os = "linux")]
-                        {
-                            IoMode::Direct
-                        }
-                        #[cfg(not(target_os = "linux"))]
-                        {
-                            IoMode::Buffered
-                        }
-                    })
-                });
-                *CACHED
-            } else {
-                IoMode::Buffered
-            }
+        pub const fn preferred() -> Self {
+            Self::Buffered
         }
     }
 
@@ -2782,15 +2719,10 @@ mod tests {
                 "Activating",
             ),
             (line!(), TenantState::Active, "Active"),
-            (
-                line!(),
-                TenantState::Stopping { progress: None },
-                "Stopping",
-            ),
             (
                 line!(),
                 TenantState::Stopping {
-                    progress: Some(completion::Barrier::default()),
+                    progress: utils::completion::Barrier::default(),
                 },
                 "Stopping",
             ),

libs/pageserver_api/src/record.rs

@@ -58,8 +58,6 @@ pub enum NeonWalRecord {
         /// to true. This record does not need the history WALs to reconstruct. See [`NeonWalRecord::will_init`] and
         /// its references in `timeline.rs`.
         will_init: bool,
-        /// Only append the record if the current image is the same as the one specified in this field.
-        only_if: Option<String>,
     },
 }
 
@@ -83,17 +81,6 @@ impl NeonWalRecord {
             append: s.as_ref().to_string(),
             clear: false,
             will_init: false,
-            only_if: None,
-        }
-    }
-
-    #[cfg(feature = "testing")]
-    pub fn wal_append_conditional(s: impl AsRef<str>, only_if: impl AsRef<str>) -> Self {
-        Self::Test {
-            append: s.as_ref().to_string(),
-            clear: false,
-            will_init: false,
-            only_if: Some(only_if.as_ref().to_string()),
         }
     }
 
@@ -103,7 +90,6 @@ impl NeonWalRecord {
             append: s.as_ref().to_string(),
             clear: true,
             will_init: false,
-            only_if: None,
         }
     }
 
@@ -113,7 +99,6 @@ impl NeonWalRecord {
             append: s.as_ref().to_string(),
             clear: true,
             will_init: true,
-            only_if: None,
         }
     }
 }

libs/pageserver_api/src/shard.rs

@@ -78,12 +78,6 @@ impl Default for ShardStripeSize {
     }
 }
 
-impl std::fmt::Display for ShardStripeSize {
-    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
-        self.0.fmt(f)
-    }
-}
-
 /// Layout version: for future upgrades where we might change how the key->shard mapping works
 #[derive(Clone, Copy, Serialize, Deserialize, Eq, PartialEq, Hash, Debug)]
 pub struct ShardLayout(u8);
@@ -92,11 +86,8 @@ const LAYOUT_V1: ShardLayout = ShardLayout(1);
 /// ShardIdentity uses a magic layout value to indicate if it is unusable
 const LAYOUT_BROKEN: ShardLayout = ShardLayout(255);
 
-/// The default stripe size in pages. 16 MiB divided by 8 kiB page size.
-///
-/// A lower stripe size distributes ingest load better across shards, but reduces IO amortization.
-/// 16 MiB appears to be a reasonable balance: <https://github.com/neondatabase/neon/pull/10510>.
-pub const DEFAULT_STRIPE_SIZE: ShardStripeSize = ShardStripeSize(16 * 1024 / 8);
+/// Default stripe size in pages: 256MiB divided by 8kiB page size.
+const DEFAULT_STRIPE_SIZE: ShardStripeSize = ShardStripeSize(256 * 1024 / 8);
 
 #[derive(thiserror::Error, Debug, PartialEq, Eq)]
 pub enum ShardConfigError {
@@ -546,7 +537,7 @@ mod tests {
             field6: 0x7d06,
         };
 
-        let shard = key_to_shard_number(ShardCount(10), ShardStripeSize(32768), &key);
+        let shard = key_to_shard_number(ShardCount(10), DEFAULT_STRIPE_SIZE, &key);
         assert_eq!(shard, ShardNumber(8));
     }
 

libs/postgres_backend/src/lib.rs

@@ -5,6 +5,7 @@
 #![deny(unsafe_code)]
 #![deny(clippy::undocumented_unsafe_blocks)]
 use std::future::Future;
+use std::io::ErrorKind;
 use std::net::SocketAddr;
 use std::os::fd::{AsRawFd, RawFd};
 use std::pin::Pin;
@@ -226,7 +227,7 @@ impl<IO: AsyncRead + AsyncWrite + Unpin> MaybeWriteOnly<IO> {
         match self {
             MaybeWriteOnly::Full(framed) => framed.read_startup_message().await,
             MaybeWriteOnly::WriteOnly(_) => {
-                Err(io::Error::other("reading from write only half").into())
+                Err(io::Error::new(ErrorKind::Other, "reading from write only half").into())
             }
             MaybeWriteOnly::Broken => panic!("IO on invalid MaybeWriteOnly"),
         }
@@ -236,7 +237,7 @@ impl<IO: AsyncRead + AsyncWrite + Unpin> MaybeWriteOnly<IO> {
         match self {
             MaybeWriteOnly::Full(framed) => framed.read_message().await,
             MaybeWriteOnly::WriteOnly(_) => {
-                Err(io::Error::other("reading from write only half").into())
+                Err(io::Error::new(ErrorKind::Other, "reading from write only half").into())
             }
             MaybeWriteOnly::Broken => panic!("IO on invalid MaybeWriteOnly"),
         }
@@ -974,7 +975,7 @@ impl<IO: AsyncRead + AsyncWrite + Unpin> AsyncWrite for CopyDataWriter<'_, IO> {
             .write_message_noflush(&BeMessage::CopyData(buf))
             // write_message only writes to the buffer, so it can fail iff the
             // message is invaid, but CopyData can't be invalid.
-            .map_err(|_| io::Error::other("failed to serialize CopyData"))?;
+            .map_err(|_| io::Error::new(ErrorKind::Other, "failed to serialize CopyData"))?;
 
         Poll::Ready(Ok(buf.len()))
     }

libs/postgres_backend/tests/simple_select.rs

@@ -85,8 +85,8 @@ static KEY: Lazy<rustls::pki_types::PrivateKeyDer<'static>> = Lazy::new(|| {
 
 static CERT: Lazy<rustls::pki_types::CertificateDer<'static>> = Lazy::new(|| {
     let mut cursor = Cursor::new(include_bytes!("cert.pem"));
-
-    rustls_pemfile::certs(&mut cursor).next().unwrap().unwrap()
+    let cert = rustls_pemfile::certs(&mut cursor).next().unwrap().unwrap();
+    cert
 });
 
 // test that basic select with ssl works

libs/pq_proto/src/framed.rs

@@ -35,7 +35,7 @@ impl ConnectionError {
     pub fn into_io_error(self) -> io::Error {
         match self {
             ConnectionError::Io(io) => io,
-            ConnectionError::Protocol(pe) => io::Error::other(pe.to_string()),
+            ConnectionError::Protocol(pe) => io::Error::new(io::ErrorKind::Other, pe.to_string()),
         }
     }
 }

libs/pq_proto/src/lib.rs

@@ -257,7 +257,7 @@ pub enum ProtocolError {
 impl ProtocolError {
     /// Proxy stream.rs uses only io::Error; provide it.
     pub fn into_io_error(self) -> io::Error {
-        io::Error::other(self.to_string())
+        io::Error::new(io::ErrorKind::Other, self.to_string())
     }
 }
 

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

@@ -212,7 +212,7 @@ impl ScramSha256 {
                     password,
                     channel_binding,
                 } => (nonce, password, channel_binding),
-                _ => return Err(io::Error::other("invalid SCRAM state")),
+                _ => return Err(io::Error::new(io::ErrorKind::Other, "invalid SCRAM state")),
             };
 
         let message =
@@ -291,7 +291,7 @@ impl ScramSha256 {
                 server_key,
                 auth_message,
             } => (server_key, auth_message),
-            _ => return Err(io::Error::other("invalid SCRAM state")),
+            _ => return Err(io::Error::new(io::ErrorKind::Other, "invalid SCRAM state")),
         };
 
         let message =
@@ -301,7 +301,10 @@ impl ScramSha256 {
 
         let verifier = match parsed {
             ServerFinalMessage::Error(e) => {
-                return Err(io::Error::other(format!("SCRAM error: {}", e)));
+                return Err(io::Error::new(
+                    io::ErrorKind::Other,
+                    format!("SCRAM error: {}", e),
+                ));
             }
             ServerFinalMessage::Verifier(verifier) => verifier,
         };

libs/remote_storage/Cargo.toml

@@ -28,7 +28,7 @@ toml_edit.workspace = true
 tracing.workspace = true
 scopeguard.workspace = true
 metrics.workspace = true
-utils = { path = "../utils", default-features = false }
+utils.workspace = true
 pin-project-lite.workspace = true
 
 azure_core.workspace = true

libs/remote_storage/src/azure_blob.rs

@@ -801,7 +801,8 @@ where
             // that support needs to be hacked in.
             //
             // including {self:?} into the message would be useful, but unsure how to unproject.
-            _ => std::task::Poll::Ready(Err(std::io::Error::other(
+            _ => std::task::Poll::Ready(Err(std::io::Error::new(
+                std::io::ErrorKind::Other,
                 "cloned or initial values cannot be read",
             ))),
         }
@@ -854,7 +855,7 @@ where
         };
         Err(azure_core::error::Error::new(
             azure_core::error::ErrorKind::Io,
-            std::io::Error::other(msg),
+            std::io::Error::new(std::io::ErrorKind::Other, msg),
         ))
     }
 

libs/utils/Cargo.toml

@@ -5,8 +5,7 @@ edition.workspace = true
 license.workspace = true
 
 [features]
-default = ["rename_noreplace"]
-rename_noreplace = []
+default = []
 # Enables test-only APIs, incuding failpoints. In particular, enables the `fail_point!` macro,
 # which adds some runtime cost to run tests on outage conditions
 testing = ["fail/failpoints"]
@@ -29,7 +28,6 @@ futures = { workspace = true }
 jsonwebtoken.workspace = true
 nix = { workspace = true, features = ["ioctl"] }
 once_cell.workspace = true
-pem.workspace = true
 pin-project-lite.workspace = true
 regex.workspace = true
 serde.workspace = true
@@ -37,7 +35,7 @@ serde_with.workspace = true
 serde_json.workspace = true
 signal-hook.workspace = true
 thiserror.workspace = true
-tokio = { workspace = true, features = ["signal"] }
+tokio.workspace = true
 tokio-tar.workspace = true
 tokio-util.workspace = true
 toml_edit = { workspace = true, features = ["serde"] }

libs/utils/src/auth.rs

@@ -11,8 +11,7 @@ use camino::Utf8Path;
 use jsonwebtoken::{
     Algorithm, DecodingKey, EncodingKey, Header, TokenData, Validation, decode, encode,
 };
-use pem::Pem;
-use serde::{Deserialize, Serialize, de::DeserializeOwned};
+use serde::{Deserialize, Serialize};
 
 use crate::id::TenantId;
 
@@ -74,10 +73,7 @@ impl SwappableJwtAuth {
     pub fn swap(&self, jwt_auth: JwtAuth) {
         self.0.swap(Arc::new(jwt_auth));
     }
-    pub fn decode<D: DeserializeOwned>(
-        &self,
-        token: &str,
-    ) -> std::result::Result<TokenData<D>, AuthError> {
+    pub fn decode(&self, token: &str) -> std::result::Result<TokenData<Claims>, AuthError> {
         self.0.load().decode(token)
     }
 }
@@ -152,10 +148,7 @@ impl JwtAuth {
     /// The function tries the stored decoding keys in succession,
     /// and returns the first yielding a successful result.
     /// If there is no working decoding key, it returns the last error.
-    pub fn decode<D: DeserializeOwned>(
-        &self,
-        token: &str,
-    ) -> std::result::Result<TokenData<D>, AuthError> {
+    pub fn decode(&self, token: &str) -> std::result::Result<TokenData<Claims>, AuthError> {
         let mut res = None;
         for decoding_key in &self.decoding_keys {
             res = Some(decode(token, decoding_key, &self.validation));
@@ -180,8 +173,8 @@ impl std::fmt::Debug for JwtAuth {
 }
 
 // this function is used only for testing purposes in CLI e g generate tokens during init
-pub fn encode_from_key_file<S: Serialize>(claims: &S, pem: &Pem) -> Result<String> {
-    let key = EncodingKey::from_ed_der(pem.contents());
+pub fn encode_from_key_file(claims: &Claims, key_data: &[u8]) -> Result<String> {
+    let key = EncodingKey::from_ed_pem(key_data)?;
     Ok(encode(&Header::new(STORAGE_TOKEN_ALGORITHM), claims, &key)?)
 }
 
@@ -195,13 +188,13 @@ mod tests {
     //
     // openssl genpkey -algorithm ed25519 -out ed25519-priv.pem
     // openssl pkey -in ed25519-priv.pem -pubout -out ed25519-pub.pem
-    const TEST_PUB_KEY_ED25519: &str = r#"
+    const TEST_PUB_KEY_ED25519: &[u8] = br#"
 -----BEGIN PUBLIC KEY-----
 MCowBQYDK2VwAyEARYwaNBayR+eGI0iXB4s3QxE3Nl2g1iWbr6KtLWeVD/w=
 -----END PUBLIC KEY-----
 "#;
 
-    const TEST_PRIV_KEY_ED25519: &str = r#"
+    const TEST_PRIV_KEY_ED25519: &[u8] = br#"
 -----BEGIN PRIVATE KEY-----
 MC4CAQAwBQYDK2VwBCIEID/Drmc1AA6U/znNRWpF3zEGegOATQxfkdWxitcOMsIH
 -----END PRIVATE KEY-----
@@ -229,9 +222,9 @@ MC4CAQAwBQYDK2VwBCIEID/Drmc1AA6U/znNRWpF3zEGegOATQxfkdWxitcOMsIH
 
         // Check it can be validated with the public key
         let auth = JwtAuth::new(vec![
-            DecodingKey::from_ed_pem(TEST_PUB_KEY_ED25519.as_bytes()).unwrap(),
+            DecodingKey::from_ed_pem(TEST_PUB_KEY_ED25519).unwrap(),
         ]);
-        let claims_from_token: Claims = auth.decode(encoded_eddsa).unwrap().claims;
+        let claims_from_token = auth.decode(encoded_eddsa).unwrap().claims;
         assert_eq!(claims_from_token, expected_claims);
     }
 
@@ -242,14 +235,13 @@ MC4CAQAwBQYDK2VwBCIEID/Drmc1AA6U/znNRWpF3zEGegOATQxfkdWxitcOMsIH
             scope: Scope::Tenant,
         };
 
-        let pem = pem::parse(TEST_PRIV_KEY_ED25519).unwrap();
-        let encoded = encode_from_key_file(&claims, &pem).unwrap();
+        let encoded = encode_from_key_file(&claims, TEST_PRIV_KEY_ED25519).unwrap();
 
         // decode it back
         let auth = JwtAuth::new(vec![
-            DecodingKey::from_ed_pem(TEST_PUB_KEY_ED25519.as_bytes()).unwrap(),
+            DecodingKey::from_ed_pem(TEST_PUB_KEY_ED25519).unwrap(),
         ]);
-        let decoded: TokenData<Claims> = auth.decode(&encoded).unwrap();
+        let decoded = auth.decode(&encoded).unwrap();
 
         assert_eq!(decoded.claims, claims);
     }

libs/utils/src/crashsafe.rs

@@ -81,9 +81,12 @@ pub fn path_with_suffix_extension(
 }
 
 pub fn fsync_file_and_parent(file_path: &Utf8Path) -> io::Result<()> {
-    let parent = file_path
-        .parent()
-        .ok_or_else(|| io::Error::other(format!("File {file_path:?} has no parent")))?;
+    let parent = file_path.parent().ok_or_else(|| {
+        io::Error::new(
+            io::ErrorKind::Other,
+            format!("File {file_path:?} has no parent"),
+        )
+    })?;
 
     fsync(file_path)?;
     fsync(parent)?;

libs/utils/src/fs_ext.rs

@@ -3,9 +3,7 @@ use std::{fs, io, path::Path};
 
 use anyhow::Context;
 
-#[cfg(feature = "rename_noreplace")]
 mod rename_noreplace;
-#[cfg(feature = "rename_noreplace")]
 pub use rename_noreplace::rename_noreplace;
 
 pub trait PathExt {

libs/utils/src/fs_ext/rename_noreplace.rs

@@ -8,7 +8,7 @@ pub fn rename_noreplace<P1: ?Sized + NixPath, P2: ?Sized + NixPath>(
     dst: &P2,
 ) -> nix::Result<()> {
     {
-        #[cfg(all(target_os = "linux", target_env = "gnu"))]
+        #[cfg(target_os = "linux")]
         {
             nix::fcntl::renameat2(
                 None,
@@ -29,7 +29,7 @@ pub fn rename_noreplace<P1: ?Sized + NixPath, P2: ?Sized + NixPath>(
             })??;
             nix::errno::Errno::result(res).map(drop)
         }
-        #[cfg(not(any(all(target_os = "linux", target_env = "gnu"), target_os = "macos")))]
+        #[cfg(not(any(target_os = "linux", target_os = "macos")))]
         {
             std::compile_error!("OS does not support no-replace renames");
         }

libs/utils/src/signals.rs

@@ -1,8 +1,6 @@
 pub use signal_hook::consts::TERM_SIGNALS;
 pub use signal_hook::consts::signal::*;
 use signal_hook::iterator::Signals;
-use tokio::signal::unix::{SignalKind, signal};
-use tracing::info;
 
 pub enum Signal {
     Quit,
@@ -38,30 +36,3 @@ impl ShutdownSignals {
         Ok(())
     }
 }
-
-/// Runs in a loop since we want to be responsive to multiple signals
-/// even after triggering shutdown (e.g. a SIGQUIT after a slow SIGTERM shutdown)
-/// <https://github.com/neondatabase/neon/issues/9740>
-pub async fn signal_handler(token: tokio_util::sync::CancellationToken) {
-    let mut sigint = signal(SignalKind::interrupt()).unwrap();
-    let mut sigterm = signal(SignalKind::terminate()).unwrap();
-    let mut sigquit = signal(SignalKind::quit()).unwrap();
-
-    loop {
-        let signal = tokio::select! {
-            _ = sigquit.recv() => {
-                info!("Got signal SIGQUIT. Terminating in immediate shutdown mode.");
-                std::process::exit(111);
-            }
-            _ = sigint.recv() => "SIGINT",
-            _ = sigterm.recv() => "SIGTERM",
-        };
-
-        if !token.is_cancelled() {
-            info!("Got signal {signal}. Terminating gracefully in fast shutdown mode.");
-            token.cancel();
-        } else {
-            info!("Got signal {signal}. Already shutting down.");
-        }
-    }
-}

object_storage/Cargo.toml

@@ -1,28 +0,0 @@
-[package]
-name = "object_storage"
-version = "0.0.1"
-edition.workspace = true
-license.workspace = true
-[dependencies]
-anyhow.workspace = true
-axum-extra.workspace = true
-axum.workspace = true
-camino.workspace = true
-futures.workspace = true
-jsonwebtoken.workspace = true
-prometheus.workspace = true
-remote_storage.workspace = true
-serde.workspace = true
-serde_json.workspace = true
-tokio-util.workspace = true
-tokio.workspace = true
-tracing.workspace = true
-utils = { path = "../libs/utils", default-features = false }
-workspace_hack.workspace = true
-[dev-dependencies]
-camino-tempfile.workspace = true
-http-body-util.workspace = true
-itertools.workspace = true
-rand.workspace = true
-test-log.workspace = true
-tower.workspace = true

object_storage/src/app.rs

@@ -1,561 +0,0 @@
-use anyhow::anyhow;
-use axum::body::{Body, Bytes};
-use axum::response::{IntoResponse, Response};
-use axum::{Router, http::StatusCode};
-use object_storage::{PrefixS3Path, S3Path, Storage, bad_request, internal_error, not_found, ok};
-use remote_storage::TimeoutOrCancel;
-use remote_storage::{DownloadError, DownloadOpts, GenericRemoteStorage, RemotePath};
-use std::{sync::Arc, time::SystemTime, time::UNIX_EPOCH};
-use tokio_util::sync::CancellationToken;
-use tracing::{error, info};
-use utils::backoff::retry;
-
-pub fn app(state: Arc<Storage>) -> Router<()> {
-    use axum::routing::{delete as _delete, get as _get};
-    let delete_prefix = _delete(delete_prefix);
-    Router::new()
-        .route(
-            "/{tenant_id}/{timeline_id}/{endpoint_id}/{*path}",
-            _get(get).put(set).delete(delete),
-        )
-        .route(
-            "/{tenant_id}/{timeline_id}/{endpoint_id}",
-            delete_prefix.clone(),
-        )
-        .route("/{tenant_id}/{timeline_id}", delete_prefix.clone())
-        .route("/{tenant_id}", delete_prefix)
-        .route("/metrics", _get(metrics))
-        .route("/status", _get(async || StatusCode::OK.into_response()))
-        .with_state(state)
-}
-
-type Result = anyhow::Result<Response, Response>;
-type State = axum::extract::State<Arc<Storage>>;
-
-const CONTENT_TYPE: &str = "content-type";
-const APPLICATION_OCTET_STREAM: &str = "application/octet-stream";
-const WARN_THRESHOLD: u32 = 3;
-const MAX_RETRIES: u32 = 10;
-
-async fn metrics() -> Result {
-    prometheus::TextEncoder::new()
-        .encode_to_string(&prometheus::gather())
-        .map(|s| s.into_response())
-        .map_err(|e| internal_error(e, "/metrics", "collecting metrics"))
-}
-
-async fn get(S3Path { path }: S3Path, state: State) -> Result {
-    info!(%path, "downloading");
-    let download_err = |e| {
-        if let DownloadError::NotFound = e {
-            info!(%path, %e, "downloading"); // 404 is not an issue of _this_ service
-            return not_found(&path);
-        }
-        internal_error(e, &path, "downloading")
-    };
-    let cancel = state.cancel.clone();
-    let opts = &DownloadOpts::default();
-
-    let stream = retry(
-        async || state.storage.download(&path, opts, &cancel).await,
-        DownloadError::is_permanent,
-        WARN_THRESHOLD,
-        MAX_RETRIES,
-        "downloading",
-        &cancel,
-    )
-    .await
-    .unwrap_or(Err(DownloadError::Cancelled))
-    .map_err(download_err)?
-    .download_stream;
-
-    Response::builder()
-        .status(StatusCode::OK)
-        .header(CONTENT_TYPE, APPLICATION_OCTET_STREAM)
-        .body(Body::from_stream(stream))
-        .map_err(|e| internal_error(e, path, "reading response"))
-}
-
-// Best solution for files is multipart upload, but remote_storage doesn't support it,
-// so we can either read Bytes in memory and push at once or forward BodyDataStream to
-// remote_storage. The latter may seem more peformant, but BodyDataStream doesn't have a
-// guaranteed size() which may produce issues while uploading to s3.
-// So, currently we're going with an in-memory copy plus a boundary to prevent uploading
-// very large files.
-async fn set(S3Path { path }: S3Path, state: State, bytes: Bytes) -> Result {
-    info!(%path, "uploading");
-    let request_len = bytes.len();
-    let max_len = state.max_upload_file_limit;
-    if request_len > max_len {
-        return Err(bad_request(
-            anyhow!("File size {request_len} exceeds max {max_len}"),
-            "uploading",
-        ));
-    }
-
-    let cancel = state.cancel.clone();
-    let fun = async || {
-        let stream = bytes_to_stream(bytes.clone());
-        state
-            .storage
-            .upload(stream, request_len, &path, None, &cancel)
-            .await
-    };
-    retry(
-        fun,
-        TimeoutOrCancel::caused_by_cancel,
-        WARN_THRESHOLD,
-        MAX_RETRIES,
-        "uploading",
-        &cancel,
-    )
-    .await
-    .unwrap_or(Err(anyhow!("uploading cancelled")))
-    .map_err(|e| internal_error(e, path, "reading response"))?;
-    Ok(ok())
-}
-
-async fn delete(S3Path { path }: S3Path, state: State) -> Result {
-    info!(%path, "deleting");
-    let cancel = state.cancel.clone();
-    retry(
-        async || state.storage.delete(&path, &cancel).await,
-        TimeoutOrCancel::caused_by_cancel,
-        WARN_THRESHOLD,
-        MAX_RETRIES,
-        "deleting",
-        &cancel,
-    )
-    .await
-    .unwrap_or(Err(anyhow!("deleting cancelled")))
-    .map_err(|e| internal_error(e, path, "deleting"))?;
-    Ok(ok())
-}
-
-async fn delete_prefix(PrefixS3Path { path }: PrefixS3Path, state: State) -> Result {
-    info!(%path, "deleting prefix");
-    let cancel = state.cancel.clone();
-    retry(
-        async || state.storage.delete_prefix(&path, &cancel).await,
-        TimeoutOrCancel::caused_by_cancel,
-        WARN_THRESHOLD,
-        MAX_RETRIES,
-        "deleting prefix",
-        &cancel,
-    )
-    .await
-    .unwrap_or(Err(anyhow!("deleting prefix cancelled")))
-    .map_err(|e| internal_error(e, path, "deleting prefix"))?;
-    Ok(ok())
-}
-
-pub async fn check_storage_permissions(
-    client: &GenericRemoteStorage,
-    cancel: CancellationToken,
-) -> anyhow::Result<()> {
-    info!("storage permissions check");
-
-    // as_nanos() as multiple instances proxying same bucket may be started at once
-    let now = SystemTime::now()
-        .duration_since(UNIX_EPOCH)?
-        .as_nanos()
-        .to_string();
-
-    let path = RemotePath::from_string(&format!("write_access_{now}"))?;
-    info!(%path, "uploading");
-
-    let body = now.to_string();
-    let stream = bytes_to_stream(Bytes::from(body.clone()));
-    client
-        .upload(stream, body.len(), &path, None, &cancel)
-        .await?;
-
-    use tokio::io::AsyncReadExt;
-    info!(%path, "downloading");
-    let download_opts = DownloadOpts {
-        kind: remote_storage::DownloadKind::Small,
-        ..Default::default()
-    };
-    let mut body_read_buf = Vec::new();
-    let stream = client
-        .download(&path, &download_opts, &cancel)
-        .await?
-        .download_stream;
-    tokio_util::io::StreamReader::new(stream)
-        .read_to_end(&mut body_read_buf)
-        .await?;
-    let body_read = String::from_utf8(body_read_buf)?;
-    if body != body_read {
-        error!(%body, %body_read, "File contents do not match");
-        anyhow::bail!("Read back file doesn't match original")
-    }
-
-    info!(%path, "removing");
-    client.delete(&path, &cancel).await
-}
-
-fn bytes_to_stream(bytes: Bytes) -> impl futures::Stream<Item = std::io::Result<Bytes>> {
-    futures::stream::once(futures::future::ready(Ok(bytes)))
-}
-
-#[cfg(test)]
-mod tests {
-    use super::*;
-    use axum::{body::Body, extract::Request, response::Response};
-    use http_body_util::BodyExt;
-    use itertools::iproduct;
-    use std::env::var;
-    use std::sync::Arc;
-    use std::time::Duration;
-    use test_log::test as testlog;
-    use tower::{Service, util::ServiceExt};
-    use utils::id::{TenantId, TimelineId};
-
-    // see libs/remote_storage/tests/test_real_s3.rs
-    const REAL_S3_ENV: &str = "ENABLE_REAL_S3_REMOTE_STORAGE";
-    const REAL_S3_BUCKET: &str = "REMOTE_STORAGE_S3_BUCKET";
-    const REAL_S3_REGION: &str = "REMOTE_STORAGE_S3_REGION";
-
-    async fn proxy() -> (Storage, Option<camino_tempfile::Utf8TempDir>) {
-        let cancel = CancellationToken::new();
-        let (dir, storage) = if var(REAL_S3_ENV).is_err() {
-            // tests execute in parallel and we need a new directory for each of them
-            let dir = camino_tempfile::tempdir().unwrap();
-            let fs =
-                remote_storage::LocalFs::new(dir.path().into(), Duration::from_secs(5)).unwrap();
-            (Some(dir), GenericRemoteStorage::LocalFs(fs))
-        } else {
-            // test_real_s3::create_s3_client is hard to reference, reimplementing here
-            let millis = SystemTime::now()
-                .duration_since(UNIX_EPOCH)
-                .unwrap()
-                .as_millis();
-            use rand::Rng;
-            let random = rand::thread_rng().r#gen::<u32>();
-
-            let s3_config = remote_storage::S3Config {
-                bucket_name: var(REAL_S3_BUCKET).unwrap(),
-                bucket_region: var(REAL_S3_REGION).unwrap(),
-                prefix_in_bucket: Some(format!("test_{millis}_{random:08x}/")),
-                endpoint: None,
-                concurrency_limit: std::num::NonZeroUsize::new(100).unwrap(),
-                max_keys_per_list_response: None,
-                upload_storage_class: None,
-            };
-            let bucket = remote_storage::S3Bucket::new(&s3_config, Duration::from_secs(1))
-                .await
-                .unwrap();
-            (None, GenericRemoteStorage::AwsS3(Arc::new(bucket)))
-        };
-
-        let proxy = Storage {
-            auth: object_storage::JwtAuth::new(TEST_PUB_KEY_ED25519).unwrap(),
-            storage,
-            cancel: cancel.clone(),
-            max_upload_file_limit: usize::MAX,
-        };
-        check_storage_permissions(&proxy.storage, cancel)
-            .await
-            .unwrap();
-        (proxy, dir)
-    }
-
-    // see libs/utils/src/auth.rs
-    const TEST_PUB_KEY_ED25519: &[u8] = b"
------BEGIN PUBLIC KEY-----
-MCowBQYDK2VwAyEARYwaNBayR+eGI0iXB4s3QxE3Nl2g1iWbr6KtLWeVD/w=
------END PUBLIC KEY-----
-";
-
-    const TEST_PRIV_KEY_ED25519: &[u8] = br#"
------BEGIN PRIVATE KEY-----
-MC4CAQAwBQYDK2VwBCIEID/Drmc1AA6U/znNRWpF3zEGegOATQxfkdWxitcOMsIH
------END PRIVATE KEY-----
-"#;
-
-    async fn request(req: Request<Body>) -> Response<Body> {
-        let (proxy, _) = proxy().await;
-        app(Arc::new(proxy))
-            .into_service()
-            .oneshot(req)
-            .await
-            .unwrap()
-    }
-
-    #[testlog(tokio::test)]
-    async fn status() {
-        let res = Request::builder()
-            .uri("/status")
-            .body(Body::empty())
-            .map(request)
-            .unwrap()
-            .await;
-        assert_eq!(res.status(), StatusCode::OK);
-    }
-
-    fn routes() -> impl Iterator<Item = (&'static str, &'static str)> {
-        iproduct!(
-            vec!["/1", "/1/2", "/1/2/3", "/1/2/3/4"],
-            vec!["GET", "PUT", "DELETE"]
-        )
-    }
-
-    #[testlog(tokio::test)]
-    async fn no_token() {
-        for (uri, method) in routes() {
-            info!(%uri, %method);
-            let res = Request::builder()
-                .uri(uri)
-                .method(method)
-                .body(Body::empty())
-                .map(request)
-                .unwrap()
-                .await;
-            assert!(matches!(
-                res.status(),
-                StatusCode::METHOD_NOT_ALLOWED | StatusCode::BAD_REQUEST
-            ));
-        }
-    }
-
-    #[testlog(tokio::test)]
-    async fn invalid_token() {
-        for (uri, method) in routes() {
-            info!(%uri, %method);
-            let status = Request::builder()
-                .uri(uri)
-                .header("Authorization", "Bearer 123")
-                .method(method)
-                .body(Body::empty())
-                .map(request)
-                .unwrap()
-                .await;
-            assert!(matches!(
-                status.status(),
-                StatusCode::METHOD_NOT_ALLOWED | StatusCode::BAD_REQUEST
-            ));
-        }
-    }
-
-    const TENANT_ID: TenantId =
-        TenantId::from_array([0, 1, 2, 3, 4, 5, 6, 7, 8, 9, 1, 2, 3, 4, 5, 6]);
-    const TIMELINE_ID: TimelineId =
-        TimelineId::from_array([0, 1, 2, 3, 4, 5, 6, 7, 8, 9, 1, 2, 3, 4, 5, 7]);
-    const ENDPOINT_ID: &str = "ep-winter-frost-a662z3vg";
-    fn token() -> String {
-        let claims = object_storage::Claims {
-            tenant_id: TENANT_ID,
-            timeline_id: TIMELINE_ID,
-            endpoint_id: ENDPOINT_ID.into(),
-            exp: u64::MAX,
-        };
-        let key = jsonwebtoken::EncodingKey::from_ed_pem(TEST_PRIV_KEY_ED25519).unwrap();
-        let header = jsonwebtoken::Header::new(object_storage::VALIDATION_ALGO);
-        jsonwebtoken::encode(&header, &claims, &key).unwrap()
-    }
-
-    #[testlog(tokio::test)]
-    async fn unauthorized() {
-        let (proxy, _) = proxy().await;
-        let mut app = app(Arc::new(proxy)).into_service();
-        let token = token();
-        let args = itertools::iproduct!(
-            vec![TENANT_ID.to_string(), TenantId::generate().to_string()],
-            vec![TIMELINE_ID.to_string(), TimelineId::generate().to_string()],
-            vec![ENDPOINT_ID, "ep-ololo"]
-        )
-        .skip(1);
-
-        for ((uri, method), (tenant, timeline, endpoint)) in iproduct!(routes(), args) {
-            info!(%uri, %method, %tenant, %timeline, %endpoint);
-            let request = Request::builder()
-                .uri(format!("/{tenant}/{timeline}/{endpoint}/sub/path/key"))
-                .method(method)
-                .header("Authorization", format!("Bearer {}", token))
-                .body(Body::empty())
-                .unwrap();
-            let status = ServiceExt::ready(&mut app)
-                .await
-                .unwrap()
-                .call(request)
-                .await
-                .unwrap()
-                .status();
-            assert_eq!(status, StatusCode::UNAUTHORIZED);
-        }
-    }
-
-    #[testlog(tokio::test)]
-    async fn method_not_allowed() {
-        let token = token();
-        let iter = iproduct!(vec!["", "/.."], vec!["GET", "PUT"]);
-        for (key, method) in iter {
-            let status = Request::builder()
-                .uri(format!("/{TENANT_ID}/{TIMELINE_ID}/{ENDPOINT_ID}{key}"))
-                .method(method)
-                .header("Authorization", format!("Bearer {token}"))
-                .body(Body::empty())
-                .map(request)
-                .unwrap()
-                .await
-                .status();
-            assert!(matches!(
-                status,
-                StatusCode::BAD_REQUEST | StatusCode::METHOD_NOT_ALLOWED
-            ));
-        }
-    }
-
-    async fn requests_chain(
-        chain: impl Iterator<Item = (String, &str, &'static str, StatusCode, bool)>,
-        token: impl Fn(&str) -> String,
-    ) {
-        let (proxy, _) = proxy().await;
-        let mut app = app(Arc::new(proxy)).into_service();
-        for (uri, method, body, expected_status, compare_body) in chain {
-            info!(%uri, %method, %body, %expected_status);
-            let bearer = format!("Bearer {}", token(&uri));
-            let request = Request::builder()
-                .uri(uri)
-                .method(method)
-                .header("Authorization", &bearer)
-                .body(Body::from(body))
-                .unwrap();
-            let response = ServiceExt::ready(&mut app)
-                .await
-                .unwrap()
-                .call(request)
-                .await
-                .unwrap();
-            assert_eq!(response.status(), expected_status);
-            if !compare_body {
-                continue;
-            }
-            let read_body = response.into_body().collect().await.unwrap().to_bytes();
-            assert_eq!(body, read_body);
-        }
-    }
-
-    #[testlog(tokio::test)]
-    async fn metrics() {
-        let uri = format!("/{TENANT_ID}/{TIMELINE_ID}/{ENDPOINT_ID}/key");
-        let req = vec![
-            (uri.clone(), "PUT", "body", StatusCode::OK, false),
-            (uri.clone(), "DELETE", "", StatusCode::OK, false),
-        ];
-        requests_chain(req.into_iter(), |_| token()).await;
-
-        let res = Request::builder()
-            .uri("/metrics")
-            .body(Body::empty())
-            .map(request)
-            .unwrap()
-            .await;
-        assert_eq!(res.status(), StatusCode::OK);
-        let body = res.into_body().collect().await.unwrap().to_bytes();
-        let body = String::from_utf8_lossy(&body);
-        tracing::debug!(%body);
-        // Storage metrics are not gathered for LocalFs
-        if var(REAL_S3_ENV).is_ok() {
-            assert!(body.contains("remote_storage_s3_deleted_objects_total"));
-        }
-        assert!(body.contains("process_threads"));
-    }
-
-    #[testlog(tokio::test)]
-    async fn insert_retrieve_remove() {
-        let uri = format!("/{TENANT_ID}/{TIMELINE_ID}/{ENDPOINT_ID}/key");
-        let chain = vec![
-            (uri.clone(), "GET", "", StatusCode::NOT_FOUND, false),
-            (uri.clone(), "PUT", "пыщьпыщь", StatusCode::OK, false),
-            (uri.clone(), "GET", "пыщьпыщь", StatusCode::OK, true),
-            (uri.clone(), "DELETE", "", StatusCode::OK, false),
-            (uri, "GET", "", StatusCode::NOT_FOUND, false),
-        ];
-        requests_chain(chain.into_iter(), |_| token()).await;
-    }
-
-    fn delete_prefix_token(uri: &str) -> String {
-        use serde::Serialize;
-        let parts = uri.split("/").collect::<Vec<&str>>();
-        #[derive(Serialize)]
-        struct PrefixClaims {
-            tenant_id: TenantId,
-            timeline_id: Option<TimelineId>,
-            endpoint_id: Option<object_storage::EndpointId>,
-            exp: u64,
-        }
-        let claims = PrefixClaims {
-            tenant_id: parts.get(1).map(|c| c.parse().unwrap()).unwrap(),
-            timeline_id: parts.get(2).map(|c| c.parse().unwrap()),
-            endpoint_id: parts.get(3).map(ToString::to_string),
-            exp: u64::MAX,
-        };
-        let key = jsonwebtoken::EncodingKey::from_ed_pem(TEST_PRIV_KEY_ED25519).unwrap();
-        let header = jsonwebtoken::Header::new(object_storage::VALIDATION_ALGO);
-        jsonwebtoken::encode(&header, &claims, &key).unwrap()
-    }
-
-    // Can't use single digit numbers as they won't be validated as TimelineId and EndpointId
-    #[testlog(tokio::test)]
-    async fn delete_prefix() {
-        let tenant_id =
-            TenantId::from_array([1, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0]).to_string();
-        let t2 = TimelineId::from_array([2, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0]);
-        let t3 = TimelineId::from_array([3, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0]);
-        let t4 = TimelineId::from_array([4, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0]);
-        let f = |timeline, path| format!("/{tenant_id}/{timeline}{path}");
-        // Why extra slash in string literals? Axum is weird with URIs:
-        // /1/2 and 1/2/ match different routes, thus first yields OK and second NOT_FOUND
-        //  as it matches /tenant/timeline/endpoint, see https://stackoverflow.com/a/75355932
-        // The cost of removing trailing slash is suprisingly hard:
-        // * Add tower dependency with NormalizePath layer
-        // * wrap Router<()> in this layer https://github.com/tokio-rs/axum/discussions/2377
-        // * Rewrite make_service() -> into_make_service()
-        // * Rewrite oneshot() (not available for NormalizePath)
-        // I didn't manage to get it working correctly
-        let chain = vec![
-            // create 1/2/3/4, 1/2/3/5, delete prefix 1/2/3 -> empty
-            (f(t2, "/3/4"), "PUT", "", StatusCode::OK, false),
-            (f(t2, "/3/4"), "PUT", "", StatusCode::OK, false), // we can override file contents
-            (f(t2, "/3/5"), "PUT", "", StatusCode::OK, false),
-            (f(t2, "/3"), "DELETE", "", StatusCode::OK, false),
-            (f(t2, "/3/4"), "GET", "", StatusCode::NOT_FOUND, false),
-            (f(t2, "/3/5"), "GET", "", StatusCode::NOT_FOUND, false),
-            // create 1/2/3/4, 1/2/5/6, delete prefix 1/2/3 -> 1/2/5/6
-            (f(t2, "/3/4"), "PUT", "", StatusCode::OK, false),
-            (f(t2, "/5/6"), "PUT", "", StatusCode::OK, false),
-            (f(t2, "/3"), "DELETE", "", StatusCode::OK, false),
-            (f(t2, "/3/4"), "GET", "", StatusCode::NOT_FOUND, false),
-            (f(t2, "/5/6"), "GET", "", StatusCode::OK, false),
-            // create 1/2/3/4, 1/2/7/8, delete prefix 1/2 -> empty
-            (f(t2, "/3/4"), "PUT", "", StatusCode::OK, false),
-            (f(t2, "/7/8"), "PUT", "", StatusCode::OK, false),
-            (f(t2, ""), "DELETE", "", StatusCode::OK, false),
-            (f(t2, "/3/4"), "GET", "", StatusCode::NOT_FOUND, false),
-            (f(t2, "/7/8"), "GET", "", StatusCode::NOT_FOUND, false),
-            // create 1/2/3/4, 1/2/5/6, 1/3/8/9, delete prefix 1/2/3 -> 1/2/5/6, 1/3/8/9
-            (f(t2, "/3/4"), "PUT", "", StatusCode::OK, false),
-            (f(t2, "/5/6"), "PUT", "", StatusCode::OK, false),
-            (f(t3, "/8/9"), "PUT", "", StatusCode::OK, false),
-            (f(t2, "/3"), "DELETE", "", StatusCode::OK, false),
-            (f(t2, "/3/4"), "GET", "", StatusCode::NOT_FOUND, false),
-            (f(t2, "/5/6"), "GET", "", StatusCode::OK, false),
-            (f(t3, "/8/9"), "GET", "", StatusCode::OK, false),
-            // create 1/4/5/6, delete prefix 1/2 -> 1/3/8/9, 1/4/5/6
-            (f(t4, "/5/6"), "PUT", "", StatusCode::OK, false),
-            (f(t2, ""), "DELETE", "", StatusCode::OK, false),
-            (f(t2, "/3/4"), "GET", "", StatusCode::NOT_FOUND, false),
-            (f(t2, "/5/6"), "GET", "", StatusCode::NOT_FOUND, false),
-            (f(t3, "/8/9"), "GET", "", StatusCode::OK, false),
-            (f(t4, "/5/6"), "GET", "", StatusCode::OK, false),
-            // delete prefix 1 -> empty
-            (format!("/{tenant_id}"), "DELETE", "", StatusCode::OK, false),
-            (f(t2, "/3/4"), "GET", "", StatusCode::NOT_FOUND, false),
-            (f(t2, "/5/6"), "GET", "", StatusCode::NOT_FOUND, false),
-            (f(t3, "/8/9"), "GET", "", StatusCode::NOT_FOUND, false),
-            (f(t4, "/5/6"), "GET", "", StatusCode::NOT_FOUND, false),
-        ];
-        requests_chain(chain.into_iter(), delete_prefix_token).await;
-    }
-}

object_storage/src/lib.rs

@@ -1,344 +0,0 @@
-use anyhow::Result;
-use axum::extract::{FromRequestParts, Path};
-use axum::response::{IntoResponse, Response};
-use axum::{RequestPartsExt, http::StatusCode, http::request::Parts};
-use axum_extra::TypedHeader;
-use axum_extra::headers::{Authorization, authorization::Bearer};
-use camino::Utf8PathBuf;
-use jsonwebtoken::{DecodingKey, Validation};
-use remote_storage::{GenericRemoteStorage, RemotePath};
-use serde::{Deserialize, Serialize};
-use std::fmt::Display;
-use std::result::Result as StdResult;
-use std::sync::Arc;
-use tokio_util::sync::CancellationToken;
-use tracing::{debug, error};
-use utils::id::{TenantId, TimelineId};
-
-// simplified version of utils::auth::JwtAuth
-pub struct JwtAuth {
-    decoding_key: DecodingKey,
-    validation: Validation,
-}
-
-pub const VALIDATION_ALGO: jsonwebtoken::Algorithm = jsonwebtoken::Algorithm::EdDSA;
-impl JwtAuth {
-    pub fn new(key: &[u8]) -> Result<Self> {
-        Ok(Self {
-            decoding_key: DecodingKey::from_ed_pem(key)?,
-            validation: Validation::new(VALIDATION_ALGO),
-        })
-    }
-
-    pub fn decode<T: serde::de::DeserializeOwned>(&self, token: &str) -> Result<T> {
-        Ok(jsonwebtoken::decode(token, &self.decoding_key, &self.validation).map(|t| t.claims)?)
-    }
-}
-
-fn normalize_key(key: &str) -> StdResult<Utf8PathBuf, String> {
-    let key = clean_utf8(&Utf8PathBuf::from(key));
-    if key.starts_with("..") || key == "." || key == "/" {
-        return Err(format!("invalid key {key}"));
-    }
-    match key.strip_prefix("/").map(Utf8PathBuf::from) {
-        Ok(p) => Ok(p),
-        _ => Ok(key),
-    }
-}
-
-// Copied from path_clean crate with PathBuf->Utf8PathBuf
-fn clean_utf8(path: &camino::Utf8Path) -> Utf8PathBuf {
-    use camino::Utf8Component as Comp;
-    let mut out = Vec::new();
-    for comp in path.components() {
-        match comp {
-            Comp::CurDir => (),
-            Comp::ParentDir => match out.last() {
-                Some(Comp::RootDir) => (),
-                Some(Comp::Normal(_)) => {
-                    out.pop();
-                }
-                None | Some(Comp::CurDir) | Some(Comp::ParentDir) | Some(Comp::Prefix(_)) => {
-                    out.push(comp)
-                }
-            },
-            comp => out.push(comp),
-        }
-    }
-    if !out.is_empty() {
-        out.iter().collect()
-    } else {
-        Utf8PathBuf::from(".")
-    }
-}
-
-pub struct Storage {
-    pub auth: JwtAuth,
-    pub storage: GenericRemoteStorage,
-    pub cancel: CancellationToken,
-    pub max_upload_file_limit: usize,
-}
-
-pub type EndpointId = String; // If needed, reuse small string from proxy/src/types.rc
-
-#[derive(Deserialize, Serialize, PartialEq)]
-pub struct Claims {
-    pub tenant_id: TenantId,
-    pub timeline_id: TimelineId,
-    pub endpoint_id: EndpointId,
-    pub exp: u64,
-}
-
-impl Display for Claims {
-    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
-        write!(
-            f,
-            "Claims(tenant_id {} timeline_id {} endpoint_id {} exp {})",
-            self.tenant_id, self.timeline_id, self.endpoint_id, self.exp
-        )
-    }
-}
-
-#[derive(Deserialize, Serialize)]
-struct KeyRequest {
-    tenant_id: TenantId,
-    timeline_id: TimelineId,
-    endpoint_id: EndpointId,
-    path: String,
-}
-
-#[derive(Debug, PartialEq)]
-pub struct S3Path {
-    pub path: RemotePath,
-}
-
-impl TryFrom<&KeyRequest> for S3Path {
-    type Error = String;
-    fn try_from(req: &KeyRequest) -> StdResult<Self, Self::Error> {
-        let KeyRequest {
-            tenant_id,
-            timeline_id,
-            endpoint_id,
-            path,
-        } = &req;
-        let prefix = format!("{tenant_id}/{timeline_id}/{endpoint_id}",);
-        let path = Utf8PathBuf::from(prefix).join(normalize_key(path)?);
-        let path = RemotePath::new(&path).unwrap(); // unwrap() because the path is already relative
-        Ok(S3Path { path })
-    }
-}
-
-fn unauthorized(route: impl Display, claims: impl Display) -> Response {
-    debug!(%route, %claims, "route doesn't match claims");
-    StatusCode::UNAUTHORIZED.into_response()
-}
-
-pub fn bad_request(err: impl Display, desc: &'static str) -> Response {
-    debug!(%err, desc);
-    (StatusCode::BAD_REQUEST, err.to_string()).into_response()
-}
-
-pub fn ok() -> Response {
-    StatusCode::OK.into_response()
-}
-
-pub fn internal_error(err: impl Display, path: impl Display, desc: &'static str) -> Response {
-    error!(%err, %path, desc);
-    StatusCode::INTERNAL_SERVER_ERROR.into_response()
-}
-
-pub fn not_found(key: impl ToString) -> Response {
-    (StatusCode::NOT_FOUND, key.to_string()).into_response()
-}
-
-impl FromRequestParts<Arc<Storage>> for S3Path {
-    type Rejection = Response;
-    async fn from_request_parts(
-        parts: &mut Parts,
-        state: &Arc<Storage>,
-    ) -> Result<Self, Self::Rejection> {
-        let Path(path): Path<KeyRequest> = parts
-            .extract()
-            .await
-            .map_err(|e| bad_request(e, "invalid route"))?;
-        let TypedHeader(Authorization(bearer)) = parts
-            .extract::<TypedHeader<Authorization<Bearer>>>()
-            .await
-            .map_err(|e| bad_request(e, "invalid token"))?;
-        let claims: Claims = state
-            .auth
-            .decode(bearer.token())
-            .map_err(|e| bad_request(e, "decoding token"))?;
-        let route = Claims {
-            tenant_id: path.tenant_id,
-            timeline_id: path.timeline_id,
-            endpoint_id: path.endpoint_id.clone(),
-            exp: claims.exp,
-        };
-        if route != claims {
-            return Err(unauthorized(route, claims));
-        }
-        (&path)
-            .try_into()
-            .map_err(|e| bad_request(e, "invalid route"))
-    }
-}
-
-#[derive(Deserialize, Serialize, PartialEq)]
-pub struct PrefixKeyPath {
-    pub tenant_id: TenantId,
-    pub timeline_id: Option<TimelineId>,
-    pub endpoint_id: Option<EndpointId>,
-}
-
-impl Display for PrefixKeyPath {
-    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
-        write!(
-            f,
-            "PrefixKeyPath(tenant_id {} timeline_id {} endpoint_id {})",
-            self.tenant_id,
-            self.timeline_id
-                .as_ref()
-                .map(ToString::to_string)
-                .unwrap_or("".to_string()),
-            self.endpoint_id
-                .as_ref()
-                .map(ToString::to_string)
-                .unwrap_or("".to_string())
-        )
-    }
-}
-
-#[derive(Debug, PartialEq)]
-pub struct PrefixS3Path {
-    pub path: RemotePath,
-}
-
-impl From<&PrefixKeyPath> for PrefixS3Path {
-    fn from(path: &PrefixKeyPath) -> Self {
-        let timeline_id = path
-            .timeline_id
-            .as_ref()
-            .map(ToString::to_string)
-            .unwrap_or("".to_string());
-        let endpoint_id = path
-            .endpoint_id
-            .as_ref()
-            .map(ToString::to_string)
-            .unwrap_or("".to_string());
-        let path = Utf8PathBuf::from(path.tenant_id.to_string())
-            .join(timeline_id)
-            .join(endpoint_id);
-        let path = RemotePath::new(&path).unwrap(); // unwrap() because the path is already relative
-        PrefixS3Path { path }
-    }
-}
-
-impl FromRequestParts<Arc<Storage>> for PrefixS3Path {
-    type Rejection = Response;
-    async fn from_request_parts(
-        parts: &mut Parts,
-        state: &Arc<Storage>,
-    ) -> Result<Self, Self::Rejection> {
-        let Path(path) = parts
-            .extract::<Path<PrefixKeyPath>>()
-            .await
-            .map_err(|e| bad_request(e, "invalid route"))?;
-        let TypedHeader(Authorization(bearer)) = parts
-            .extract::<TypedHeader<Authorization<Bearer>>>()
-            .await
-            .map_err(|e| bad_request(e, "invalid token"))?;
-        let claims: PrefixKeyPath = state
-            .auth
-            .decode(bearer.token())
-            .map_err(|e| bad_request(e, "invalid token"))?;
-        if path != claims {
-            return Err(unauthorized(path, claims));
-        }
-        Ok((&path).into())
-    }
-}
-
-#[cfg(test)]
-mod tests {
-    use super::*;
-
-    #[test]
-    fn normalize_key() {
-        let f = super::normalize_key;
-        assert_eq!(f("hello/world/..").unwrap(), Utf8PathBuf::from("hello"));
-        assert_eq!(
-            f("ololo/1/../../not_ololo").unwrap(),
-            Utf8PathBuf::from("not_ololo")
-        );
-        assert!(f("ololo/1/../../../").is_err());
-        assert!(f(".").is_err());
-        assert!(f("../").is_err());
-        assert!(f("").is_err());
-        assert_eq!(f("/1/2/3").unwrap(), Utf8PathBuf::from("1/2/3"));
-        assert!(f("/1/2/3/../../../").is_err());
-        assert!(f("/1/2/3/../../../../").is_err());
-    }
-
-    const TENANT_ID: TenantId =
-        TenantId::from_array([1, 1, 2, 3, 4, 5, 6, 7, 8, 9, 1, 2, 3, 4, 5, 6]);
-    const TIMELINE_ID: TimelineId =
-        TimelineId::from_array([1, 1, 2, 3, 4, 5, 6, 7, 8, 9, 1, 2, 3, 4, 5, 7]);
-    const ENDPOINT_ID: &str = "ep-winter-frost-a662z3vg";
-
-    #[test]
-    fn s3_path() {
-        let auth = Claims {
-            tenant_id: TENANT_ID,
-            timeline_id: TIMELINE_ID,
-            endpoint_id: ENDPOINT_ID.into(),
-            exp: u64::MAX,
-        };
-        let s3_path = |key| {
-            let path = &format!("{TENANT_ID}/{TIMELINE_ID}/{ENDPOINT_ID}/{key}");
-            let path = RemotePath::from_string(path).unwrap();
-            S3Path { path }
-        };
-
-        let path = "cache_key".to_string();
-        let mut key_path = KeyRequest {
-            path,
-            tenant_id: auth.tenant_id,
-            timeline_id: auth.timeline_id,
-            endpoint_id: auth.endpoint_id,
-        };
-        assert_eq!(S3Path::try_from(&key_path).unwrap(), s3_path(key_path.path));
-
-        key_path.path = "we/can/have/nested/paths".to_string();
-        assert_eq!(S3Path::try_from(&key_path).unwrap(), s3_path(key_path.path));
-
-        key_path.path = "../error/hello/../".to_string();
-        assert!(S3Path::try_from(&key_path).is_err());
-    }
-
-    #[test]
-    fn prefix_s3_path() {
-        let mut path = PrefixKeyPath {
-            tenant_id: TENANT_ID,
-            timeline_id: None,
-            endpoint_id: None,
-        };
-        let prefix_path = |s: String| RemotePath::from_string(&s).unwrap();
-        assert_eq!(
-            PrefixS3Path::from(&path).path,
-            prefix_path(format!("{TENANT_ID}"))
-        );
-
-        path.timeline_id = Some(TIMELINE_ID);
-        assert_eq!(
-            PrefixS3Path::from(&path).path,
-            prefix_path(format!("{TENANT_ID}/{TIMELINE_ID}"))
-        );
-
-        path.endpoint_id = Some(ENDPOINT_ID.into());
-        assert_eq!(
-            PrefixS3Path::from(&path).path,
-            prefix_path(format!("{TENANT_ID}/{TIMELINE_ID}/{ENDPOINT_ID}"))
-        );
-    }
-}

object_storage/src/main.rs

@@ -1,65 +0,0 @@
-//! `object_storage` is a service which provides API for uploading and downloading
-//! files. It is used by compute and control plane for accessing LFC prewarm data.
-//! This service is deployed either as a separate component or as part of compute image
-//! for large computes.
-mod app;
-use anyhow::Context;
-use tracing::info;
-use utils::logging;
-
-//see set()
-const fn max_upload_file_limit() -> usize {
-    100 * 1024 * 1024
-}
-
-#[derive(serde::Deserialize)]
-#[serde(tag = "type")]
-struct Config {
-    listen: std::net::SocketAddr,
-    pemfile: camino::Utf8PathBuf,
-    #[serde(flatten)]
-    storage_config: remote_storage::RemoteStorageConfig,
-    #[serde(default = "max_upload_file_limit")]
-    max_upload_file_limit: usize,
-}
-
-#[tokio::main]
-async fn main() -> anyhow::Result<()> {
-    logging::init(
-        logging::LogFormat::Plain,
-        logging::TracingErrorLayerEnablement::EnableWithRustLogFilter,
-        logging::Output::Stdout,
-    )?;
-
-    let config: String = std::env::args().skip(1).take(1).collect();
-    if config.is_empty() {
-        anyhow::bail!("Usage: object_storage config.json")
-    }
-    info!("Reading config from {config}");
-    let config = std::fs::read_to_string(config.clone())?;
-    let config: Config = serde_json::from_str(&config).context("parsing config")?;
-    info!("Reading pemfile from {}", config.pemfile.clone());
-    let pemfile = std::fs::read(config.pemfile.clone())?;
-    info!("Loading public key from {}", config.pemfile.clone());
-    let auth = object_storage::JwtAuth::new(&pemfile)?;
-
-    let listener = tokio::net::TcpListener::bind(config.listen).await.unwrap();
-    info!("listening on {}", listener.local_addr().unwrap());
-
-    let storage = remote_storage::GenericRemoteStorage::from_config(&config.storage_config).await?;
-    let cancel = tokio_util::sync::CancellationToken::new();
-    app::check_storage_permissions(&storage, cancel.clone()).await?;
-
-    let proxy = std::sync::Arc::new(object_storage::Storage {
-        auth,
-        storage,
-        cancel: cancel.clone(),
-        max_upload_file_limit: config.max_upload_file_limit,
-    });
-
-    tokio::spawn(utils::signals::signal_handler(cancel.clone()));
-    axum::serve(listener, app::app(proxy))
-        .with_graceful_shutdown(async move { cancel.cancelled().await })
-        .await?;
-    Ok(())
-}

pageserver/Cargo.toml

@@ -10,8 +10,6 @@ default = []
 # which adds some runtime cost to run tests on outage conditions
 testing = ["fail/failpoints", "pageserver_api/testing", "wal_decoder/testing", "pageserver_client/testing"]
 
-fuzz-read-path = ["testing"]
-
 [dependencies]
 anyhow.workspace = true
 arc-swap.workspace = true
@@ -35,7 +33,6 @@ humantime.workspace = true
 humantime-serde.workspace = true
 hyper0.workspace = true
 itertools.workspace = true
-jsonwebtoken.workspace = true
 md5.workspace = true
 nix.workspace = true
 # hack to get the number of worker threads tokio uses
@@ -78,7 +75,6 @@ metrics.workspace = true
 pageserver_api.workspace = true
 pageserver_client.workspace = true # for ResponseErrorMessageExt TOOD refactor that
 pageserver_compaction.workspace = true
-pem.workspace = true
 postgres_connection.workspace = true
 postgres_ffi.workspace = true
 pq_proto.workspace = true

pageserver/benches/bench_ingest.rs

@@ -126,7 +126,7 @@ async fn ingest(
             max_concurrency: NonZeroUsize::new(1).unwrap(),
         });
         let (_desc, path) = layer
-            .write_to_disk(&ctx, None, l0_flush_state.inner(), &gate, cancel.clone())
+            .write_to_disk(&ctx, None, l0_flush_state.inner())
             .await?
             .unwrap();
         tokio::fs::remove_file(path).await?;

pageserver/benches/bench_walredo.rs

@@ -65,7 +65,7 @@ use bytes::{Buf, Bytes};
 use criterion::{BenchmarkId, Criterion};
 use once_cell::sync::Lazy;
 use pageserver::config::PageServerConf;
-use pageserver::walredo::{PostgresRedoManager, RedoAttemptType};
+use pageserver::walredo::PostgresRedoManager;
 use pageserver_api::key::Key;
 use pageserver_api::record::NeonWalRecord;
 use pageserver_api::shard::TenantShardId;
@@ -223,14 +223,7 @@ impl Request {
 
         // TODO: avoid these clones
         manager
-            .request_redo(
-                *key,
-                *lsn,
-                base_img.clone(),
-                records.clone(),
-                *pg_version,
-                RedoAttemptType::ReadPage,
-            )
+            .request_redo(*key, *lsn, base_img.clone(), records.clone(), *pg_version)
             .await
             .context("request_redo")
     }

pageserver/pagebench/src/cmd/getpage_latest_lsn.rs

@@ -68,13 +68,6 @@ pub(crate) struct Args {
     targets: Option<Vec<TenantTimelineId>>,
 }
 
-/// State shared by all clients
-#[derive(Debug)]
-struct SharedState {
-    start_work_barrier: tokio::sync::Barrier,
-    live_stats: LiveStats,
-}
-
 #[derive(Debug, Default)]
 struct LiveStats {
     completed_requests: AtomicU64,
@@ -247,26 +240,24 @@ async fn main_impl(
         all_ranges
     };
 
+    let live_stats = Arc::new(LiveStats::default());
+
     let num_live_stats_dump = 1;
     let num_work_sender_tasks = args.num_clients.get() * timelines.len();
     let num_main_impl = 1;
 
-    let shared_state = Arc::new(SharedState {
-        start_work_barrier: tokio::sync::Barrier::new(
-            num_live_stats_dump + num_work_sender_tasks + num_main_impl,
-        ),
-        live_stats: LiveStats::default(),
-    });
-    let cancel = CancellationToken::new();
+    let start_work_barrier = Arc::new(tokio::sync::Barrier::new(
+        num_live_stats_dump + num_work_sender_tasks + num_main_impl,
+    ));
 
-    let ss = shared_state.clone();
     tokio::spawn({
+        let stats = Arc::clone(&live_stats);
+        let start_work_barrier = Arc::clone(&start_work_barrier);
         async move {
-            ss.start_work_barrier.wait().await;
+            start_work_barrier.wait().await;
             loop {
                 let start = std::time::Instant::now();
                 tokio::time::sleep(std::time::Duration::from_secs(1)).await;
-                let stats = &ss.live_stats;
                 let completed_requests = stats.completed_requests.swap(0, Ordering::Relaxed);
                 let missed = stats.missed.swap(0, Ordering::Relaxed);
                 let elapsed = start.elapsed();
@@ -279,12 +270,14 @@ async fn main_impl(
         }
     });
 
+    let cancel = CancellationToken::new();
+
     let rps_period = args
         .per_client_rate
         .map(|rps_limit| Duration::from_secs_f64(1.0 / (rps_limit as f64)));
     let make_worker: &dyn Fn(WorkerId) -> Pin<Box<dyn Send + Future<Output = ()>>> = &|worker_id| {
-        let ss = shared_state.clone();
-        let cancel = cancel.clone();
+        let live_stats = live_stats.clone();
+        let start_work_barrier = start_work_barrier.clone();
         let ranges: Vec<KeyRange> = all_ranges
             .iter()
             .filter(|r| r.timeline == worker_id.timeline)
@@ -294,8 +287,85 @@ async fn main_impl(
             rand::distributions::weighted::WeightedIndex::new(ranges.iter().map(|v| v.len()))
                 .unwrap();
 
+        let cancel = cancel.clone();
         Box::pin(async move {
-            client_libpq(args, worker_id, ss, cancel, rps_period, ranges, weights).await
+            let client =
+                pageserver_client::page_service::Client::new(args.page_service_connstring.clone())
+                    .await
+                    .unwrap();
+            let mut client = client
+                .pagestream(worker_id.timeline.tenant_id, worker_id.timeline.timeline_id)
+                .await
+                .unwrap();
+
+            start_work_barrier.wait().await;
+            let client_start = Instant::now();
+            let mut ticks_processed = 0;
+            let mut inflight = VecDeque::new();
+            while !cancel.is_cancelled() {
+                // Detect if a request took longer than the RPS rate
+                if let Some(period) = &rps_period {
+                    let periods_passed_until_now =
+                        usize::try_from(client_start.elapsed().as_micros() / period.as_micros())
+                            .unwrap();
+
+                    if periods_passed_until_now > ticks_processed {
+                        live_stats.missed((periods_passed_until_now - ticks_processed) as u64);
+                    }
+                    ticks_processed = periods_passed_until_now;
+                }
+
+                while inflight.len() < args.queue_depth.get() {
+                    let start = Instant::now();
+                    let req = {
+                        let mut rng = rand::thread_rng();
+                        let r = &ranges[weights.sample(&mut rng)];
+                        let key: i128 = rng.gen_range(r.start..r.end);
+                        let key = Key::from_i128(key);
+                        assert!(key.is_rel_block_key());
+                        let (rel_tag, block_no) = key
+                            .to_rel_block()
+                            .expect("we filter non-rel-block keys out above");
+                        PagestreamGetPageRequest {
+                            hdr: PagestreamRequest {
+                                reqid: 0,
+                                request_lsn: if rng.gen_bool(args.req_latest_probability) {
+                                    Lsn::MAX
+                                } else {
+                                    r.timeline_lsn
+                                },
+                                not_modified_since: r.timeline_lsn,
+                            },
+                            rel: rel_tag,
+                            blkno: block_no,
+                        }
+                    };
+                    client.getpage_send(req).await.unwrap();
+                    inflight.push_back(start);
+                }
+
+                let start = inflight.pop_front().unwrap();
+                client.getpage_recv().await.unwrap();
+                let end = Instant::now();
+                live_stats.request_done();
+                ticks_processed += 1;
+                STATS.with(|stats| {
+                    stats
+                        .borrow()
+                        .lock()
+                        .unwrap()
+                        .observe(end.duration_since(start))
+                        .unwrap();
+                });
+
+                if let Some(period) = &rps_period {
+                    let next_at = client_start
+                        + Duration::from_micros(
+                            (ticks_processed) as u64 * u64::try_from(period.as_micros()).unwrap(),
+                        );
+                    tokio::time::sleep_until(next_at.into()).await;
+                }
+            }
         })
     };
 
@@ -317,7 +387,7 @@ async fn main_impl(
     };
 
     info!("waiting for everything to become ready");
-    shared_state.start_work_barrier.wait().await;
+    start_work_barrier.wait().await;
     info!("work started");
     if let Some(runtime) = args.runtime {
         tokio::time::sleep(runtime.into()).await;
@@ -346,91 +416,3 @@ async fn main_impl(
 
     anyhow::Ok(())
 }
-
-async fn client_libpq(
-    args: &Args,
-    worker_id: WorkerId,
-    shared_state: Arc<SharedState>,
-    cancel: CancellationToken,
-    rps_period: Option<Duration>,
-    ranges: Vec<KeyRange>,
-    weights: rand::distributions::weighted::WeightedIndex<i128>,
-) {
-    let client = pageserver_client::page_service::Client::new(args.page_service_connstring.clone())
-        .await
-        .unwrap();
-    let mut client = client
-        .pagestream(worker_id.timeline.tenant_id, worker_id.timeline.timeline_id)
-        .await
-        .unwrap();
-
-    shared_state.start_work_barrier.wait().await;
-    let client_start = Instant::now();
-    let mut ticks_processed = 0;
-    let mut inflight = VecDeque::new();
-    while !cancel.is_cancelled() {
-        // Detect if a request took longer than the RPS rate
-        if let Some(period) = &rps_period {
-            let periods_passed_until_now =
-                usize::try_from(client_start.elapsed().as_micros() / period.as_micros()).unwrap();
-
-            if periods_passed_until_now > ticks_processed {
-                shared_state
-                    .live_stats
-                    .missed((periods_passed_until_now - ticks_processed) as u64);
-            }
-            ticks_processed = periods_passed_until_now;
-        }
-
-        while inflight.len() < args.queue_depth.get() {
-            let start = Instant::now();
-            let req = {
-                let mut rng = rand::thread_rng();
-                let r = &ranges[weights.sample(&mut rng)];
-                let key: i128 = rng.gen_range(r.start..r.end);
-                let key = Key::from_i128(key);
-                assert!(key.is_rel_block_key());
-                let (rel_tag, block_no) = key
-                    .to_rel_block()
-                    .expect("we filter non-rel-block keys out above");
-                PagestreamGetPageRequest {
-                    hdr: PagestreamRequest {
-                        reqid: 0,
-                        request_lsn: if rng.gen_bool(args.req_latest_probability) {
-                            Lsn::MAX
-                        } else {
-                            r.timeline_lsn
-                        },
-                        not_modified_since: r.timeline_lsn,
-                    },
-                    rel: rel_tag,
-                    blkno: block_no,
-                }
-            };
-            client.getpage_send(req).await.unwrap();
-            inflight.push_back(start);
-        }
-
-        let start = inflight.pop_front().unwrap();
-        client.getpage_recv().await.unwrap();
-        let end = Instant::now();
-        shared_state.live_stats.request_done();
-        ticks_processed += 1;
-        STATS.with(|stats| {
-            stats
-                .borrow()
-                .lock()
-                .unwrap()
-                .observe(end.duration_since(start))
-                .unwrap();
-        });
-
-        if let Some(period) = &rps_period {
-            let next_at = client_start
-                + Duration::from_micros(
-                    (ticks_processed) as u64 * u64::try_from(period.as_micros()).unwrap(),
-                );
-            tokio::time::sleep_until(next_at.into()).await;
-        }
-    }
-}

pageserver/src/basebackup.rs

@@ -34,7 +34,7 @@ use utils::lsn::Lsn;
 use crate::context::RequestContext;
 use crate::pgdatadir_mapping::Version;
 use crate::tenant::storage_layer::IoConcurrency;
-use crate::tenant::timeline::{GetVectoredError, VersionedKeySpaceQuery};
+use crate::tenant::timeline::GetVectoredError;
 use crate::tenant::{PageReconstructError, Timeline};
 
 #[derive(Debug, thiserror::Error)]
@@ -353,10 +353,9 @@ where
             let mut slru_builder = SlruSegmentsBuilder::new(&mut self.ar);
 
             for part in slru_partitions.parts {
-                let query = VersionedKeySpaceQuery::uniform(part, self.lsn);
                 let blocks = self
                     .timeline
-                    .get_vectored(query, self.io_concurrency.clone(), self.ctx)
+                    .get_vectored(part, self.lsn, self.io_concurrency.clone(), self.ctx)
                     .await?;
 
                 for (key, block) in blocks {

pageserver/src/bin/pageserver.rs

@@ -31,6 +31,7 @@ use pageserver::{
 };
 use postgres_backend::AuthType;
 use remote_storage::GenericRemoteStorage;
+use tokio::signal::unix::SignalKind;
 use tokio::time::Instant;
 use tokio_util::sync::CancellationToken;
 use tracing::*;
@@ -416,18 +417,8 @@ fn start_pageserver(
     // The storage_broker::connect call needs to happen inside a tokio runtime thread.
     let broker_client = WALRECEIVER_RUNTIME
         .block_on(async {
-            let tls_config = storage_broker::ClientTlsConfig::new().ca_certificates(
-                conf.ssl_ca_certs
-                    .iter()
-                    .map(pem::encode)
-                    .map(storage_broker::Certificate::from_pem),
-            );
             // Note: we do not attempt connecting here (but validate endpoints sanity).
-            storage_broker::connect(
-                conf.broker_endpoint.clone(),
-                conf.broker_keepalive_interval,
-                tls_config,
-            )
+            storage_broker::connect(conf.broker_endpoint.clone(), conf.broker_keepalive_interval)
         })
         .with_context(|| {
             format!(
@@ -462,24 +453,6 @@ fn start_pageserver(
     info!("Using auth for http API: {:#?}", conf.http_auth_type);
     info!("Using auth for pg connections: {:#?}", conf.pg_auth_type);
 
-    let tls_server_config = if conf.listen_https_addr.is_some() || conf.enable_tls_page_service_api
-    {
-        let resolver = BACKGROUND_RUNTIME.block_on(ReloadingCertificateResolver::new(
-            "main",
-            &conf.ssl_key_file,
-            &conf.ssl_cert_file,
-            conf.ssl_cert_reload_period,
-        ))?;
-
-        let server_config = rustls::ServerConfig::builder()
-            .with_no_client_auth()
-            .with_cert_resolver(resolver);
-
-        Some(Arc::new(server_config))
-    } else {
-        None
-    };
-
     match var("NEON_AUTH_TOKEN") {
         Ok(v) => {
             info!("Loaded JWT token for authentication with Safekeeper");
@@ -698,11 +671,17 @@ fn start_pageserver(
 
         let https_task = match https_listener {
             Some(https_listener) => {
-                let tls_server_config = tls_server_config
-                    .clone()
-                    .expect("tls_server_config is set earlier if https is enabled");
+                let resolver = MGMT_REQUEST_RUNTIME.block_on(ReloadingCertificateResolver::new(
+                    &conf.ssl_key_file,
+                    &conf.ssl_cert_file,
+                    conf.ssl_cert_reload_period,
+                ))?;
 
-                let tls_acceptor = tokio_rustls::TlsAcceptor::from(tls_server_config);
+                let server_config = rustls::ServerConfig::builder()
+                    .with_no_client_auth()
+                    .with_cert_resolver(resolver);
+
+                let tls_acceptor = tokio_rustls::TlsAcceptor::from(Arc::new(server_config));
 
                 let server =
                     http_utils::server::Server::new(service, https_listener, Some(tls_acceptor))?;
@@ -758,11 +737,6 @@ fn start_pageserver(
             tokio::net::TcpListener::from_std(pageserver_listener)
                 .context("create tokio listener")?
         },
-        if conf.enable_tls_page_service_api {
-            tls_server_config
-        } else {
-            None
-        },
     );
 
     // All started up! Now just sit and wait for shutdown signal.
@@ -770,7 +744,32 @@ fn start_pageserver(
         let signal_token = CancellationToken::new();
         let signal_cancel = signal_token.child_token();
 
-        tokio::spawn(utils::signals::signal_handler(signal_token));
+        // Spawn signal handlers. Runs in a loop since we want to be responsive to multiple signals
+        // even after triggering shutdown (e.g. a SIGQUIT after a slow SIGTERM shutdown). See:
+        // https://github.com/neondatabase/neon/issues/9740.
+        tokio::spawn(async move {
+            let mut sigint = tokio::signal::unix::signal(SignalKind::interrupt()).unwrap();
+            let mut sigterm = tokio::signal::unix::signal(SignalKind::terminate()).unwrap();
+            let mut sigquit = tokio::signal::unix::signal(SignalKind::quit()).unwrap();
+
+            loop {
+                let signal = tokio::select! {
+                    _ = sigquit.recv() => {
+                        info!("Got signal SIGQUIT. Terminating in immediate shutdown mode.");
+                        std::process::exit(111);
+                    }
+                    _ = sigint.recv() => "SIGINT",
+                    _ = sigterm.recv() => "SIGTERM",
+                };
+
+                if !signal_token.is_cancelled() {
+                    info!("Got signal {signal}. Terminating gracefully in fast shutdown mode.");
+                    signal_token.cancel();
+                } else {
+                    info!("Got signal {signal}. Already shutting down.");
+                }
+            }
+        });
 
         // Wait for cancellation signal and shut down the pageserver.
         //