feat(proxy): allow proxy to connect to separate compute compared to mock cplane

.github/workflows/benchmarking.yml

@@ -308,7 +308,6 @@ jobs:
           "image": [ "'"$image_default"'" ],
           "include": [{ "pg_version": 16, "region_id": "'"$region_id_default"'", "platform": "neonvm-captest-freetier",       "db_size": "3gb" ,"runner": '"$runner_default"', "image": "'"$image_default"'" },
                       { "pg_version": 16, "region_id": "'"$region_id_default"'", "platform": "neonvm-captest-new",            "db_size": "10gb","runner": '"$runner_default"', "image": "'"$image_default"'" },
-                      { "pg_version": 16, "region_id": "'"$region_id_default"'", "platform": "neonvm-captest-new-many-tables","db_size": "10gb","runner": '"$runner_default"', "image": "'"$image_default"'" },
                       { "pg_version": 16, "region_id": "'"$region_id_default"'", "platform": "neonvm-captest-new",            "db_size": "50gb","runner": '"$runner_default"', "image": "'"$image_default"'" },
                       { "pg_version": 16, "region_id": "azure-eastus2",          "platform": "neonvm-azure-captest-freetier", "db_size": "3gb" ,"runner": '"$runner_azure"',   "image": "neondatabase/build-tools:pinned-bookworm" },
                       { "pg_version": 16, "region_id": "azure-eastus2",          "platform": "neonvm-azure-captest-new",      "db_size": "10gb","runner": '"$runner_azure"',   "image": "neondatabase/build-tools:pinned-bookworm" },
@@ -411,7 +410,7 @@ jobs:
         aws-oicd-role-arn: ${{ vars.DEV_AWS_OIDC_ROLE_ARN }}
 
     - name: Create Neon Project
-      if: contains(fromJson('["neonvm-captest-new", "neonvm-captest-new-many-tables", "neonvm-captest-freetier", "neonvm-azure-captest-freetier", "neonvm-azure-captest-new"]'), matrix.platform)
+      if: contains(fromJson('["neonvm-captest-new", "neonvm-captest-freetier", "neonvm-azure-captest-freetier", "neonvm-azure-captest-new"]'), matrix.platform)
       id: create-neon-project
       uses: ./.github/actions/neon-project-create
       with:
@@ -430,7 +429,7 @@ jobs:
           neonvm-captest-sharding-reuse)
             CONNSTR=${{ secrets.BENCHMARK_CAPTEST_SHARDING_CONNSTR }}
             ;;
-          neonvm-captest-new | neonvm-captest-new-many-tables | neonvm-captest-freetier | neonvm-azure-captest-new | neonvm-azure-captest-freetier)
+          neonvm-captest-new | neonvm-captest-freetier | neonvm-azure-captest-new | neonvm-azure-captest-freetier)
             CONNSTR=${{ steps.create-neon-project.outputs.dsn }}
             ;;
           rds-aurora)
@@ -447,26 +446,6 @@ jobs:
 
         echo "connstr=${CONNSTR}" >> $GITHUB_OUTPUT
 
-    # we want to compare Neon project OLTP throughput and latency at scale factor 10 GB 
-    # without (neonvm-captest-new)
-    # and with (neonvm-captest-new-many-tables) many relations in the database
-    - name: Create many relations before the run
-      if: contains(fromJson('["neonvm-captest-new-many-tables"]'), matrix.platform)
-      uses: ./.github/actions/run-python-test-set
-      with:
-        build_type: ${{ env.BUILD_TYPE }}
-        test_selection: performance
-        run_in_parallel: false
-        save_perf_report: ${{ env.SAVE_PERF_REPORT }}
-        extra_params: -m remote_cluster --timeout 21600 -k test_perf_many_relations
-        pg_version: ${{ env.DEFAULT_PG_VERSION }}
-        aws-oicd-role-arn: ${{ vars.DEV_AWS_OIDC_ROLE_ARN }}
-      env:
-        BENCHMARK_CONNSTR: ${{ steps.set-up-connstr.outputs.connstr }}
-        VIP_VAP_ACCESS_TOKEN: "${{ secrets.VIP_VAP_ACCESS_TOKEN }}"
-        PERF_TEST_RESULT_CONNSTR: "${{ secrets.PERF_TEST_RESULT_CONNSTR }}"
-        TEST_NUM_RELATIONS: 10000
-
     - name: Benchmark init
       uses: ./.github/actions/run-python-test-set
       with:

.github/workflows/cloud-regress.yml

@@ -21,8 +21,6 @@ concurrency:
 
 permissions:
   id-token: write # aws-actions/configure-aws-credentials
-  statuses: write
-  contents: write
 
 jobs:
   regress:

Cargo.lock

@@ -10,9 +10,9 @@ checksum = "8b5ace29ee3216de37c0546865ad08edef58b0f9e76838ed8959a84a990e58c5"
 
 [[package]]
 name = "addr2line"
-version = "0.24.2"
+version = "0.21.0"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "dfbe277e56a376000877090da837660b4427aad530e3028d44e0bffe4f89a1c1"
+checksum = "8a30b2e23b9e17a9f90641c7ab1549cd9b44f296d3ccbf309d2863cfe398a0cb"
 dependencies = [
  "gimli",
 ]
@@ -23,12 +23,6 @@ version = "1.0.2"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "f26201604c87b1e01bd3d98f8d5d9a8fcbb815e8cedb41ffccbeb4bf593a35fe"
 
-[[package]]
-name = "adler2"
-version = "2.0.0"
-source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "512761e0bb2578dd7380c6baaa0f4ce03e84f95e960231d1dec8bf4d7d6e2627"
-
 [[package]]
 name = "ahash"
 version = "0.8.11"
@@ -877,17 +871,17 @@ dependencies = [
 
 [[package]]
 name = "backtrace"
-version = "0.3.74"
+version = "0.3.69"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "8d82cb332cdfaed17ae235a638438ac4d4839913cc2af585c3c6746e8f8bee1a"
+checksum = "2089b7e3f35b9dd2d0ed921ead4f6d318c27680d4a5bd167b3ee120edb105837"
 dependencies = [
  "addr2line",
+ "cc",
  "cfg-if",
  "libc",
- "miniz_oxide 0.8.0",
+ "miniz_oxide",
  "object",
  "rustc-demangle",
- "windows-targets 0.52.6",
 ]
 
 [[package]]
@@ -1133,7 +1127,7 @@ dependencies = [
  "num-traits",
  "serde",
  "wasm-bindgen",
- "windows-targets 0.52.6",
+ "windows-targets 0.52.4",
 ]
 
 [[package]]
@@ -2113,7 +2107,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "3b9429470923de8e8cbd4d2dc513535400b4b3fef0319fb5c4e1f520a7bef743"
 dependencies = [
  "crc32fast",
- "miniz_oxide 0.7.1",
+ "miniz_oxide",
 ]
 
 [[package]]
@@ -2314,9 +2308,9 @@ dependencies = [
 
 [[package]]
 name = "gimli"
-version = "0.31.1"
+version = "0.28.1"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "07e28edb80900c19c28f1072f2e8aeca7fa06b23cd4169cefe1af5aa3260783f"
+checksum = "4271d37baee1b8c7e4b708028c57d816cf9d2434acb33a549475f78c181f6253"
 
 [[package]]
 name = "git-version"
@@ -3410,15 +3404,6 @@ dependencies = [
  "adler",
 ]
 
-[[package]]
-name = "miniz_oxide"
-version = "0.8.0"
-source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "e2d80299ef12ff69b16a84bb182e3b9df68b5a91574d3d4fa6e41b65deec4df1"
-dependencies = [
- "adler2",
-]
-
 [[package]]
 name = "mio"
 version = "0.8.11"
@@ -3653,9 +3638,9 @@ dependencies = [
 
 [[package]]
 name = "object"
-version = "0.36.5"
+version = "0.32.2"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "aedf0a2d09c573ed1d8d85b30c119153926a2b36dce0ab28322c09a117a4683e"
+checksum = "a6a622008b6e321afc04970976f62ee297fdbaa6f95318ca343e3eebb9648441"
 dependencies = [
  "memchr",
 ]
@@ -3945,7 +3930,6 @@ dependencies = [
  "serde_json",
  "serde_path_to_error",
  "serde_with",
- "serial_test",
  "smallvec",
  "storage_broker",
  "strum",
@@ -5339,9 +5323,9 @@ dependencies = [
 
 [[package]]
 name = "rustc-demangle"
-version = "0.1.24"
+version = "0.1.23"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "719b953e2095829ee67db738b3bfa9fa368c94900df327b3f07fe6e794d2fe1f"
+checksum = "d626bb9dae77e28219937af045c257c28bfd3f69333c512553507f5f9798cb76"
 
 [[package]]
 name = "rustc-hash"
@@ -5613,15 +5597,6 @@ dependencies = [
  "winapi-util",
 ]
 
-[[package]]
-name = "scc"
-version = "2.2.6"
-source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "94b13f8ea6177672c49d12ed964cca44836f59621981b04a3e26b87e675181de"
-dependencies = [
- "sdd",
-]
-
 [[package]]
 name = "schannel"
 version = "0.1.23"
@@ -5662,12 +5637,6 @@ version = "0.4.1"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "621e3680f3e07db4c9c2c3fb07c6223ab2fab2e54bd3c04c3ae037990f428c32"
 
-[[package]]
-name = "sdd"
-version = "3.0.5"
-source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "478f121bb72bbf63c52c93011ea1791dca40140dfe13f8336c4c5ac952c33aa9"
-
 [[package]]
 name = "sec1"
 version = "0.3.0"
@@ -5957,31 +5926,6 @@ dependencies = [
  "serde",
 ]
 
-[[package]]
-name = "serial_test"
-version = "3.2.0"
-source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "1b258109f244e1d6891bf1053a55d63a5cd4f8f4c30cf9a1280989f80e7a1fa9"
-dependencies = [
- "futures",
- "log",
- "once_cell",
- "parking_lot 0.12.1",
- "scc",
- "serial_test_derive",
-]
-
-[[package]]
-name = "serial_test_derive"
-version = "3.2.0"
-source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "5d69265a08751de7844521fd15003ae0a888e035773ba05695c5c759a6f89eef"
-dependencies = [
- "proc-macro2",
- "quote",
- "syn 2.0.90",
-]
-
 [[package]]
 name = "sha1"
 version = "0.10.5"
@@ -7275,7 +7219,6 @@ dependencies = [
  "anyhow",
  "arc-swap",
  "async-compression",
- "backtrace",
  "bincode",
  "byteorder",
  "bytes",
@@ -7286,14 +7229,12 @@ dependencies = [
  "criterion",
  "diatomic-waker",
  "fail",
- "flate2",
  "futures",
  "git-version",
  "hex",
  "hex-literal",
  "humantime",
  "hyper 0.14.30",
- "itertools 0.10.5",
  "jemalloc_pprof",
  "jsonwebtoken",
  "metrics",
@@ -7650,7 +7591,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "e48a53791691ab099e5e2ad123536d0fff50652600abaf43bbf952894110d0be"
 dependencies = [
  "windows-core",
- "windows-targets 0.52.6",
+ "windows-targets 0.52.4",
 ]
 
 [[package]]
@@ -7659,7 +7600,7 @@ version = "0.52.0"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "33ab640c8d7e35bf8ba19b884ba838ceb4fba93a4e8c65a9059d08afcfc683d9"
 dependencies = [
- "windows-targets 0.52.6",
+ "windows-targets 0.52.4",
 ]
 
 [[package]]
@@ -7677,7 +7618,7 @@ version = "0.52.0"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "282be5f36a8ce781fad8c8ae18fa3f9beff57ec1b52cb3de0789201425d9a33d"
 dependencies = [
- "windows-targets 0.52.6",
+ "windows-targets 0.52.4",
 ]
 
 [[package]]
@@ -7697,18 +7638,17 @@ dependencies = [
 
 [[package]]
 name = "windows-targets"
-version = "0.52.6"
+version = "0.52.4"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "9b724f72796e036ab90c1021d4780d4d3d648aca59e491e6b98e725b84e99973"
+checksum = "7dd37b7e5ab9018759f893a1952c9420d060016fc19a472b4bb20d1bdd694d1b"
 dependencies = [
- "windows_aarch64_gnullvm 0.52.6",
- "windows_aarch64_msvc 0.52.6",
- "windows_i686_gnu 0.52.6",
- "windows_i686_gnullvm",
- "windows_i686_msvc 0.52.6",
- "windows_x86_64_gnu 0.52.6",
- "windows_x86_64_gnullvm 0.52.6",
- "windows_x86_64_msvc 0.52.6",
+ "windows_aarch64_gnullvm 0.52.4",
+ "windows_aarch64_msvc 0.52.4",
+ "windows_i686_gnu 0.52.4",
+ "windows_i686_msvc 0.52.4",
+ "windows_x86_64_gnu 0.52.4",
+ "windows_x86_64_gnullvm 0.52.4",
+ "windows_x86_64_msvc 0.52.4",
 ]
 
 [[package]]
@@ -7719,9 +7659,9 @@ checksum = "91ae572e1b79dba883e0d315474df7305d12f569b400fcf90581b06062f7e1bc"
 
 [[package]]
 name = "windows_aarch64_gnullvm"
-version = "0.52.6"
+version = "0.52.4"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "32a4622180e7a0ec044bb555404c800bc9fd9ec262ec147edd5989ccd0c02cd3"
+checksum = "bcf46cf4c365c6f2d1cc93ce535f2c8b244591df96ceee75d8e83deb70a9cac9"
 
 [[package]]
 name = "windows_aarch64_msvc"
@@ -7731,9 +7671,9 @@ checksum = "b2ef27e0d7bdfcfc7b868b317c1d32c641a6fe4629c171b8928c7b08d98d7cf3"
 
 [[package]]
 name = "windows_aarch64_msvc"
-version = "0.52.6"
+version = "0.52.4"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "09ec2a7bb152e2252b53fa7803150007879548bc709c039df7627cabbd05d469"
+checksum = "da9f259dd3bcf6990b55bffd094c4f7235817ba4ceebde8e6d11cd0c5633b675"
 
 [[package]]
 name = "windows_i686_gnu"
@@ -7743,15 +7683,9 @@ checksum = "622a1962a7db830d6fd0a69683c80a18fda201879f0f447f065a3b7467daa241"
 
 [[package]]
 name = "windows_i686_gnu"
-version = "0.52.6"
+version = "0.52.4"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "8e9b5ad5ab802e97eb8e295ac6720e509ee4c243f69d781394014ebfe8bbfa0b"
-
-[[package]]
-name = "windows_i686_gnullvm"
-version = "0.52.6"
-source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "0eee52d38c090b3caa76c563b86c3a4bd71ef1a819287c19d586d7334ae8ed66"
+checksum = "b474d8268f99e0995f25b9f095bc7434632601028cf86590aea5c8a5cb7801d3"
 
 [[package]]
 name = "windows_i686_msvc"
@@ -7761,9 +7695,9 @@ checksum = "4542c6e364ce21bf45d69fdd2a8e455fa38d316158cfd43b3ac1c5b1b19f8e00"
 
 [[package]]
 name = "windows_i686_msvc"
-version = "0.52.6"
+version = "0.52.4"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "240948bc05c5e7c6dabba28bf89d89ffce3e303022809e73deaefe4f6ec56c66"
+checksum = "1515e9a29e5bed743cb4415a9ecf5dfca648ce85ee42e15873c3cd8610ff8e02"
 
 [[package]]
 name = "windows_x86_64_gnu"
@@ -7773,9 +7707,9 @@ checksum = "ca2b8a661f7628cbd23440e50b05d705db3686f894fc9580820623656af974b1"
 
 [[package]]
 name = "windows_x86_64_gnu"
-version = "0.52.6"
+version = "0.52.4"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "147a5c80aabfbf0c7d901cb5895d1de30ef2907eb21fbbab29ca94c5b08b1a78"
+checksum = "5eee091590e89cc02ad514ffe3ead9eb6b660aedca2183455434b93546371a03"
 
 [[package]]
 name = "windows_x86_64_gnullvm"
@@ -7785,9 +7719,9 @@ checksum = "7896dbc1f41e08872e9d5e8f8baa8fdd2677f29468c4e156210174edc7f7b953"
 
 [[package]]
 name = "windows_x86_64_gnullvm"
-version = "0.52.6"
+version = "0.52.4"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "24d5b23dc417412679681396f2b49f3de8c1473deb516bd34410872eff51ed0d"
+checksum = "77ca79f2451b49fa9e2af39f0747fe999fcda4f5e241b2898624dca97a1f2177"
 
 [[package]]
 name = "windows_x86_64_msvc"
@@ -7797,9 +7731,9 @@ checksum = "1a515f5799fe4961cb532f983ce2b23082366b898e52ffbce459c86f67c8378a"
 
 [[package]]
 name = "windows_x86_64_msvc"
-version = "0.52.6"
+version = "0.52.4"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "589f6da84c646204747d1270a2a5661ea66ed1cced2631d546fdfb155959f9ec"
+checksum = "32b752e52a2da0ddfbdbcc6fceadfeede4c939ed16d13e648833a61dfb611ed8"
 
 [[package]]
 name = "winnow"

Cargo.toml

@@ -52,7 +52,6 @@ anyhow = { version = "1.0", features = ["backtrace"] }
 arc-swap = "1.6"
 async-compression = { version = "0.4.0", features = ["tokio", "gzip", "zstd"] }
 atomic-take = "1.1.0"
-backtrace = "0.3.74"
 flate2 = "1.0.26"
 async-stream = "0.3"
 async-trait = "0.1"
@@ -161,7 +160,6 @@ serde_json = "1"
 serde_path_to_error = "0.1"
 serde_with = { version = "2.0", features = [ "base64" ] }
 serde_assert = "0.5.0"
-serial_test = "3.2.0"
 sha2 = "0.10.2"
 signal-hook = "0.3"
 smallvec = "1.11"

compute/compute-node.Dockerfile

@@ -35,12 +35,10 @@ RUN case $DEBIAN_VERSION in \
       ;; \
     esac && \
     apt update &&  \
-    apt install --no-install-recommends --no-install-suggests -y \
-    ninja-build git autoconf automake libtool build-essential bison flex libreadline-dev \
+    apt install --no-install-recommends -y git autoconf automake libtool build-essential bison flex libreadline-dev \
     zlib1g-dev libxml2-dev libcurl4-openssl-dev libossp-uuid-dev wget ca-certificates pkg-config libssl-dev \
     libicu-dev libxslt1-dev liblz4-dev libzstd-dev zstd \
-    $VERSION_INSTALLS \
-    && apt clean && rm -rf /var/lib/apt/lists/*
+    $VERSION_INSTALLS
 
 #########################################################################################
 #
@@ -115,12 +113,10 @@ ARG DEBIAN_VERSION
 ARG PG_VERSION
 COPY --from=pg-build /usr/local/pgsql/ /usr/local/pgsql/
 RUN apt update && \
-    apt install --no-install-recommends --no-install-suggests -y \
-    gdal-bin libboost-dev libboost-thread-dev libboost-filesystem-dev \
+    apt install --no-install-recommends -y gdal-bin libboost-dev libboost-thread-dev libboost-filesystem-dev \
     libboost-system-dev libboost-iostreams-dev libboost-program-options-dev libboost-timer-dev \
     libcgal-dev libgdal-dev libgmp-dev libmpfr-dev libopenscenegraph-dev libprotobuf-c-dev \
-    protobuf-c-compiler xsltproc \
-    && apt clean && rm -rf /var/lib/apt/lists/*
+    protobuf-c-compiler xsltproc
 
 
 # Postgis 3.5.0 requires SFCGAL 1.4+
@@ -147,9 +143,9 @@ RUN case "${DEBIAN_VERSION}" in \
     wget https://gitlab.com/sfcgal/SFCGAL/-/archive/v${SFCGAL_VERSION}/SFCGAL-v${SFCGAL_VERSION}.tar.gz -O SFCGAL.tar.gz && \
     echo "${SFCGAL_CHECKSUM} SFCGAL.tar.gz" | sha256sum --check && \
     mkdir sfcgal-src && cd sfcgal-src && tar xzf ../SFCGAL.tar.gz --strip-components=1 -C . && \
-    cmake -DCMAKE_BUILD_TYPE=Release -GNinja . && ninja -j $(getconf _NPROCESSORS_ONLN) && \
-    DESTDIR=/sfcgal ninja install -j $(getconf _NPROCESSORS_ONLN) && \
-    ninja clean && cp -R /sfcgal/* /
+    cmake -DCMAKE_BUILD_TYPE=Release . && make -j $(getconf _NPROCESSORS_ONLN) && \
+    DESTDIR=/sfcgal make install -j $(getconf _NPROCESSORS_ONLN) && \
+    make clean && cp -R /sfcgal/* /
 
 ENV PATH="/usr/local/pgsql/bin:$PATH"
 
@@ -217,9 +213,9 @@ RUN case "${PG_VERSION}" in \
     echo "${PGROUTING_CHECKSUM} pgrouting.tar.gz" | sha256sum --check && \
     mkdir pgrouting-src && cd pgrouting-src && tar xzf ../pgrouting.tar.gz --strip-components=1 -C . && \
     mkdir build && cd build && \
-    cmake -GNinja -DCMAKE_BUILD_TYPE=Release .. && \
-    ninja -j $(getconf _NPROCESSORS_ONLN) && \
-    ninja -j $(getconf _NPROCESSORS_ONLN) install && \
+    cmake -DCMAKE_BUILD_TYPE=Release .. && \
+    make -j $(getconf _NPROCESSORS_ONLN) && \
+    make -j $(getconf _NPROCESSORS_ONLN) install && \
     echo 'trusted = true' >> /usr/local/pgsql/share/extension/pgrouting.control && \
     find /usr/local/pgsql -type f | sed 's|^/usr/local/pgsql/||' > /after.txt &&\
     cp /usr/local/pgsql/share/extension/pgrouting.control /extensions/postgis && \
@@ -239,9 +235,7 @@ COPY --from=pg-build /usr/local/pgsql/ /usr/local/pgsql/
 COPY compute/patches/plv8-3.1.10.patch /plv8-3.1.10.patch
 
 RUN apt update && \
-    apt install --no-install-recommends --no-install-suggests -y \
-    ninja-build python3-dev libncurses5 binutils clang \
-    && apt clean && rm -rf /var/lib/apt/lists/*
+    apt install --no-install-recommends -y ninja-build python3-dev libncurses5 binutils clang
 
 # plv8 3.2.3 supports v17
 # last release v3.2.3 - Sep 7, 2024
@@ -307,10 +301,9 @@ RUN mkdir -p /h3/usr/ && \
     echo "ec99f1f5974846bde64f4513cf8d2ea1b8d172d2218ab41803bf6a63532272bc h3.tar.gz" | sha256sum --check && \
     mkdir h3-src && cd h3-src && tar xzf ../h3.tar.gz --strip-components=1 -C . && \
     mkdir build && cd build && \
-    cmake .. -GNinja -DBUILD_BENCHMARKS=0 -DCMAKE_BUILD_TYPE=Release \
-        -DBUILD_FUZZERS=0 -DBUILD_FILTERS=0 -DBUILD_GENERATORS=0 -DBUILD_TESTING=0 \
-    && ninja -j $(getconf _NPROCESSORS_ONLN) && \
-    DESTDIR=/h3 ninja install && \
+    cmake .. -DCMAKE_BUILD_TYPE=Release && \
+    make -j $(getconf _NPROCESSORS_ONLN) && \
+    DESTDIR=/h3 make install && \
     cp -R /h3/usr / && \
     rm -rf build
 
@@ -657,15 +650,14 @@ FROM build-deps AS rdkit-pg-build
 ARG PG_VERSION
 COPY --from=pg-build /usr/local/pgsql/ /usr/local/pgsql/
 
-RUN apt update && \
-    apt install --no-install-recommends --no-install-suggests -y \
+RUN apt-get update && \
+    apt-get install --no-install-recommends -y \
         libboost-iostreams1.74-dev \
         libboost-regex1.74-dev \
         libboost-serialization1.74-dev \
         libboost-system1.74-dev \
         libeigen3-dev \
-        libboost-all-dev \
-    && apt clean && rm -rf /var/lib/apt/lists/*
+        libboost-all-dev
 
 # rdkit Release_2024_09_1 supports v17
 # last release Release_2024_09_1 - Sep 27, 2024
@@ -701,8 +693,6 @@ RUN case "${PG_VERSION}" in \
         -D RDK_BUILD_MOLINTERCHANGE_SUPPORT=OFF \
         -D RDK_BUILD_YAEHMOP_SUPPORT=OFF \
         -D RDK_BUILD_STRUCTCHECKER_SUPPORT=OFF \
-        -D RDK_TEST_MULTITHREADED=OFF \
-        -D RDK_BUILD_CPP_TESTS=OFF \
         -D RDK_USE_URF=OFF \
         -D RDK_BUILD_PGSQL=ON \
         -D RDK_PGSQL_STATIC=ON \
@@ -714,10 +704,9 @@ RUN case "${PG_VERSION}" in \
         -D RDK_INSTALL_COMIC_FONTS=OFF \
         -D RDK_BUILD_FREETYPE_SUPPORT=OFF \
         -D CMAKE_BUILD_TYPE=Release \
-        -GNinja \
         . && \
-    ninja -j $(getconf _NPROCESSORS_ONLN) && \
-    ninja -j $(getconf _NPROCESSORS_ONLN) install && \
+    make -j $(getconf _NPROCESSORS_ONLN) && \
+    make -j $(getconf _NPROCESSORS_ONLN) install && \
     echo 'trusted = true' >> /usr/local/pgsql/share/extension/rdkit.control
 
 #########################################################################################
@@ -860,9 +849,8 @@ FROM build-deps AS rust-extensions-build
 ARG PG_VERSION
 COPY --from=pg-build /usr/local/pgsql/ /usr/local/pgsql/
 
-RUN apt update && \
-    apt install --no-install-recommends --no-install-suggests -y curl libclang-dev && \
-    apt clean && rm -rf /var/lib/apt/lists/* && \
+RUN apt-get update && \
+    apt-get install --no-install-recommends -y curl libclang-dev && \
     useradd -ms /bin/bash nonroot -b /home
 
 ENV HOME=/home/nonroot
@@ -897,9 +885,8 @@ FROM build-deps AS rust-extensions-build-pgrx12
 ARG PG_VERSION
 COPY --from=pg-build /usr/local/pgsql/ /usr/local/pgsql/
 
-RUN apt update && \
-    apt install --no-install-recommends --no-install-suggests -y curl libclang-dev && \
-    apt clean && rm -rf /var/lib/apt/lists/* && \
+RUN apt-get update && \
+    apt-get install --no-install-recommends -y curl libclang-dev && \
     useradd -ms /bin/bash nonroot -b /home
 
 ENV HOME=/home/nonroot
@@ -927,22 +914,18 @@ FROM rust-extensions-build-pgrx12 AS pg-onnx-build
 
 # cmake 3.26 or higher is required, so installing it using pip (bullseye-backports has cmake 3.25).
 # Install it using virtual environment, because Python 3.11 (the default version on Debian 12 (Bookworm)) complains otherwise
-RUN apt update && apt install --no-install-recommends --no-install-suggests -y \
-    python3 python3-pip python3-venv && \
-    apt clean && rm -rf /var/lib/apt/lists/* && \
+RUN apt-get update && apt-get install -y python3 python3-pip python3-venv && \
     python3 -m venv venv && \
     . venv/bin/activate && \
     python3 -m pip install cmake==3.30.5 && \
     wget https://github.com/microsoft/onnxruntime/archive/refs/tags/v1.18.1.tar.gz -O onnxruntime.tar.gz && \
     mkdir onnxruntime-src && cd onnxruntime-src && tar xzf ../onnxruntime.tar.gz --strip-components=1 -C . && \
-    ./build.sh --config Release --parallel --cmake_generator Ninja \
-    --skip_submodule_sync --skip_tests --allow_running_as_root
+    ./build.sh --config Release --parallel --skip_submodule_sync --skip_tests --allow_running_as_root
 
 
 FROM pg-onnx-build AS pgrag-pg-build
 
-RUN apt update && apt install --no-install-recommends --no-install-suggests -y protobuf-compiler \
-    && apt clean && rm -rf /var/lib/apt/lists/* && \
+RUN apt-get install -y protobuf-compiler && \
     wget https://github.com/neondatabase-labs/pgrag/archive/refs/tags/v0.0.0.tar.gz -O pgrag.tar.gz &&  \
     echo "2cbe394c1e74fc8bcad9b52d5fbbfb783aef834ca3ce44626cfd770573700bb4 pgrag.tar.gz" | sha256sum --check && \
     mkdir pgrag-src && cd pgrag-src && tar xzf ../pgrag.tar.gz --strip-components=1 -C . && \
@@ -1185,25 +1168,6 @@ RUN case "${PG_VERSION}" in \
     make BUILD_TYPE=release -j $(getconf _NPROCESSORS_ONLN) install && \
     echo 'trusted = true' >> /usr/local/pgsql/share/extension/pg_mooncake.control
 
-#########################################################################################
-#
-# Layer "pg_repack"
-# compile pg_repack extension
-#
-#########################################################################################
-
-FROM build-deps AS pg-repack-build
-ARG PG_VERSION
-COPY --from=pg-build /usr/local/pgsql/ /usr/local/pgsql/
-
-ENV PATH="/usr/local/pgsql/bin/:$PATH"
-
-RUN wget https://github.com/reorg/pg_repack/archive/refs/tags/ver_1.5.2.tar.gz -O pg_repack.tar.gz && \
-    echo '4516cad42251ed3ad53ff619733004db47d5755acac83f75924cd94d1c4fb681 pg_repack.tar.gz' | sha256sum --check && \
-    mkdir pg_repack-src && cd pg_repack-src && tar xzf ../pg_repack.tar.gz --strip-components=1 -C . && \
-    make -j $(getconf _NPROCESSORS_ONLN) && \
-    make -j $(getconf _NPROCESSORS_ONLN) install
-
 #########################################################################################
 #
 # Layer "neon-pg-ext-build"
@@ -1249,7 +1213,6 @@ COPY --from=pg-anon-pg-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-mooncake-build /usr/local/pgsql/ /usr/local/pgsql/
-COPY --from=pg-repack-build /usr/local/pgsql/ /usr/local/pgsql/
 COPY pgxn/ pgxn/
 
 RUN make -j $(getconf _NPROCESSORS_ONLN) \
@@ -1316,8 +1279,8 @@ COPY --from=compute-tools /home/nonroot/target/release-line-debug-size-lto/fast_
 
 FROM debian:$DEBIAN_FLAVOR AS pgbouncer
 RUN set -e \
-    && apt update \
-    && apt install --no-install-suggests --no-install-recommends -y \
+    && apt-get update \
+    && apt-get install --no-install-recommends -y \
         build-essential \
         git \
         ca-certificates \
@@ -1325,8 +1288,7 @@ RUN set -e \
         automake \
         libevent-dev \
         libtool \
-        pkg-config \
-    && apt clean && rm -rf /var/lib/apt/lists/*
+        pkg-config
 
 # Use `dist_man_MANS=` to skip manpage generation (which requires python3/pandoc)
 ENV PGBOUNCER_TAG=pgbouncer_1_22_1
@@ -1556,30 +1518,28 @@ RUN apt update && \
         locales \
         procps \
         ca-certificates \
-        curl \
-        unzip \
         $VERSION_INSTALLS && \
-    apt clean && rm -rf /var/lib/apt/lists/* /tmp/* /var/tmp/* && \
+    rm -rf /var/lib/apt/lists/* /tmp/* /var/tmp/* && \
     localedef -i en_US -c -f UTF-8 -A /usr/share/locale/locale.alias en_US.UTF-8
 
-# aws cli is used by fast_import (curl and unzip above are at this time only used for this installation step)
+# s5cmd 2.2.2 from https://github.com/peak/s5cmd/releases/tag/v2.2.2
+# used by fast_import
 ARG TARGETARCH
+ADD https://github.com/peak/s5cmd/releases/download/v2.2.2/s5cmd_2.2.2_linux_$TARGETARCH.deb /tmp/s5cmd.deb
 RUN set -ex; \
+    \
+    # Determine the expected checksum based on TARGETARCH
     if [ "${TARGETARCH}" = "amd64" ]; then \
-        TARGETARCH_ALT="x86_64"; \
-        CHECKSUM="c9a9df3770a3ff9259cb469b6179e02829687a464e0824d5c32d378820b53a00"; \
+        CHECKSUM="392c385320cd5ffa435759a95af77c215553d967e4b1c0fffe52e4f14c29cf85"; \
     elif [ "${TARGETARCH}" = "arm64" ]; then \
-        TARGETARCH_ALT="aarch64"; \
-        CHECKSUM="8181730be7891582b38b028112e81b4899ca817e8c616aad807c9e9d1289223a"; \
+        CHECKSUM="939bee3cf4b5604ddb00e67f8c157b91d7c7a5b553d1fbb6890fad32894b7b46"; \
     else \
         echo "Unsupported architecture: ${TARGETARCH}"; exit 1; \
     fi; \
-    curl -L "https://awscli.amazonaws.com/awscli-exe-linux-${TARGETARCH_ALT}-2.17.5.zip" -o /tmp/awscliv2.zip; \
-    echo "${CHECKSUM}  /tmp/awscliv2.zip" | sha256sum -c -; \
-    unzip /tmp/awscliv2.zip -d /tmp/awscliv2; \
-    /tmp/awscliv2/aws/install; \
-    rm -rf /tmp/awscliv2.zip /tmp/awscliv2; \
-    true
+    \
+    # Compute and validate the checksum
+    echo "${CHECKSUM}  /tmp/s5cmd.deb" | sha256sum -c -
+RUN dpkg -i /tmp/s5cmd.deb && rm /tmp/s5cmd.deb
 
 ENV LANG=en_US.utf8
 USER postgres

compute/etc/neon_collector.jsonnet

@@ -3,7 +3,7 @@
   metrics: [
     import 'sql_exporter/checkpoints_req.libsonnet',
     import 'sql_exporter/checkpoints_timed.libsonnet',
-    import 'sql_exporter/compute_backpressure_throttling_seconds_total.libsonnet',
+    import 'sql_exporter/compute_backpressure_throttling_seconds.libsonnet',
     import 'sql_exporter/compute_current_lsn.libsonnet',
     import 'sql_exporter/compute_logical_snapshot_files.libsonnet',
     import 'sql_exporter/compute_logical_snapshots_bytes.libsonnet',

compute/etc/sql_exporter/compute_backpressure_throttling_seconds.libsonnet

@@ -1,10 +1,10 @@
 {
-  metric_name: 'compute_backpressure_throttling_seconds_total',
-  type: 'counter',
+  metric_name: 'compute_backpressure_throttling_seconds',
+  type: 'gauge',
   help: 'Time compute has spent throttled',
   key_labels: null,
   values: [
     'throttled',
   ],
-  query: importstr 'sql_exporter/compute_backpressure_throttling_seconds_total.sql',
+  query: importstr 'sql_exporter/compute_backpressure_throttling_seconds.sql',
 }

compute_tools/src/bin/fast_import.rs

@@ -34,12 +34,12 @@ use nix::unistd::Pid;
 use tracing::{info, info_span, warn, Instrument};
 use utils::fs_ext::is_directory_empty;
 
-#[path = "fast_import/aws_s3_sync.rs"]
-mod aws_s3_sync;
 #[path = "fast_import/child_stdio_to_log.rs"]
 mod child_stdio_to_log;
 #[path = "fast_import/s3_uri.rs"]
 mod s3_uri;
+#[path = "fast_import/s5cmd.rs"]
+mod s5cmd;
 
 #[derive(clap::Parser)]
 struct Args {
@@ -326,7 +326,7 @@ pub(crate) async fn main() -> anyhow::Result<()> {
     }
 
     info!("upload pgdata");
-    aws_s3_sync::sync(Utf8Path::new(&pgdata_dir), &s3_prefix.append("/pgdata/"))
+    s5cmd::sync(Utf8Path::new(&pgdata_dir), &s3_prefix.append("/"))
         .await
         .context("sync dump directory to destination")?;
 
@@ -334,10 +334,10 @@ pub(crate) async fn main() -> anyhow::Result<()> {
     {
         let status_dir = working_directory.join("status");
         std::fs::create_dir(&status_dir).context("create status directory")?;
-        let status_file = status_dir.join("pgdata");
+        let status_file = status_dir.join("status");
         std::fs::write(&status_file, serde_json::json!({"done": true}).to_string())
             .context("write status file")?;
-        aws_s3_sync::sync(&status_dir, &s3_prefix.append("/status/"))
+        s5cmd::sync(&status_file, &s3_prefix.append("/status/pgdata"))
             .await
             .context("sync status directory to destination")?;
     }

compute_tools/src/bin/fast_import/s5cmd.rs

@@ -4,21 +4,24 @@ use camino::Utf8Path;
 use super::s3_uri::S3Uri;
 
 pub(crate) async fn sync(local: &Utf8Path, remote: &S3Uri) -> anyhow::Result<()> {
-    let mut builder = tokio::process::Command::new("aws");
+    let mut builder = tokio::process::Command::new("s5cmd");
+    // s5cmd uses aws-sdk-go v1, hence doesn't support AWS_ENDPOINT_URL
+    if let Some(val) = std::env::var_os("AWS_ENDPOINT_URL") {
+        builder.arg("--endpoint-url").arg(val);
+    }
     builder
-        .arg("s3")
         .arg("sync")
         .arg(local.as_str())
         .arg(remote.to_string());
     let st = builder
         .spawn()
-        .context("spawn aws s3 sync")?
+        .context("spawn s5cmd")?
         .wait()
         .await
-        .context("wait for aws s3 sync")?;
+        .context("wait for s5cmd")?;
     if st.success() {
         Ok(())
     } else {
-        Err(anyhow::anyhow!("aws s3 sync failed"))
+        Err(anyhow::anyhow!("s5cmd failed"))
     }
 }

control_plane/src/bin/neon_local.rs

@@ -19,7 +19,6 @@ use control_plane::storage_controller::{
     NeonStorageControllerStartArgs, NeonStorageControllerStopArgs, StorageController,
 };
 use control_plane::{broker, local_env};
-use nix::fcntl::{flock, FlockArg};
 use pageserver_api::config::{
     DEFAULT_HTTP_LISTEN_PORT as DEFAULT_PAGESERVER_HTTP_PORT,
     DEFAULT_PG_LISTEN_PORT as DEFAULT_PAGESERVER_PG_PORT,
@@ -37,8 +36,6 @@ use safekeeper_api::{
 };
 use std::borrow::Cow;
 use std::collections::{BTreeSet, HashMap};
-use std::fs::File;
-use std::os::fd::AsRawFd;
 use std::path::PathBuf;
 use std::process::exit;
 use std::str::FromStr;
@@ -692,21 +689,6 @@ struct TimelineTreeEl {
     pub children: BTreeSet<TimelineId>,
 }
 
-/// A flock-based guard over the neon_local repository directory
-struct RepoLock {
-    _file: File,
-}
-
-impl RepoLock {
-    fn new() -> Result<Self> {
-        let repo_dir = File::open(local_env::base_path())?;
-        let repo_dir_fd = repo_dir.as_raw_fd();
-        flock(repo_dir_fd, FlockArg::LockExclusive)?;
-
-        Ok(Self { _file: repo_dir })
-    }
-}
-
 // Main entry point for the 'neon_local' CLI utility
 //
 // This utility helps to manage neon installation. That includes following:
@@ -718,14 +700,9 @@ fn main() -> Result<()> {
     let cli = Cli::parse();
 
     // Check for 'neon init' command first.
-    let (subcommand_result, _lock) = if let NeonLocalCmd::Init(args) = cli.command {
-        (handle_init(&args).map(|env| Some(Cow::Owned(env))), None)
+    let subcommand_result = if let NeonLocalCmd::Init(args) = cli.command {
+        handle_init(&args).map(|env| Some(Cow::Owned(env)))
     } else {
-        // This tool uses a collection of simple files to store its state, and consequently
-        // it is not generally safe to run multiple commands concurrently.  Rather than expect
-        // all callers to know this, use a lock file to protect against concurrent execution.
-        let _repo_lock = RepoLock::new().unwrap();
-
         // all other commands need an existing config
         let env = LocalEnv::load_config(&local_env::base_path()).context("Error loading config")?;
         let original_env = env.clone();
@@ -751,12 +728,11 @@ fn main() -> Result<()> {
             NeonLocalCmd::Mappings(subcmd) => handle_mappings(&subcmd, env),
         };
 
-        let subcommand_result = if &original_env != env {
+        if &original_env != env {
             subcommand_result.map(|()| Some(Cow::Borrowed(env)))
         } else {
             subcommand_result.map(|()| None)
-        };
-        (subcommand_result, Some(_repo_lock))
+        }
     };
 
     match subcommand_result {
@@ -946,7 +922,7 @@ fn handle_init(args: &InitCmdArgs) -> anyhow::Result<LocalEnv> {
     } else {
         // User (likely interactive) did not provide a description of the environment, give them the default
         NeonLocalInitConf {
-            control_plane_api: Some(DEFAULT_PAGESERVER_CONTROL_PLANE_API.parse().unwrap()),
+            control_plane_api: Some(Some(DEFAULT_PAGESERVER_CONTROL_PLANE_API.parse().unwrap())),
             broker: NeonBroker {
                 listen_addr: DEFAULT_BROKER_ADDR.parse().unwrap(),
             },
@@ -1742,15 +1718,18 @@ async fn handle_start_all_impl(
             broker::start_broker_process(env, &retry_timeout).await
         });
 
-        js.spawn(async move {
-            let storage_controller = StorageController::from_env(env);
-            storage_controller
-                .start(NeonStorageControllerStartArgs::with_default_instance_id(
-                    retry_timeout,
-                ))
-                .await
-                .map_err(|e| e.context("start storage_controller"))
-        });
+        // Only start the storage controller if the pageserver is configured to need it
+        if env.control_plane_api.is_some() {
+            js.spawn(async move {
+                let storage_controller = StorageController::from_env(env);
+                storage_controller
+                    .start(NeonStorageControllerStartArgs::with_default_instance_id(
+                        retry_timeout,
+                    ))
+                    .await
+                    .map_err(|e| e.context("start storage_controller"))
+            });
+        }
 
         for ps_conf in &env.pageservers {
             js.spawn(async move {
@@ -1795,6 +1774,10 @@ async fn neon_start_status_check(
     const RETRY_INTERVAL: Duration = Duration::from_millis(100);
     const NOTICE_AFTER_RETRIES: Duration = Duration::from_secs(5);
 
+    if env.control_plane_api.is_none() {
+        return Ok(());
+    }
+
     let storcon = StorageController::from_env(env);
 
     let retries = retry_timeout.as_millis() / RETRY_INTERVAL.as_millis();

control_plane/src/endpoint.rs

@@ -316,10 +316,6 @@ impl Endpoint {
         // and can cause errors like 'no unpinned buffers available', see
         // <https://github.com/neondatabase/neon/issues/9956>
         conf.append("shared_buffers", "1MB");
-        // Postgres defaults to effective_io_concurrency=1, which does not exercise the pageserver's
-        // batching logic.  Set this to 2 so that we exercise the code a bit without letting
-        // individual tests do a lot of concurrent work on underpowered test machines
-        conf.append("effective_io_concurrency", "2");
         conf.append("fsync", "off");
         conf.append("max_connections", "100");
         conf.append("wal_level", "logical");

control_plane/src/local_env.rs

@@ -76,7 +76,7 @@ pub struct LocalEnv {
 
     // 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,
+    pub control_plane_api: Option<Url>,
 
     // Control plane upcall API for storage controller.  If set, this will be propagated into the
     // storage controller's configuration.
@@ -133,7 +133,7 @@ pub struct NeonLocalInitConf {
     pub storage_controller: Option<NeonStorageControllerConf>,
     pub pageservers: Vec<NeonLocalInitPageserverConf>,
     pub safekeepers: Vec<SafekeeperConf>,
-    pub control_plane_api: Option<Url>,
+    pub control_plane_api: Option<Option<Url>>,
     pub control_plane_compute_hook_api: Option<Option<Url>>,
 }
 
@@ -180,7 +180,7 @@ impl NeonStorageControllerConf {
     const DEFAULT_MAX_WARMING_UP_INTERVAL: std::time::Duration = std::time::Duration::from_secs(30);
 
     // Very tight heartbeat interval to speed up tests
-    const DEFAULT_HEARTBEAT_INTERVAL: std::time::Duration = std::time::Duration::from_millis(1000);
+    const DEFAULT_HEARTBEAT_INTERVAL: std::time::Duration = std::time::Duration::from_millis(100);
 }
 
 impl Default for NeonStorageControllerConf {
@@ -535,7 +535,7 @@ impl LocalEnv {
                 storage_controller,
                 pageservers,
                 safekeepers,
-                control_plane_api: control_plane_api.unwrap(),
+                control_plane_api,
                 control_plane_compute_hook_api,
                 branch_name_mappings,
             }
@@ -638,7 +638,7 @@ impl LocalEnv {
                 storage_controller: self.storage_controller.clone(),
                 pageservers: vec![], // it's skip_serializing anyway
                 safekeepers: self.safekeepers.clone(),
-                control_plane_api: Some(self.control_plane_api.clone()),
+                control_plane_api: self.control_plane_api.clone(),
                 control_plane_compute_hook_api: self.control_plane_compute_hook_api.clone(),
                 branch_name_mappings: self.branch_name_mappings.clone(),
             },
@@ -768,7 +768,7 @@ impl LocalEnv {
             storage_controller: storage_controller.unwrap_or_default(),
             pageservers: pageservers.iter().map(Into::into).collect(),
             safekeepers,
-            control_plane_api: control_plane_api.unwrap(),
+            control_plane_api: control_plane_api.unwrap_or_default(),
             control_plane_compute_hook_api: control_plane_compute_hook_api.unwrap_or_default(),
             branch_name_mappings: Default::default(),
         };

control_plane/src/pageserver.rs

@@ -95,19 +95,21 @@ impl PageServerNode {
 
         let mut overrides = vec![pg_distrib_dir_param, broker_endpoint_param];
 
-        overrides.push(format!(
-            "control_plane_api='{}'",
-            self.env.control_plane_api.as_str()
-        ));
+        if let Some(control_plane_api) = &self.env.control_plane_api {
+            overrides.push(format!(
+                "control_plane_api='{}'",
+                control_plane_api.as_str()
+            ));
 
-        // Storage controller uses the same auth as pageserver: if JWT is enabled
-        // for us, we will also need it to talk to them.
-        if matches!(conf.http_auth_type, AuthType::NeonJWT) {
-            let jwt_token = self
-                .env
-                .generate_auth_token(&Claims::new(None, Scope::GenerationsApi))
-                .unwrap();
-            overrides.push(format!("control_plane_api_token='{}'", jwt_token));
+            // Storage controller uses the same auth as pageserver: if JWT is enabled
+            // for us, we will also need it to talk to them.
+            if matches!(conf.http_auth_type, AuthType::NeonJWT) {
+                let jwt_token = self
+                    .env
+                    .generate_auth_token(&Claims::new(None, Scope::GenerationsApi))
+                    .unwrap();
+                overrides.push(format!("control_plane_api_token='{}'", jwt_token));
+            }
         }
 
         if !conf.other.contains_key("remote_storage") {

control_plane/src/storage_controller.rs

@@ -338,7 +338,7 @@ impl StorageController {
                         .port(),
                 )
             } else {
-                let listen_url = self.env.control_plane_api.clone();
+                let listen_url = self.env.control_plane_api.clone().unwrap();
 
                 let listen = format!(
                     "{}:{}",
@@ -708,7 +708,7 @@ impl StorageController {
         } else {
             // The configured URL has the /upcall path prefix for pageservers to use: we will strip that out
             // for general purpose API access.
-            let listen_url = self.env.control_plane_api.clone();
+            let listen_url = self.env.control_plane_api.clone().unwrap();
             Url::from_str(&format!(
                 "http://{}:{}/{path}",
                 listen_url.host_str().unwrap(),

control_plane/storcon_cli/src/main.rs

@@ -5,8 +5,7 @@ use clap::{Parser, Subcommand};
 use pageserver_api::{
     controller_api::{
         AvailabilityZone, NodeAvailabilityWrapper, NodeDescribeResponse, NodeShardResponse,
-        SafekeeperDescribeResponse, ShardSchedulingPolicy, TenantCreateRequest,
-        TenantDescribeResponse, TenantPolicyRequest,
+        ShardSchedulingPolicy, TenantCreateRequest, TenantDescribeResponse, TenantPolicyRequest,
     },
     models::{
         EvictionPolicy, EvictionPolicyLayerAccessThreshold, LocationConfigSecondary,
@@ -212,8 +211,6 @@ enum Command {
         #[arg(long)]
         timeout: humantime::Duration,
     },
-    /// List safekeepers known to the storage controller
-    Safekeepers {},
 }
 
 #[derive(Parser)]
@@ -1023,31 +1020,6 @@ async fn main() -> anyhow::Result<()> {
                 "Fill was cancelled for node {node_id}. Schedulling policy is now {final_policy:?}"
             );
         }
-        Command::Safekeepers {} => {
-            let mut resp = storcon_client
-                .dispatch::<(), Vec<SafekeeperDescribeResponse>>(
-                    Method::GET,
-                    "control/v1/safekeeper".to_string(),
-                    None,
-                )
-                .await?;
-
-            resp.sort_by(|a, b| a.id.cmp(&b.id));
-
-            let mut table = comfy_table::Table::new();
-            table.set_header(["Id", "Version", "Host", "Port", "Http Port", "AZ Id"]);
-            for sk in resp {
-                table.add_row([
-                    format!("{}", sk.id),
-                    format!("{}", sk.version),
-                    sk.host,
-                    format!("{}", sk.port),
-                    format!("{}", sk.http_port),
-                    sk.availability_zone_id.to_string(),
-                ]);
-            }
-            println!("{table}");
-        }
     }
 
     Ok(())

libs/pageserver_api/src/controller_api.rs

@@ -372,23 +372,6 @@ pub struct MetadataHealthListOutdatedResponse {
     pub health_records: Vec<MetadataHealthRecord>,
 }
 
-/// Publicly exposed safekeeper description
-///
-/// The `active` flag which we have in the DB is not included on purpose: it is deprecated.
-#[derive(Serialize, Deserialize, Clone)]
-pub struct SafekeeperDescribeResponse {
-    pub id: NodeId,
-    pub region_id: String,
-    /// 1 is special, it means just created (not currently posted to storcon).
-    /// Zero or negative is not really expected.
-    /// Otherwise the number from `release-$(number_of_commits_on_branch)` tag.
-    pub version: i64,
-    pub host: String,
-    pub port: i32,
-    pub http_port: i32,
-    pub availability_zone_id: String,
-}
-
 #[cfg(test)]
 mod test {
     use super::*;

libs/pageserver_api/src/models.rs

@@ -6,7 +6,6 @@ pub mod utilization;
 use camino::Utf8PathBuf;
 pub use utilization::PageserverUtilization;
 
-use core::ops::Range;
 use std::{
     collections::HashMap,
     fmt::Display,
@@ -29,7 +28,6 @@ use utils::{
 };
 
 use crate::{
-    key::Key,
     reltag::RelTag,
     shard::{ShardCount, ShardStripeSize, TenantShardId},
 };
@@ -212,68 +210,6 @@ pub enum TimelineState {
     Broken { reason: String, backtrace: String },
 }
 
-#[serde_with::serde_as]
-#[derive(Debug, Clone, serde::Deserialize, serde::Serialize)]
-pub struct CompactLsnRange {
-    pub start: Lsn,
-    pub end: Lsn,
-}
-
-#[serde_with::serde_as]
-#[derive(Debug, Clone, serde::Deserialize, serde::Serialize)]
-pub struct CompactKeyRange {
-    #[serde_as(as = "serde_with::DisplayFromStr")]
-    pub start: Key,
-    #[serde_as(as = "serde_with::DisplayFromStr")]
-    pub end: Key,
-}
-
-impl From<Range<Lsn>> for CompactLsnRange {
-    fn from(range: Range<Lsn>) -> Self {
-        Self {
-            start: range.start,
-            end: range.end,
-        }
-    }
-}
-
-impl From<Range<Key>> for CompactKeyRange {
-    fn from(range: Range<Key>) -> Self {
-        Self {
-            start: range.start,
-            end: range.end,
-        }
-    }
-}
-
-impl From<CompactLsnRange> for Range<Lsn> {
-    fn from(range: CompactLsnRange) -> Self {
-        range.start..range.end
-    }
-}
-
-impl From<CompactKeyRange> for Range<Key> {
-    fn from(range: CompactKeyRange) -> Self {
-        range.start..range.end
-    }
-}
-
-impl CompactLsnRange {
-    pub fn above(lsn: Lsn) -> Self {
-        Self {
-            start: lsn,
-            end: Lsn::MAX,
-        }
-    }
-}
-
-#[derive(Debug, Clone, Serialize)]
-pub struct CompactInfoResponse {
-    pub compact_key_range: Option<CompactKeyRange>,
-    pub compact_lsn_range: Option<CompactLsnRange>,
-    pub sub_compaction: bool,
-}
-
 #[derive(Serialize, Deserialize, Clone)]
 pub struct TimelineCreateRequest {
     pub new_timeline_id: TimelineId,

libs/postgres_ffi/src/wal_generator.rs

@@ -106,11 +106,11 @@ impl<R: RecordGenerator> WalGenerator<R> {
     const TIMELINE_ID: u32 = 1;
 
     /// Creates a new WAL generator with the given record generator.
-    pub fn new(record_generator: R, start_lsn: Lsn) -> WalGenerator<R> {
+    pub fn new(record_generator: R) -> WalGenerator<R> {
         Self {
             record_generator,
-            lsn: start_lsn,
-            prev_lsn: start_lsn,
+            lsn: Lsn(0),
+            prev_lsn: Lsn(0),
         }
     }
 

libs/proxy/postgres-protocol2/Cargo.toml

@@ -1,7 +1,7 @@
 [package]
 name = "postgres-protocol2"
 version = "0.1.0"
-edition = "2021"
+edition = "2018"
 license = "MIT/Apache-2.0"
 
 [dependencies]

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

@@ -9,7 +9,8 @@
 //!
 //! This library assumes that the `client_encoding` backend parameter has been
 //! set to `UTF8`. It will most likely not behave properly if that is not the case.
-#![warn(missing_docs, clippy::all)]
+#![doc(html_root_url = "https://docs.rs/postgres-protocol/0.6")]
+#![warn(missing_docs, rust_2018_idioms, clippy::all)]
 
 use byteorder::{BigEndian, ByteOrder};
 use bytes::{BufMut, BytesMut};

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

@@ -3,6 +3,7 @@
 
 use byteorder::{BigEndian, ByteOrder};
 use bytes::{Buf, BufMut, BytesMut};
+use std::convert::TryFrom;
 use std::error::Error;
 use std::io;
 use std::marker;

libs/proxy/postgres-types2/Cargo.toml

@@ -1,7 +1,7 @@
 [package]
 name = "postgres-types2"
 version = "0.1.0"
-edition = "2021"
+edition = "2018"
 license = "MIT/Apache-2.0"
 
 [dependencies]

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

@@ -2,7 +2,8 @@
 //!
 //! This crate is used by the `tokio-postgres` and `postgres` crates. You normally don't need to depend directly on it
 //! unless you want to define your own `ToSql` or `FromSql` definitions.
-#![warn(clippy::all, missing_docs)]
+#![doc(html_root_url = "https://docs.rs/postgres-types/0.2")]
+#![warn(clippy::all, rust_2018_idioms, missing_docs)]
 
 use fallible_iterator::FallibleIterator;
 use postgres_protocol2::types;

libs/proxy/tokio-postgres2/Cargo.toml

@@ -1,7 +1,7 @@
 [package]
 name = "tokio-postgres2"
 version = "0.1.0"
-edition = "2021"
+edition = "2018"
 license = "MIT/Apache-2.0"
 
 [dependencies]

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

@@ -1,5 +1,5 @@
 //! An asynchronous, pipelined, PostgreSQL client.
-#![warn(clippy::all)]
+#![warn(rust_2018_idioms, clippy::all)]
 
 pub use crate::cancel_token::CancelToken;
 pub use crate::client::{Client, SocketConfig};

libs/utils/Cargo.toml

@@ -15,20 +15,17 @@ arc-swap.workspace = true
 sentry.workspace = true
 async-compression.workspace = true
 anyhow.workspace = true
-backtrace.workspace = true
 bincode.workspace = true
 bytes.workspace = true
 camino.workspace = true
 chrono.workspace = true
 diatomic-waker.workspace = true
-flate2.workspace = true
 git-version.workspace = true
 hex = { workspace = true, features = ["serde"] }
 humantime.workspace = true
 hyper0 = { workspace = true, features = ["full"] }
-itertools.workspace = true
 fail.workspace = true
-futures = { workspace = true }
+futures = { workspace = true}
 jemalloc_pprof.workspace = true
 jsonwebtoken.workspace = true
 nix.workspace = true

libs/utils/src/http/endpoint.rs

@@ -1,22 +1,15 @@
 use crate::auth::{AuthError, Claims, SwappableJwtAuth};
 use crate::http::error::{api_error_handler, route_error_handler, ApiError};
 use crate::http::request::{get_query_param, parse_query_param};
-use crate::pprof;
-use ::pprof::protos::Message as _;
-use ::pprof::ProfilerGuardBuilder;
 use anyhow::{anyhow, Context};
-use bytes::{Bytes, BytesMut};
 use hyper::header::{HeaderName, AUTHORIZATION, CONTENT_DISPOSITION};
 use hyper::http::HeaderValue;
 use hyper::Method;
 use hyper::{header::CONTENT_TYPE, Body, Request, Response};
 use metrics::{register_int_counter, Encoder, IntCounter, TextEncoder};
 use once_cell::sync::Lazy;
-use regex::Regex;
 use routerify::ext::RequestExt;
 use routerify::{Middleware, RequestInfo, Router, RouterBuilder};
-use tokio::sync::{mpsc, Mutex};
-use tokio_stream::wrappers::ReceiverStream;
 use tokio_util::io::ReaderStream;
 use tracing::{debug, info, info_span, warn, Instrument};
 
@@ -25,6 +18,11 @@ use std::io::Write as _;
 use std::str::FromStr;
 use std::time::Duration;
 
+use bytes::{Bytes, BytesMut};
+use pprof::protos::Message as _;
+use tokio::sync::{mpsc, Mutex};
+use tokio_stream::wrappers::ReceiverStream;
+
 static SERVE_METRICS_COUNT: Lazy<IntCounter> = Lazy::new(|| {
     register_int_counter!(
         "libmetrics_metric_handler_requests_total",
@@ -367,7 +365,7 @@ pub async fn profile_cpu_handler(req: Request<Body>) -> Result<Response<Body>, A
 
     // Take the profile.
     let report = tokio::task::spawn_blocking(move || {
-        let guard = ProfilerGuardBuilder::default()
+        let guard = pprof::ProfilerGuardBuilder::default()
             .frequency(frequency_hz)
             .blocklist(&["libc", "libgcc", "pthread", "vdso"])
             .build()?;
@@ -459,34 +457,10 @@ pub async fn profile_heap_handler(req: Request<Body>) -> Result<Response<Body>,
         }
 
         Format::Pprof => {
-            let data = tokio::task::spawn_blocking(move || {
-                let bytes = prof_ctl.dump_pprof()?;
-                // Symbolize the profile.
-                // TODO: consider moving this upstream to jemalloc_pprof and avoiding the
-                // serialization roundtrip.
-                static STRIP_FUNCTIONS: Lazy<Vec<(Regex, bool)>> = Lazy::new(|| {
-                    // Functions to strip from profiles. If true, also remove child frames.
-                    vec![
-                        (Regex::new("^__rust").unwrap(), false),
-                        (Regex::new("^_start$").unwrap(), false),
-                        (Regex::new("^irallocx_prof").unwrap(), true),
-                        (Regex::new("^prof_alloc_prep").unwrap(), true),
-                        (Regex::new("^std::rt::lang_start").unwrap(), false),
-                        (Regex::new("^std::sys::backtrace::__rust").unwrap(), false),
-                    ]
-                });
-                let profile = pprof::decode(&bytes)?;
-                let profile = pprof::symbolize(profile)?;
-                let profile = pprof::strip_locations(
-                    profile,
-                    &["libc", "libgcc", "pthread", "vdso"],
-                    &STRIP_FUNCTIONS,
-                );
-                pprof::encode(&profile)
-            })
-            .await
-            .map_err(|join_err| ApiError::InternalServerError(join_err.into()))?
-            .map_err(ApiError::InternalServerError)?;
+            let data = tokio::task::spawn_blocking(move || prof_ctl.dump_pprof())
+                .await
+                .map_err(|join_err| ApiError::InternalServerError(join_err.into()))?
+                .map_err(ApiError::InternalServerError)?;
             Response::builder()
                 .status(200)
                 .header(CONTENT_TYPE, "application/octet-stream")

libs/utils/src/lib.rs

@@ -96,8 +96,6 @@ pub mod circuit_breaker;
 
 pub mod try_rcu;
 
-pub mod pprof;
-
 // Re-export used in macro. Avoids adding git-version as dep in target crates.
 #[doc(hidden)]
 pub use git_version;

libs/utils/src/pprof.rs

@@ -1,190 +0,0 @@
-use flate2::write::{GzDecoder, GzEncoder};
-use flate2::Compression;
-use itertools::Itertools as _;
-use once_cell::sync::Lazy;
-use pprof::protos::{Function, Line, Message as _, Profile};
-use regex::Regex;
-
-use std::borrow::Cow;
-use std::collections::{HashMap, HashSet};
-use std::ffi::c_void;
-use std::io::Write as _;
-
-/// Decodes a gzip-compressed Protobuf-encoded pprof profile.
-pub fn decode(bytes: &[u8]) -> anyhow::Result<Profile> {
-    let mut gz = GzDecoder::new(Vec::new());
-    gz.write_all(bytes)?;
-    Ok(Profile::parse_from_bytes(&gz.finish()?)?)
-}
-
-/// Encodes a pprof profile as gzip-compressed Protobuf.
-pub fn encode(profile: &Profile) -> anyhow::Result<Vec<u8>> {
-    let mut gz = GzEncoder::new(Vec::new(), Compression::default());
-    profile.write_to_writer(&mut gz)?;
-    Ok(gz.finish()?)
-}
-
-/// Symbolizes a pprof profile using the current binary.
-pub fn symbolize(mut profile: Profile) -> anyhow::Result<Profile> {
-    if !profile.function.is_empty() {
-        return Ok(profile); // already symbolized
-    }
-
-    // Collect function names.
-    let mut functions: HashMap<String, Function> = HashMap::new();
-    let mut strings: HashMap<String, i64> = profile
-        .string_table
-        .into_iter()
-        .enumerate()
-        .map(|(i, s)| (s, i as i64))
-        .collect();
-
-    // Helper to look up or register a string.
-    let mut string_id = |s: &str| -> i64 {
-        // Don't use .entry() to avoid unnecessary allocations.
-        if let Some(id) = strings.get(s) {
-            return *id;
-        }
-        let id = strings.len() as i64;
-        strings.insert(s.to_string(), id);
-        id
-    };
-
-    for loc in &mut profile.location {
-        if !loc.line.is_empty() {
-            continue;
-        }
-
-        // Resolve the line and function for each location.
-        backtrace::resolve(loc.address as *mut c_void, |symbol| {
-            let Some(symname) = symbol.name() else {
-                return;
-            };
-            let mut name = symname.to_string();
-
-            // Strip the Rust monomorphization suffix from the symbol name.
-            static SUFFIX_REGEX: Lazy<Regex> =
-                Lazy::new(|| Regex::new("::h[0-9a-f]{16}$").expect("invalid regex"));
-            if let Some(m) = SUFFIX_REGEX.find(&name) {
-                name.truncate(m.start());
-            }
-
-            let function_id = match functions.get(&name) {
-                Some(function) => function.id,
-                None => {
-                    let id = functions.len() as u64 + 1;
-                    let system_name = String::from_utf8_lossy(symname.as_bytes());
-                    let filename = symbol
-                        .filename()
-                        .map(|path| path.to_string_lossy())
-                        .unwrap_or(Cow::Borrowed(""));
-                    let function = Function {
-                        id,
-                        name: string_id(&name),
-                        system_name: string_id(&system_name),
-                        filename: string_id(&filename),
-                        ..Default::default()
-                    };
-                    functions.insert(name, function);
-                    id
-                }
-            };
-            loc.line.push(Line {
-                function_id,
-                line: symbol.lineno().unwrap_or(0) as i64,
-                ..Default::default()
-            });
-        });
-    }
-
-    // Store the resolved functions, and mark the mapping as resolved.
-    profile.function = functions.into_values().sorted_by_key(|f| f.id).collect();
-    profile.string_table = strings
-        .into_iter()
-        .sorted_by_key(|(_, i)| *i)
-        .map(|(s, _)| s)
-        .collect();
-
-    for mapping in &mut profile.mapping {
-        mapping.has_functions = true;
-        mapping.has_filenames = true;
-    }
-
-    Ok(profile)
-}
-
-/// Strips locations (stack frames) matching the given mappings (substring) or function names
-/// (regex). The function bool specifies whether child frames should be stripped as well.
-///
-/// The string definitions are left behind in the profile for simplicity, to avoid rewriting all
-/// string references.
-pub fn strip_locations(
-    mut profile: Profile,
-    mappings: &[&str],
-    functions: &[(Regex, bool)],
-) -> Profile {
-    // Strip mappings.
-    let mut strip_mappings: HashSet<u64> = HashSet::new();
-
-    profile.mapping.retain(|mapping| {
-        let Some(name) = profile.string_table.get(mapping.filename as usize) else {
-            return true;
-        };
-        if mappings.iter().any(|substr| name.contains(substr)) {
-            strip_mappings.insert(mapping.id);
-            return false;
-        }
-        true
-    });
-
-    // Strip functions.
-    let mut strip_functions: HashMap<u64, bool> = HashMap::new();
-
-    profile.function.retain(|function| {
-        let Some(name) = profile.string_table.get(function.name as usize) else {
-            return true;
-        };
-        for (regex, strip_children) in functions {
-            if regex.is_match(name) {
-                strip_functions.insert(function.id, *strip_children);
-                return false;
-            }
-        }
-        true
-    });
-
-    // Strip locations. The bool specifies whether child frames should be stripped too.
-    let mut strip_locations: HashMap<u64, bool> = HashMap::new();
-
-    profile.location.retain(|location| {
-        for line in &location.line {
-            if let Some(strip_children) = strip_functions.get(&line.function_id) {
-                strip_locations.insert(location.id, *strip_children);
-                return false;
-            }
-        }
-        if strip_mappings.contains(&location.mapping_id) {
-            strip_locations.insert(location.id, false);
-            return false;
-        }
-        true
-    });
-
-    // Strip sample locations.
-    for sample in &mut profile.sample {
-        // First, find the uppermost function with child removal and truncate the stack.
-        if let Some(truncate) = sample
-            .location_id
-            .iter()
-            .rposition(|id| strip_locations.get(id) == Some(&true))
-        {
-            sample.location_id.drain(..=truncate);
-        }
-        // Next, strip any individual frames without child removal.
-        sample
-            .location_id
-            .retain(|id| !strip_locations.contains_key(id));
-    }
-
-    profile
-}

pageserver/Cargo.toml

@@ -94,7 +94,6 @@ procfs.workspace = true
 [dev-dependencies]
 criterion.workspace = true
 hex-literal.workspace = true
-serial_test.workspace = true
 tokio = { workspace = true, features = ["process", "sync", "fs", "rt", "io-util", "time", "test-util"] }
 indoc.workspace = true
 

pageserver/src/http/routes.rs

@@ -97,8 +97,8 @@ use crate::tenant::{LogicalSizeCalculationCause, PageReconstructError};
 use crate::DEFAULT_PG_VERSION;
 use crate::{disk_usage_eviction_task, tenant};
 use pageserver_api::models::{
-    CompactInfoResponse, StatusResponse, TenantConfigRequest, TenantInfo, TimelineCreateRequest,
-    TimelineGcRequest, TimelineInfo,
+    StatusResponse, TenantConfigRequest, TenantInfo, TimelineCreateRequest, TimelineGcRequest,
+    TimelineInfo,
 };
 use utils::{
     auth::SwappableJwtAuth,
@@ -2039,34 +2039,6 @@ async fn timeline_cancel_compact_handler(
     .await
 }
 
-// Get compact info of a timeline
-async fn timeline_compact_info_handler(
-    request: Request<Body>,
-    _cancel: CancellationToken,
-) -> Result<Response<Body>, ApiError> {
-    let tenant_shard_id: TenantShardId = parse_request_param(&request, "tenant_shard_id")?;
-    let timeline_id: TimelineId = parse_request_param(&request, "timeline_id")?;
-    check_permission(&request, Some(tenant_shard_id.tenant_id))?;
-    let state = get_state(&request);
-    async {
-        let tenant = state
-            .tenant_manager
-            .get_attached_tenant_shard(tenant_shard_id)?;
-        let res = tenant.get_scheduled_compaction_tasks(timeline_id);
-        let mut resp = Vec::new();
-        for item in res {
-            resp.push(CompactInfoResponse {
-                compact_key_range: item.compact_key_range,
-                compact_lsn_range: item.compact_lsn_range,
-                sub_compaction: item.sub_compaction,
-            });
-        }
-        json_response(StatusCode::OK, resp)
-    }
-    .instrument(info_span!("timeline_compact_info", tenant_id = %tenant_shard_id.tenant_id, shard_id = %tenant_shard_id.shard_slug(), %timeline_id))
-    .await
-}
-
 // Run compaction immediately on given timeline.
 async fn timeline_compact_handler(
     mut request: Request<Body>,
@@ -3428,10 +3400,6 @@ pub fn make_router(
             "/v1/tenant/:tenant_shard_id/timeline/:timeline_id/do_gc",
             |r| api_handler(r, timeline_gc_handler),
         )
-        .get(
-            "/v1/tenant/:tenant_shard_id/timeline/:timeline_id/compact",
-            |r| api_handler(r, timeline_compact_info_handler),
-        )
         .put(
             "/v1/tenant/:tenant_shard_id/timeline/:timeline_id/compact",
             |r| api_handler(r, timeline_compact_handler),

pageserver/src/tenant.rs

@@ -3122,23 +3122,6 @@ impl Tenant {
         }
     }
 
-    pub(crate) fn get_scheduled_compaction_tasks(
-        &self,
-        timeline_id: TimelineId,
-    ) -> Vec<CompactOptions> {
-        use itertools::Itertools;
-        let guard = self.scheduled_compaction_tasks.lock().unwrap();
-        guard
-            .get(&timeline_id)
-            .map(|tline_pending_tasks| {
-                tline_pending_tasks
-                    .iter()
-                    .map(|x| x.options.clone())
-                    .collect_vec()
-            })
-            .unwrap_or_default()
-    }
-
     /// Schedule a compaction task for a timeline.
     pub(crate) async fn schedule_compaction(
         &self,
@@ -5776,13 +5759,13 @@ mod tests {
     use timeline::{CompactOptions, DeltaLayerTestDesc};
     use utils::id::TenantId;
 
-    #[cfg(feature = "testing")]
-    use models::CompactLsnRange;
     #[cfg(feature = "testing")]
     use pageserver_api::record::NeonWalRecord;
     #[cfg(feature = "testing")]
     use timeline::compaction::{KeyHistoryRetention, KeyLogAtLsn};
     #[cfg(feature = "testing")]
+    use timeline::CompactLsnRange;
+    #[cfg(feature = "testing")]
     use timeline::GcInfo;
 
     static TEST_KEY: Lazy<Key> =
@@ -9651,7 +9634,7 @@ mod tests {
     #[cfg(feature = "testing")]
     #[tokio::test]
     async fn test_simple_bottom_most_compaction_on_branch() -> anyhow::Result<()> {
-        use models::CompactLsnRange;
+        use timeline::CompactLsnRange;
 
         let harness = TenantHarness::create("test_simple_bottom_most_compaction_on_branch").await?;
         let (tenant, ctx) = harness.load().await;

pageserver/src/tenant/checks.rs

@@ -1,15 +1,12 @@
 use std::collections::BTreeSet;
 
 use itertools::Itertools;
-use pageserver_compaction::helpers::overlaps_with;
 
 use super::storage_layer::LayerName;
 
 /// Checks whether a layer map is valid (i.e., is a valid result of the current compaction algorithm if nothing goes wrong).
 ///
-/// The function implements a fast path check and a slow path check.
-///
-/// The fast path checks if we can split the LSN range of a delta layer only at the LSNs of the delta layers. For example,
+/// The function checks if we can split the LSN range of a delta layer only at the LSNs of the delta layers. For example,
 ///
 /// ```plain
 /// |       |                 |       |
@@ -28,47 +25,31 @@ use super::storage_layer::LayerName;
 /// |       |    |   4   |    |       |
 ///
 /// If layer 2 and 4 contain the same single key, this is also a valid layer map.
-///
-/// However, if a partial compaction is still going on, it is possible that we get a layer map not satisfying the above condition.
-/// Therefore, we fallback to simply check if any of the two delta layers overlap. (See "A slow path...")
 pub fn check_valid_layermap(metadata: &[LayerName]) -> Option<String> {
     let mut lsn_split_point = BTreeSet::new(); // TODO: use a better data structure (range tree / range set?)
     let mut all_delta_layers = Vec::new();
     for name in metadata {
         if let LayerName::Delta(layer) = name {
-            all_delta_layers.push(layer.clone());
+            if layer.key_range.start.next() != layer.key_range.end {
+                all_delta_layers.push(layer.clone());
+            }
         }
     }
     for layer in &all_delta_layers {
-        if layer.key_range.start.next() != layer.key_range.end {
-            let lsn_range = &layer.lsn_range;
-            lsn_split_point.insert(lsn_range.start);
-            lsn_split_point.insert(lsn_range.end);
-        }
+        let lsn_range = &layer.lsn_range;
+        lsn_split_point.insert(lsn_range.start);
+        lsn_split_point.insert(lsn_range.end);
     }
-    for (idx, layer) in all_delta_layers.iter().enumerate() {
-        if layer.key_range.start.next() == layer.key_range.end {
-            continue;
-        }
+    for layer in &all_delta_layers {
         let lsn_range = layer.lsn_range.clone();
         let intersects = lsn_split_point.range(lsn_range).collect_vec();
         if intersects.len() > 1 {
-            // A slow path to check if the layer intersects with any other delta layer.
-            for (other_idx, other_layer) in all_delta_layers.iter().enumerate() {
-                if other_idx == idx {
-                    // do not check self intersects with self
-                    continue;
-                }
-                if overlaps_with(&layer.lsn_range, &other_layer.lsn_range)
-                    && overlaps_with(&layer.key_range, &other_layer.key_range)
-                {
-                    let err = format!(
-                            "layer violates the layer map LSN split assumption: layer {} intersects with layer {}",
-                            layer, other_layer
-                        );
-                    return Some(err);
-                }
-            }
+            let err = format!(
+                "layer violates the layer map LSN split assumption: layer {} intersects with LSN [{}]",
+                layer,
+                intersects.into_iter().map(|lsn| lsn.to_string()).join(", ")
+            );
+            return Some(err);
         }
     }
     None

pageserver/src/tenant/storage_layer/layer.rs

@@ -1,6 +1,5 @@
 use anyhow::Context;
 use camino::{Utf8Path, Utf8PathBuf};
-use once_cell::sync::Lazy;
 use pageserver_api::keyspace::KeySpace;
 use pageserver_api::models::HistoricLayerInfo;
 use pageserver_api::shard::{ShardIdentity, ShardIndex, TenantShardId};
@@ -2097,36 +2096,6 @@ impl Default for LayerImplMetrics {
 }
 
 impl LayerImplMetrics {
-    /// Resets the layer metrics to 0, for use in tests. Since this is a global static, metrics will
-    /// be shared across tests, and must be reset in each test case.
-    #[cfg(test)]
-    fn reset(&self) {
-        // Destructure to error on new fields.
-        let LayerImplMetrics {
-            started_evictions,
-            completed_evictions,
-            cancelled_evictions,
-            started_deletes,
-            completed_deletes,
-            failed_deletes,
-            rare_counters,
-            inits_cancelled,
-            redownload_after,
-            time_to_evict,
-        } = self;
-
-        started_evictions.reset();
-        completed_evictions.reset();
-        cancelled_evictions.values().for_each(|c| c.reset());
-        started_deletes.reset();
-        completed_deletes.reset();
-        failed_deletes.values().for_each(|c| c.reset());
-        rare_counters.values().for_each(|c| c.reset());
-        inits_cancelled.reset();
-        redownload_after.local().clear();
-        time_to_evict.local().clear();
-    }
-
     fn inc_started_evictions(&self) {
         self.started_evictions.inc();
     }
@@ -2278,4 +2247,5 @@ impl RareEvent {
     }
 }
 
-pub(crate) static LAYER_IMPL_METRICS: Lazy<LayerImplMetrics> = Lazy::new(LayerImplMetrics::default);
+pub(crate) static LAYER_IMPL_METRICS: once_cell::sync::Lazy<LayerImplMetrics> =
+    once_cell::sync::Lazy::new(LayerImplMetrics::default);

pageserver/src/tenant/storage_layer/layer/tests.rs

@@ -1,7 +1,6 @@
 use std::time::UNIX_EPOCH;
 
 use pageserver_api::key::CONTROLFILE_KEY;
-use serial_test::serial;
 use tokio::task::JoinSet;
 use utils::{
     completion::{self, Completion},
@@ -22,10 +21,7 @@ const FOREVER: std::time::Duration = std::time::Duration::from_secs(ADVANCE.as_s
 
 /// Demonstrate the API and resident -> evicted -> resident -> deleted transitions.
 #[tokio::test]
-#[serial]
 async fn smoke_test() {
-    LAYER_IMPL_METRICS.reset();
-
     let handle = tokio::runtime::Handle::current();
 
     let h = TenantHarness::create("smoke_test").await.unwrap();
@@ -202,10 +198,7 @@ async fn smoke_test() {
 /// This test demonstrates a previous hang when a eviction and deletion were requested at the same
 /// time. Now both of them complete per Arc drop semantics.
 #[tokio::test(start_paused = true)]
-#[serial]
 async fn evict_and_wait_on_wanted_deleted() {
-    LAYER_IMPL_METRICS.reset();
-
     // this is the runtime on which Layer spawns the blocking tasks on
     let handle = tokio::runtime::Handle::current();
 
@@ -282,10 +275,7 @@ async fn evict_and_wait_on_wanted_deleted() {
 /// This test ensures we are able to read the layer while the layer eviction has been
 /// started but not completed.
 #[test]
-#[serial]
 fn read_wins_pending_eviction() {
-    LAYER_IMPL_METRICS.reset();
-
     let rt = tokio::runtime::Builder::new_current_thread()
         .max_blocking_threads(1)
         .enable_all()
@@ -405,7 +395,6 @@ fn read_wins_pending_eviction() {
 
 /// Use failpoint to delay an eviction starting to get a VersionCheckFailed.
 #[test]
-#[serial]
 fn multiple_pending_evictions_in_order() {
     let name = "multiple_pending_evictions_in_order";
     let in_order = true;
@@ -414,7 +403,6 @@ fn multiple_pending_evictions_in_order() {
 
 /// Use failpoint to reorder later eviction before first to get a UnexpectedEvictedState.
 #[test]
-#[serial]
 fn multiple_pending_evictions_out_of_order() {
     let name = "multiple_pending_evictions_out_of_order";
     let in_order = false;
@@ -422,8 +410,6 @@ fn multiple_pending_evictions_out_of_order() {
 }
 
 fn multiple_pending_evictions_scenario(name: &'static str, in_order: bool) {
-    LAYER_IMPL_METRICS.reset();
-
     let rt = tokio::runtime::Builder::new_current_thread()
         .max_blocking_threads(1)
         .enable_all()
@@ -601,10 +587,7 @@ fn multiple_pending_evictions_scenario(name: &'static str, in_order: bool) {
 /// disk but the layer internal state says it has not been initialized. Futhermore, it allows us to
 /// have non-repairing `Layer::is_likely_resident`.
 #[tokio::test(start_paused = true)]
-#[serial]
 async fn cancelled_get_or_maybe_download_does_not_cancel_eviction() {
-    LAYER_IMPL_METRICS.reset();
-
     let handle = tokio::runtime::Handle::current();
     let h = TenantHarness::create("cancelled_get_or_maybe_download_does_not_cancel_eviction")
         .await
@@ -682,8 +665,8 @@ async fn cancelled_get_or_maybe_download_does_not_cancel_eviction() {
 }
 
 #[tokio::test(start_paused = true)]
-#[serial]
 async fn evict_and_wait_does_not_wait_for_download() {
+    // let handle = tokio::runtime::Handle::current();
     let h = TenantHarness::create("evict_and_wait_does_not_wait_for_download")
         .await
         .unwrap();
@@ -776,13 +759,10 @@ async fn evict_and_wait_does_not_wait_for_download() {
 ///
 /// Also checks that the same does not happen on a non-evicted layer (regression test).
 #[tokio::test(start_paused = true)]
-#[serial]
 async fn eviction_cancellation_on_drop() {
     use bytes::Bytes;
     use pageserver_api::value::Value;
 
-    LAYER_IMPL_METRICS.reset();
-
     // this is the runtime on which Layer spawns the blocking tasks on
     let handle = tokio::runtime::Handle::current();
 

pageserver/src/tenant/timeline.rs

@@ -31,9 +31,9 @@ use pageserver_api::{
     },
     keyspace::{KeySpaceAccum, KeySpaceRandomAccum, SparseKeyPartitioning},
     models::{
-        CompactKeyRange, CompactLsnRange, CompactionAlgorithm, CompactionAlgorithmSettings,
-        DownloadRemoteLayersTaskInfo, DownloadRemoteLayersTaskSpawnRequest, EvictionPolicy,
-        InMemoryLayerInfo, LayerMapInfo, LsnLease, TimelineState,
+        CompactionAlgorithm, CompactionAlgorithmSettings, DownloadRemoteLayersTaskInfo,
+        DownloadRemoteLayersTaskSpawnRequest, EvictionPolicy, InMemoryLayerInfo, LayerMapInfo,
+        LsnLease, TimelineState,
     },
     reltag::BlockNumber,
     shard::{ShardIdentity, ShardNumber, TenantShardId},
@@ -788,6 +788,63 @@ pub(crate) struct CompactRequest {
     pub sub_compaction_max_job_size_mb: Option<u64>,
 }
 
+#[serde_with::serde_as]
+#[derive(Debug, Clone, serde::Deserialize)]
+pub(crate) struct CompactLsnRange {
+    pub start: Lsn,
+    pub end: Lsn,
+}
+
+#[serde_with::serde_as]
+#[derive(Debug, Clone, serde::Deserialize)]
+pub(crate) struct CompactKeyRange {
+    #[serde_as(as = "serde_with::DisplayFromStr")]
+    pub start: Key,
+    #[serde_as(as = "serde_with::DisplayFromStr")]
+    pub end: Key,
+}
+
+impl From<Range<Lsn>> for CompactLsnRange {
+    fn from(range: Range<Lsn>) -> Self {
+        Self {
+            start: range.start,
+            end: range.end,
+        }
+    }
+}
+
+impl From<Range<Key>> for CompactKeyRange {
+    fn from(range: Range<Key>) -> Self {
+        Self {
+            start: range.start,
+            end: range.end,
+        }
+    }
+}
+
+impl From<CompactLsnRange> for Range<Lsn> {
+    fn from(range: CompactLsnRange) -> Self {
+        range.start..range.end
+    }
+}
+
+impl From<CompactKeyRange> for Range<Key> {
+    fn from(range: CompactKeyRange) -> Self {
+        range.start..range.end
+    }
+}
+
+impl CompactLsnRange {
+    #[cfg(test)]
+    #[cfg(feature = "testing")]
+    pub fn above(lsn: Lsn) -> Self {
+        Self {
+            start: lsn,
+            end: Lsn::MAX,
+        }
+    }
+}
+
 #[derive(Debug, Clone, Default)]
 pub(crate) struct CompactOptions {
     pub flags: EnumSet<CompactFlags>,

pageserver/src/tenant/timeline/compaction.rs

@@ -29,7 +29,6 @@ use utils::id::TimelineId;
 use crate::context::{AccessStatsBehavior, RequestContext, RequestContextBuilder};
 use crate::page_cache;
 use crate::statvfs::Statvfs;
-use crate::tenant::checks::check_valid_layermap;
 use crate::tenant::remote_timeline_client::WaitCompletionError;
 use crate::tenant::storage_layer::batch_split_writer::{
     BatchWriterResult, SplitDeltaLayerWriter, SplitImageLayerWriter,
@@ -1824,7 +1823,7 @@ impl Timeline {
         // by estimating the amount of files read for a compaction job. We should also partition on LSN.
         let ((dense_ks, sparse_ks), _) = {
             let Ok(partition) = self.partitioning.try_lock() else {
-                bail!("failed to acquire partition lock during gc-compaction");
+                bail!("failed to acquire partition lock");
             };
             partition.clone()
         };
@@ -2157,14 +2156,15 @@ impl Timeline {
 
         // Step 1: construct a k-merge iterator over all layers.
         // Also, verify if the layer map can be split by drawing a horizontal line at every LSN start/end split point.
-        let layer_names = job_desc
-            .selected_layers
-            .iter()
-            .map(|layer| layer.layer_desc().layer_name())
-            .collect_vec();
-        if let Some(err) = check_valid_layermap(&layer_names) {
-            bail!("gc-compaction layer map check failed because {}, cannot proceed with compaction due to potential data loss", err);
-        }
+        // disable the check for now because we need to adjust the check for partial compactions, will enable later.
+        // let layer_names = job_desc
+        //     .selected_layers
+        //     .iter()
+        //     .map(|layer| layer.layer_desc().layer_name())
+        //     .collect_vec();
+        // if let Some(err) = check_valid_layermap(&layer_names) {
+        //     warn!("gc-compaction layer map check failed because {}, this is normal if partial compaction is not finished yet", err);
+        // }
         // The maximum LSN we are processing in this compaction loop
         let end_lsn = job_desc
             .selected_layers
@@ -2546,48 +2546,13 @@ impl Timeline {
         );
 
         // Step 3: Place back to the layer map.
-
-        // First, do a sanity check to ensure the newly-created layer map does not contain overlaps.
-        let all_layers = {
-            let guard = self.layers.read().await;
-            let layer_map = guard.layer_map()?;
-            layer_map.iter_historic_layers().collect_vec()
-        };
-
-        let mut final_layers = all_layers
-            .iter()
-            .map(|layer| layer.layer_name())
-            .collect::<HashSet<_>>();
-        for layer in &layer_selection {
-            final_layers.remove(&layer.layer_desc().layer_name());
-        }
-        for layer in &compact_to {
-            final_layers.insert(layer.layer_desc().layer_name());
-        }
-        let final_layers = final_layers.into_iter().collect_vec();
-
-        // TODO: move this check before we call `finish` on image layer writers. However, this will require us to get the layer name before we finish
-        // the writer, so potentially, we will need a function like `ImageLayerBatchWriter::get_all_pending_layer_keys` to get all the keys that are
-        // in the writer before finalizing the persistent layers. Now we would leave some dangling layers on the disk if the check fails.
-        if let Some(err) = check_valid_layermap(&final_layers) {
-            bail!("gc-compaction layer map check failed after compaction because {}, compaction result not applied to the layer map due to potential data loss", err);
-        }
-
-        // Between the sanity check and this compaction update, there could be new layers being flushed, but it should be fine because we only
-        // operate on L1 layers.
         {
+            // TODO: sanity check if the layer map is valid (i.e., should not have overlaps)
             let mut guard = self.layers.write().await;
             guard
                 .open_mut()?
                 .finish_gc_compaction(&layer_selection, &compact_to, &self.metrics)
         };
-
-        // Schedule an index-only upload to update the `latest_gc_cutoff` in the index_part.json.
-        // Otherwise, after restart, the index_part only contains the old `latest_gc_cutoff` and
-        // find_gc_cutoffs will try accessing things below the cutoff. TODO: ideally, this should
-        // be batched into `schedule_compaction_update`.
-        let disk_consistent_lsn = self.disk_consistent_lsn.load();
-        self.schedule_uploads(disk_consistent_lsn, None)?;
         self.remote_client
             .schedule_compaction_update(&layer_selection, &compact_to)?;
 

pgxn/neon/libpagestore.c

@@ -827,6 +827,7 @@ pageserver_send(shardno_t shard_no, NeonRequest *request)
 	{
 		while (!pageserver_connect(shard_no, shard->n_reconnect_attempts < max_reconnect_attempts ? LOG : ERROR))
 		{
+			HandleMainLoopInterrupts();
 			shard->n_reconnect_attempts += 1;
 		}
 		shard->n_reconnect_attempts = 0;

proxy/src/auth/backend/mod.rs

@@ -678,9 +678,6 @@ mod tests {
         .await
         .unwrap();
 
-        // flush the final server message
-        stream.flush().await.unwrap();
-
         handle.await.unwrap();
     }
 

proxy/src/bin/local_proxy.rs

@@ -271,6 +271,7 @@ fn build_config(args: &LocalProxyCliArgs) -> anyhow::Result<&'static ProxyConfig
     Ok(Box::leak(Box::new(ProxyConfig {
         tls_config: None,
         metric_collection: None,
+        allow_self_signed_compute: false,
         http_config,
         authentication_config: AuthenticationConfig {
             jwks_cache: JwkCache::default(),

proxy/src/bin/pg_sni_router.rs

@@ -229,7 +229,7 @@ async fn ssl_handshake<S: AsyncRead + AsyncWrite + Unpin>(
 
             let (raw, read_buf) = stream.into_inner();
             // TODO: Normally, client doesn't send any data before
-            // server says TLS handshake is ok and read_buf is empty.
+            // server says TLS handshake is ok and read_buf is empy.
             // However, you could imagine pipelining of postgres
             // SSLRequest + TLS ClientHello in one hunk similar to
             // pipelining in our node js driver. We should probably

proxy/src/bin/proxy.rs

@@ -129,6 +129,9 @@ struct ProxyCliArgs {
     /// lock for `connect_compute` api method. example: "shards=32,permits=4,epoch=10m,timeout=1s". (use `permits=0` to disable).
     #[clap(long, default_value = config::ConcurrencyLockOptions::DEFAULT_OPTIONS_CONNECT_COMPUTE_LOCK)]
     connect_compute_lock: String,
+    /// Allow self-signed certificates for compute nodes (for testing)
+    #[clap(long, default_value_t = false, value_parser = clap::builder::BoolishValueParser::new(), action = clap::ArgAction::Set)]
+    allow_self_signed_compute: bool,
     #[clap(flatten)]
     sql_over_http: SqlOverHttpArgs,
     /// timeout for scram authentication protocol
@@ -561,6 +564,9 @@ fn build_config(args: &ProxyCliArgs) -> anyhow::Result<&'static ProxyConfig> {
         _ => bail!("either both or neither tls-key and tls-cert must be specified"),
     };
 
+    if args.allow_self_signed_compute {
+        warn!("allowing self-signed compute certificates");
+    }
     let backup_metric_collection_config = config::MetricBackupCollectionConfig {
         interval: args.metric_backup_collection_interval,
         remote_storage_config: args.metric_backup_collection_remote_storage.clone(),
@@ -635,6 +641,7 @@ fn build_config(args: &ProxyCliArgs) -> anyhow::Result<&'static ProxyConfig> {
     let config = ProxyConfig {
         tls_config,
         metric_collection,
+        allow_self_signed_compute: args.allow_self_signed_compute,
         http_config,
         authentication_config,
         proxy_protocol_v2: args.proxy_protocol_v2,

proxy/src/cancellation.rs

@@ -3,11 +3,8 @@ use std::sync::Arc;
 
 use dashmap::DashMap;
 use ipnet::{IpNet, Ipv4Net, Ipv6Net};
-use once_cell::sync::OnceCell;
-use postgres_client::tls::MakeTlsConnect;
-use postgres_client::CancelToken;
+use postgres_client::{CancelToken, NoTls};
 use pq_proto::CancelKeyData;
-use rustls::crypto::ring;
 use thiserror::Error;
 use tokio::net::TcpStream;
 use tokio::sync::Mutex;
@@ -15,11 +12,9 @@ use tracing::{debug, info};
 use uuid::Uuid;
 
 use crate::auth::{check_peer_addr_is_in_list, IpPattern};
-use crate::compute::load_certs;
 use crate::error::ReportableError;
 use crate::ext::LockExt;
 use crate::metrics::{CancellationRequest, CancellationSource, Metrics};
-use crate::postgres_rustls::MakeRustlsConnect;
 use crate::rate_limiter::LeakyBucketRateLimiter;
 use crate::redis::cancellation_publisher::{
     CancellationPublisher, CancellationPublisherMut, RedisPublisherClient,
@@ -179,10 +174,7 @@ impl<P: CancellationPublisher> CancellationHandler<P> {
                 source: self.from,
                 kind: crate::metrics::CancellationOutcome::Found,
             });
-        info!(
-            "cancelling query per user's request using key {key}, hostname {}, address: {}",
-            cancel_closure.hostname, cancel_closure.socket_addr
-        );
+        info!("cancelling query per user's request using key {key}");
         cancel_closure.try_cancel_query().await
     }
 
@@ -229,8 +221,6 @@ impl<P: CancellationPublisherMut> CancellationHandler<Option<Arc<Mutex<P>>>> {
     }
 }
 
-static TLS_ROOTS: OnceCell<Arc<rustls::RootCertStore>> = OnceCell::new();
-
 /// This should've been a [`std::future::Future`], but
 /// it's impossible to name a type of an unboxed future
 /// (we'd need something like `#![feature(type_alias_impl_trait)]`).
@@ -239,7 +229,6 @@ pub struct CancelClosure {
     socket_addr: SocketAddr,
     cancel_token: CancelToken,
     ip_allowlist: Vec<IpPattern>,
-    hostname: String, // for pg_sni router
 }
 
 impl CancelClosure {
@@ -247,49 +236,17 @@ impl CancelClosure {
         socket_addr: SocketAddr,
         cancel_token: CancelToken,
         ip_allowlist: Vec<IpPattern>,
-        hostname: String,
     ) -> Self {
         Self {
             socket_addr,
             cancel_token,
             ip_allowlist,
-            hostname,
         }
     }
     /// Cancels the query running on user's compute node.
     pub(crate) async fn try_cancel_query(self) -> Result<(), CancelError> {
         let socket = TcpStream::connect(self.socket_addr).await?;
-
-        let root_store = TLS_ROOTS
-            .get_or_try_init(load_certs)
-            .map_err(|_e| {
-                CancelError::IO(std::io::Error::new(
-                    std::io::ErrorKind::Other,
-                    "TLS root store initialization failed".to_string(),
-                ))
-            })?
-            .clone();
-
-        let client_config =
-            rustls::ClientConfig::builder_with_provider(Arc::new(ring::default_provider()))
-                .with_safe_default_protocol_versions()
-                .expect("ring should support the default protocol versions")
-                .with_root_certificates(root_store)
-                .with_no_client_auth();
-
-        let mut mk_tls = crate::postgres_rustls::MakeRustlsConnect::new(client_config);
-        let tls = <MakeRustlsConnect as MakeTlsConnect<tokio::net::TcpStream>>::make_tls_connect(
-            &mut mk_tls,
-            &self.hostname,
-        )
-        .map_err(|e| {
-            CancelError::IO(std::io::Error::new(
-                std::io::ErrorKind::Other,
-                e.to_string(),
-            ))
-        })?;
-
-        self.cancel_token.cancel_query_raw(socket, tls).await?;
+        self.cancel_token.cancel_query_raw(socket, NoTls).await?;
         debug!("query was cancelled");
         Ok(())
     }

proxy/src/compute.rs

@@ -10,6 +10,7 @@ use postgres_client::tls::MakeTlsConnect;
 use postgres_client::{CancelToken, RawConnection};
 use postgres_protocol::message::backend::NoticeResponseBody;
 use pq_proto::StartupMessageParams;
+use rustls::client::danger::ServerCertVerifier;
 use rustls::crypto::ring;
 use rustls::pki_types::InvalidDnsNameError;
 use thiserror::Error;
@@ -250,6 +251,7 @@ impl ConnCfg {
     pub(crate) async fn connect(
         &self,
         ctx: &RequestContext,
+        allow_self_signed_compute: bool,
         aux: MetricsAuxInfo,
         timeout: Duration,
     ) -> Result<PostgresConnection, ConnectionError> {
@@ -257,17 +259,25 @@ impl ConnCfg {
         let (socket_addr, stream, host) = self.connect_raw(timeout).await?;
         drop(pause);
 
-        let root_store = TLS_ROOTS
-            .get_or_try_init(load_certs)
-            .map_err(ConnectionError::TlsCertificateError)?
-            .clone();
-
-        let client_config =
+        let client_config = if allow_self_signed_compute {
+            // Allow all certificates for creating the connection
+            let verifier = Arc::new(AcceptEverythingVerifier);
+            rustls::ClientConfig::builder_with_provider(Arc::new(ring::default_provider()))
+                .with_safe_default_protocol_versions()
+                .expect("ring should support the default protocol versions")
+                .dangerous()
+                .with_custom_certificate_verifier(verifier)
+        } else {
+            let root_store = TLS_ROOTS
+                .get_or_try_init(load_certs)
+                .map_err(ConnectionError::TlsCertificateError)?
+                .clone();
             rustls::ClientConfig::builder_with_provider(Arc::new(ring::default_provider()))
                 .with_safe_default_protocol_versions()
                 .expect("ring should support the default protocol versions")
                 .with_root_certificates(root_store)
-                .with_no_client_auth();
+        };
+        let client_config = client_config.with_no_client_auth();
 
         let mut mk_tls = crate::postgres_rustls::MakeRustlsConnect::new(client_config);
         let tls = <MakeRustlsConnect as MakeTlsConnect<tokio::net::TcpStream>>::make_tls_connect(
@@ -309,7 +319,6 @@ impl ConnCfg {
                 secret_key,
             },
             vec![],
-            host.to_string(),
         );
 
         let connection = PostgresConnection {
@@ -341,7 +350,7 @@ fn filtered_options(options: &str) -> Option<String> {
     Some(options)
 }
 
-pub(crate) fn load_certs() -> Result<Arc<rustls::RootCertStore>, Vec<rustls_native_certs::Error>> {
+fn load_certs() -> Result<Arc<rustls::RootCertStore>, Vec<rustls_native_certs::Error>> {
     let der_certs = rustls_native_certs::load_native_certs();
 
     if !der_certs.errors.is_empty() {
@@ -354,6 +363,50 @@ pub(crate) fn load_certs() -> Result<Arc<rustls::RootCertStore>, Vec<rustls_nati
 }
 static TLS_ROOTS: OnceCell<Arc<rustls::RootCertStore>> = OnceCell::new();
 
+#[derive(Debug)]
+struct AcceptEverythingVerifier;
+impl ServerCertVerifier for AcceptEverythingVerifier {
+    fn supported_verify_schemes(&self) -> Vec<rustls::SignatureScheme> {
+        use rustls::SignatureScheme;
+        // The schemes for which `SignatureScheme::supported_in_tls13` returns true.
+        vec![
+            SignatureScheme::ECDSA_NISTP521_SHA512,
+            SignatureScheme::ECDSA_NISTP384_SHA384,
+            SignatureScheme::ECDSA_NISTP256_SHA256,
+            SignatureScheme::RSA_PSS_SHA512,
+            SignatureScheme::RSA_PSS_SHA384,
+            SignatureScheme::RSA_PSS_SHA256,
+            SignatureScheme::ED25519,
+        ]
+    }
+    fn verify_server_cert(
+        &self,
+        _end_entity: &rustls::pki_types::CertificateDer<'_>,
+        _intermediates: &[rustls::pki_types::CertificateDer<'_>],
+        _server_name: &rustls::pki_types::ServerName<'_>,
+        _ocsp_response: &[u8],
+        _now: rustls::pki_types::UnixTime,
+    ) -> Result<rustls::client::danger::ServerCertVerified, rustls::Error> {
+        Ok(rustls::client::danger::ServerCertVerified::assertion())
+    }
+    fn verify_tls12_signature(
+        &self,
+        _message: &[u8],
+        _cert: &rustls::pki_types::CertificateDer<'_>,
+        _dss: &rustls::DigitallySignedStruct,
+    ) -> Result<rustls::client::danger::HandshakeSignatureValid, rustls::Error> {
+        Ok(rustls::client::danger::HandshakeSignatureValid::assertion())
+    }
+    fn verify_tls13_signature(
+        &self,
+        _message: &[u8],
+        _cert: &rustls::pki_types::CertificateDer<'_>,
+        _dss: &rustls::DigitallySignedStruct,
+    ) -> Result<rustls::client::danger::HandshakeSignatureValid, rustls::Error> {
+        Ok(rustls::client::danger::HandshakeSignatureValid::assertion())
+    }
+}
+
 #[cfg(test)]
 mod tests {
     use super::*;

proxy/src/config.rs

@@ -25,6 +25,7 @@ use crate::types::Host;
 pub struct ProxyConfig {
     pub tls_config: Option<TlsConfig>,
     pub metric_collection: Option<MetricCollectionConfig>,
+    pub allow_self_signed_compute: bool,
     pub http_config: HttpConfig,
     pub authentication_config: AuthenticationConfig,
     pub proxy_protocol_v2: ProxyProtocolV2,

proxy/src/console_redirect_proxy.rs

@@ -213,6 +213,7 @@ pub(crate) async fn handle_client<S: AsyncRead + AsyncWrite + Unpin>(
             params_compat: true,
             params: &params,
             locks: &config.connect_compute_locks,
+            allow_self_signed_compute: config.allow_self_signed_compute,
         },
         &user_info,
         config.wake_compute_retry_config,

proxy/src/control_plane/client/mock.rs

@@ -74,44 +74,39 @@ impl MockControlPlane {
 
             tokio::spawn(connection);
 
-            let secret = if let Some(entry) = get_execute_postgres_query(
-                &client,
-                "select rolpassword from pg_catalog.pg_authid where rolname = $1",
-                &[&&*user_info.user],
-                "rolpassword",
-            )
-            .await?
-            {
-                info!("got a secret: {entry}"); // safe since it's not a prod scenario
-                let secret = scram::ServerSecret::parse(&entry).map(AuthSecret::Scram);
-                secret.or_else(|| parse_md5(&entry).map(AuthSecret::Md5))
-            } else {
-                warn!("user '{}' does not exist", user_info.user);
-                None
+            let secret = {
+                let query = "select rolpassword from pg_catalog.pg_authid where rolname = $1";
+                if let Some(row) = client.query_opt(query, &[&&*user_info.user]).await? {
+                    let entry: String = row.get("rolpassword");
+
+                    info!("got a secret: {entry}"); // safe since it's not a prod scenario
+                    let secret = scram::ServerSecret::parse(&entry).map(AuthSecret::Scram);
+                    secret.or_else(|| parse_md5(&entry).map(AuthSecret::Md5))
+                } else {
+                    warn!("user '{}' does not exist", user_info.user);
+                    None
+                }
             };
 
-            let allowed_ips = if self.ip_allowlist_check_enabled {
-                match get_execute_postgres_query(
-                    &client,
-                    "select allowed_ips from neon_control_plane.endpoints where endpoint_id = $1",
-                    &[&user_info.endpoint.as_str()],
-                    "allowed_ips",
-                )
-                .await?
-                {
-                    Some(s) => {
-                        info!("got allowed_ips: {s}");
-                        s.split(',')
-                            .map(|s| {
-                                IpPattern::from_str(s).expect("mocked ip pattern should be correct")
-                            })
-                            .collect()
-                    }
-                    None => vec![],
+            let mut allowed_ips = vec![];
+            if self.ip_allowlist_check_enabled {
+                let query =
+                    "select allowed_ips from neon_control_plane.endpoints where endpoint_id = $1";
+                let row = client
+                    .query_one(query, &[&user_info.endpoint.as_str()])
+                    .await?;
+
+                let s: Option<String> = row.get("allowed_ips");
+                if let Some(s) = s {
+                    info!("got allowed_ips: {s}");
+                    allowed_ips = s
+                        .split(',')
+                        .map(|s| {
+                            IpPattern::from_str(s).expect("mocked ip pattern should be correct")
+                        })
+                        .collect();
                 }
-            } else {
-                vec![]
-            };
+            }
 
             Ok((secret, allowed_ips))
         }
@@ -134,11 +129,8 @@ impl MockControlPlane {
 
         let connection = tokio::spawn(connection);
 
-        let res = client.query(
-                "select id, jwks_url, audience, role_names from neon_control_plane.endpoint_jwks where endpoint_id = $1",
-                &[&endpoint.as_str()],
-            )
-            .await?;
+        let query = "select id, jwks_url, audience, role_names from neon_control_plane.endpoint_jwks where endpoint_id = $1";
+        let res = client.query(query, &[&endpoint.as_str()]).await?;
 
         let mut rows = vec![];
         for row in res {
@@ -161,11 +153,23 @@ impl MockControlPlane {
         Ok(rows)
     }
 
-    async fn do_wake_compute(&self) -> Result<NodeInfo, WakeComputeError> {
-        let mut config = compute::ConnCfg::new(
-            self.endpoint.host_str().unwrap_or("localhost").to_owned(),
-            self.endpoint.port().unwrap_or(5432),
-        );
+    async fn do_wake_compute(
+        &self,
+        user_info: &ComputeUserInfo,
+    ) -> Result<NodeInfo, WakeComputeError> {
+        let (client, connection) =
+            tokio_postgres::connect(self.endpoint.as_str(), tokio_postgres::NoTls).await?;
+        tokio::spawn(connection);
+
+        let query = "select host, port from neon_control_plane.endpoints where endpoint_id = $1";
+        let row = client
+            .query_one(query, &[&user_info.endpoint.as_str()])
+            .await?;
+
+        let host: String = row.get("host");
+        let port: i32 = row.get("port");
+
+        let mut config = compute::ConnCfg::new(host, port as u16);
         config.ssl_mode(postgres_client::config::SslMode::Disable);
 
         let node = NodeInfo {
@@ -182,26 +186,6 @@ impl MockControlPlane {
     }
 }
 
-async fn get_execute_postgres_query(
-    client: &Client,
-    query: &str,
-    params: &[&(dyn tokio_postgres::types::ToSql + Sync)],
-    idx: &str,
-) -> Result<Option<String>, GetAuthInfoError> {
-    let rows = client.query(query, params).await?;
-
-    // We can get at most one row, because `rolname` is unique.
-    let Some(row) = rows.first() else {
-        // This means that the user doesn't exist, so there can be no secret.
-        // However, this is still a *valid* outcome which is very similar
-        // to getting `404 Not found` from the Neon console.
-        return Ok(None);
-    };
-
-    let entry = row.try_get(idx).map_err(MockApiError::PasswordNotSet)?;
-    Ok(Some(entry))
-}
-
 impl super::ControlPlaneApi for MockControlPlane {
     #[tracing::instrument(skip_all)]
     async fn get_role_secret(
@@ -239,9 +223,11 @@ impl super::ControlPlaneApi for MockControlPlane {
     async fn wake_compute(
         &self,
         _ctx: &RequestContext,
-        _user_info: &ComputeUserInfo,
+        user_info: &ComputeUserInfo,
     ) -> Result<CachedNodeInfo, WakeComputeError> {
-        self.do_wake_compute().map_ok(Cached::new_uncached).await
+        self.do_wake_compute(user_info)
+            .map_ok(Cached::new_uncached)
+            .await
     }
 }
 

proxy/src/control_plane/mod.rs

@@ -73,9 +73,12 @@ impl NodeInfo {
     pub(crate) async fn connect(
         &self,
         ctx: &RequestContext,
+        allow_self_signed_compute: bool,
         timeout: Duration,
     ) -> Result<compute::PostgresConnection, compute::ConnectionError> {
-        self.config.connect(ctx, self.aux.clone(), timeout).await
+        self.config
+            .connect(ctx, allow_self_signed_compute, self.aux.clone(), timeout)
+            .await
     }
 
     pub(crate) fn reuse_settings(&mut self, other: Self) {

proxy/src/proxy/connect_compute.rs

@@ -73,6 +73,9 @@ pub(crate) struct TcpMechanism<'a> {
 
     /// connect_to_compute concurrency lock
     pub(crate) locks: &'static ApiLocks<Host>,
+
+    /// Whether we should accept self-signed certificates (for testing)
+    pub(crate) allow_self_signed_compute: bool,
 }
 
 #[async_trait]
@@ -90,7 +93,11 @@ impl ConnectMechanism for TcpMechanism<'_> {
     ) -> Result<PostgresConnection, Self::Error> {
         let host = node_info.config.get_host();
         let permit = self.locks.get_permit(&host).await?;
-        permit.release_result(node_info.connect(ctx, timeout).await)
+        permit.release_result(
+            node_info
+                .connect(ctx, self.allow_self_signed_compute, timeout)
+                .await,
+        )
     }
 
     fn update_connect_config(&self, config: &mut compute::ConnCfg) {

proxy/src/proxy/mod.rs

@@ -348,6 +348,8 @@ pub(crate) async fn handle_client<S: AsyncRead + AsyncWrite + Unpin>(
             params_compat,
             params: &params,
             locks: &config.connect_compute_locks,
+            // only used for console redirect testing.
+            allow_self_signed_compute: false,
         },
         &user_info,
         config.wake_compute_retry_config,

proxy/src/sasl/stream.rs

@@ -50,12 +50,6 @@ impl<S: AsyncWrite + Unpin> SaslStream<'_, S> {
         self.stream.write_message(&msg.to_reply()).await?;
         Ok(())
     }
-
-    // Queue a SASL message for the client.
-    fn send_noflush(&mut self, msg: &ServerMessage<&str>) -> io::Result<()> {
-        self.stream.write_message_noflush(&msg.to_reply())?;
-        Ok(())
-    }
 }
 
 /// SASL authentication outcome.
@@ -91,7 +85,7 @@ impl<S: AsyncRead + AsyncWrite + Unpin> SaslStream<'_, S> {
                     continue;
                 }
                 Step::Success(result, reply) => {
-                    self.send_noflush(&ServerMessage::Final(&reply))?;
+                    self.send(&ServerMessage::Final(&reply)).await?;
                     Outcome::Success(result)
                 }
                 Step::Failure(reason) => Outcome::Failure(reason),

safekeeper/Cargo.toml

@@ -9,7 +9,6 @@ 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"]
-benchmarking = []
 
 [dependencies]
 async-stream.workspace = true
@@ -78,4 +77,3 @@ tracing-subscriber = { workspace = true, features = ["json"] }
 [[bench]]
 name = "receive_wal"
 harness = false
-required-features = ["benchmarking"]

safekeeper/benches/benchutils.rs

@@ -1,18 +1,18 @@
 use std::sync::Arc;
 
-use crate::rate_limit::RateLimiter;
-use crate::safekeeper::{ProposerAcceptorMessage, ProposerElected, SafeKeeper, TermHistory};
-use crate::state::{TimelinePersistentState, TimelineState};
-use crate::timeline::{get_timeline_dir, SharedState, StateSK, Timeline};
-use crate::timelines_set::TimelinesSet;
-use crate::wal_backup::remote_timeline_path;
-use crate::{control_file, wal_storage, SafeKeeperConf};
 use camino_tempfile::Utf8TempDir;
+use safekeeper::rate_limit::RateLimiter;
+use safekeeper::safekeeper::{ProposerAcceptorMessage, ProposerElected, SafeKeeper, TermHistory};
+use safekeeper::state::{TimelinePersistentState, TimelineState};
+use safekeeper::timeline::{get_timeline_dir, SharedState, StateSK, Timeline};
+use safekeeper::timelines_set::TimelinesSet;
+use safekeeper::wal_backup::remote_timeline_path;
+use safekeeper::{control_file, wal_storage, SafeKeeperConf};
 use tokio::fs::create_dir_all;
 use utils::id::{NodeId, TenantTimelineId};
 use utils::lsn::Lsn;
 
-/// A Safekeeper testing or benchmarking environment. Uses a tempdir for storage, removed on drop.
+/// A Safekeeper benchmarking environment. Uses a tempdir for storage, removed on drop.
 pub struct Env {
     /// Whether to enable fsync.
     pub fsync: bool,
@@ -21,7 +21,7 @@ pub struct Env {
 }
 
 impl Env {
-    /// Creates a new test or benchmarking environment in a temporary directory. fsync controls whether to
+    /// Creates a new benchmarking environment in a temporary directory. fsync controls whether to
     /// enable fsyncing.
     pub fn new(fsync: bool) -> anyhow::Result<Self> {
         let tempdir = camino_tempfile::tempdir()?;
@@ -47,7 +47,6 @@ impl Env {
         &self,
         node_id: NodeId,
         ttid: TenantTimelineId,
-        start_lsn: Lsn,
     ) -> anyhow::Result<SafeKeeper<control_file::FileStorage, wal_storage::PhysicalStorage>> {
         let conf = self.make_conf(node_id);
 
@@ -68,9 +67,9 @@ impl Env {
         safekeeper
             .process_msg(&ProposerAcceptorMessage::Elected(ProposerElected {
                 term: 1,
-                start_streaming_at: start_lsn,
-                term_history: TermHistory(vec![(1, start_lsn).into()]),
-                timeline_start_lsn: start_lsn,
+                start_streaming_at: Lsn(0),
+                term_history: TermHistory(vec![(1, Lsn(0)).into()]),
+                timeline_start_lsn: Lsn(0),
             }))
             .await?;
 
@@ -83,13 +82,12 @@ impl Env {
         &self,
         node_id: NodeId,
         ttid: TenantTimelineId,
-        start_lsn: Lsn,
     ) -> anyhow::Result<Arc<Timeline>> {
         let conf = Arc::new(self.make_conf(node_id));
         let timeline_dir = get_timeline_dir(&conf, &ttid);
         let remote_path = remote_timeline_path(&ttid)?;
 
-        let safekeeper = self.make_safekeeper(node_id, ttid, start_lsn).await?;
+        let safekeeper = self.make_safekeeper(node_id, ttid).await?;
         let shared_state = SharedState::new(StateSK::Loaded(safekeeper));
 
         let timeline = Timeline::new(

safekeeper/benches/receive_wal.rs

@@ -1,7 +1,11 @@
 //! WAL ingestion benchmarks.
 
+#[path = "benchutils.rs"]
+mod benchutils;
+
 use std::io::Write as _;
 
+use benchutils::Env;
 use bytes::BytesMut;
 use camino_tempfile::tempfile;
 use criterion::{criterion_group, criterion_main, BatchSize, Bencher, Criterion};
@@ -12,7 +16,6 @@ use safekeeper::receive_wal::{self, WalAcceptor};
 use safekeeper::safekeeper::{
     AcceptorProposerMessage, AppendRequest, AppendRequestHeader, ProposerAcceptorMessage,
 };
-use safekeeper::test_utils::Env;
 use tokio::io::AsyncWriteExt as _;
 use utils::id::{NodeId, TenantTimelineId};
 use utils::lsn::Lsn;
@@ -73,15 +76,12 @@ fn bench_process_msg(c: &mut Criterion) {
         assert!(size >= prefixlen);
         let message = vec![0; size - prefixlen];
 
-        let walgen = &mut WalGenerator::new(LogicalMessageGenerator::new(prefix, &message), Lsn(0));
+        let walgen = &mut WalGenerator::new(LogicalMessageGenerator::new(prefix, &message));
 
         // Set up the Safekeeper.
         let env = Env::new(fsync)?;
-        let mut safekeeper = runtime.block_on(env.make_safekeeper(
-            NodeId(1),
-            TenantTimelineId::generate(),
-            Lsn(0),
-        ))?;
+        let mut safekeeper =
+            runtime.block_on(env.make_safekeeper(NodeId(1), TenantTimelineId::generate()))?;
 
         b.iter_batched_ref(
             // Pre-construct WAL records and requests. Criterion will batch them.
@@ -134,8 +134,7 @@ fn bench_wal_acceptor(c: &mut Criterion) {
         let runtime = tokio::runtime::Runtime::new()?; // needs multithreaded
 
         let env = Env::new(fsync)?;
-        let walgen =
-            &mut WalGenerator::new(LogicalMessageGenerator::new(c"prefix", b"message"), Lsn(0));
+        let walgen = &mut WalGenerator::new(LogicalMessageGenerator::new(c"prefix", b"message"));
 
         // Create buffered channels that can fit all requests, to avoid blocking on channels.
         let (msg_tx, msg_rx) = tokio::sync::mpsc::channel(n);
@@ -146,7 +145,7 @@ fn bench_wal_acceptor(c: &mut Criterion) {
             // TODO: WalAcceptor doesn't actually need a full timeline, only
             // Safekeeper::process_msg(). Consider decoupling them to simplify the setup.
             let tli = env
-                .make_timeline(NodeId(1), TenantTimelineId::generate(), Lsn(0))
+                .make_timeline(NodeId(1), TenantTimelineId::generate())
                 .await?
                 .wal_residence_guard()
                 .await?;
@@ -240,7 +239,7 @@ fn bench_wal_acceptor_throughput(c: &mut Criterion) {
         assert!(size >= prefixlen);
         let message = vec![0; size - prefixlen];
 
-        let walgen = &mut WalGenerator::new(LogicalMessageGenerator::new(prefix, &message), Lsn(0));
+        let walgen = &mut WalGenerator::new(LogicalMessageGenerator::new(prefix, &message));
 
         // Construct and spawn the WalAcceptor task.
         let env = Env::new(fsync)?;
@@ -250,7 +249,7 @@ fn bench_wal_acceptor_throughput(c: &mut Criterion) {
 
         runtime.block_on(async {
             let tli = env
-                .make_timeline(NodeId(1), TenantTimelineId::generate(), Lsn(0))
+                .make_timeline(NodeId(1), TenantTimelineId::generate())
                 .await?
                 .wal_residence_guard()
                 .await?;

safekeeper/src/http/routes.rs

@@ -564,7 +564,7 @@ pub fn make_router(
     if conf.http_auth.is_some() {
         router = router.middleware(auth_middleware(|request| {
             const ALLOWLIST_ROUTES: &[&str] =
-                &["/v1/status", "/metrics", "/profile/cpu", "/profile/heap"];
+                &["/v1/status", "/metrics", "/profile/cpu", "profile/heap"];
             if ALLOWLIST_ROUTES.contains(&request.uri().path()) {
                 None
             } else {

safekeeper/src/lib.rs

@@ -43,9 +43,6 @@ pub mod wal_reader_stream;
 pub mod wal_service;
 pub mod wal_storage;
 
-#[cfg(any(test, feature = "benchmarking"))]
-pub mod test_utils;
-
 mod timelines_global_map;
 use std::sync::Arc;
 pub use timelines_global_map::GlobalTimelines;

safekeeper/src/send_interpreted_wal.rs

@@ -94,14 +94,9 @@ impl<IO: AsyncRead + AsyncWrite + Unpin> InterpretedWalSender<'_, IO> {
                         }
                     }
 
-                    let max_next_record_lsn = match max_next_record_lsn {
-                        Some(lsn) => lsn,
-                        None => { continue; }
-                    };
-
                     let batch = InterpretedWalRecords {
                         records,
-                        next_record_lsn: Some(max_next_record_lsn),
+                        next_record_lsn: max_next_record_lsn
                     };
 
                     tx.send(Batch {wal_end_lsn, available_wal_end_lsn, records: batch}).await.unwrap();

safekeeper/tests/walproposer_sim/walproposer_disk.rs

@@ -18,7 +18,7 @@ impl DiskWalProposer {
                 internal_available_lsn: Lsn(0),
                 prev_lsn: Lsn(0),
                 disk: BlockStorage::new(),
-                wal_generator: WalGenerator::new(LogicalMessageGenerator::new(c"", &[]), Lsn(0)),
+                wal_generator: WalGenerator::new(LogicalMessageGenerator::new(c"", &[])),
             }),
         })
     }

storage_controller/src/compute_hook.rs

@@ -1,4 +1,3 @@
-use std::borrow::Cow;
 use std::error::Error as _;
 use std::sync::Arc;
 use std::{collections::HashMap, time::Duration};
@@ -7,7 +6,6 @@ use control_plane::endpoint::{ComputeControlPlane, EndpointStatus};
 use control_plane::local_env::LocalEnv;
 use futures::StreamExt;
 use hyper::StatusCode;
-use pageserver_api::controller_api::AvailabilityZone;
 use pageserver_api::shard::{ShardCount, ShardNumber, ShardStripeSize, TenantShardId};
 use postgres_connection::parse_host_port;
 use serde::{Deserialize, Serialize};
@@ -30,9 +28,6 @@ struct UnshardedComputeHookTenant {
     // Which node is this tenant attached to
     node_id: NodeId,
 
-    // The tenant's preferred AZ, so that we may pass this on to the control plane
-    preferred_az: Option<AvailabilityZone>,
-
     // Must hold this lock to send a notification.
     send_lock: Arc<tokio::sync::Mutex<Option<ComputeRemoteState>>>,
 }
@@ -41,9 +36,6 @@ struct ShardedComputeHookTenant {
     shard_count: ShardCount,
     shards: Vec<(ShardNumber, NodeId)>,
 
-    // The tenant's preferred AZ, so that we may pass this on to the control plane
-    preferred_az: Option<AvailabilityZone>,
-
     // Must hold this lock to send a notification.  The contents represent
     // the last successfully sent notification, and are used to coalesce multiple
     // updates by only sending when there is a chance since our last successful send.
@@ -72,24 +64,17 @@ enum ComputeHookTenant {
 
 impl ComputeHookTenant {
     /// Construct with at least one shard's information
-    fn new(
-        tenant_shard_id: TenantShardId,
-        stripe_size: ShardStripeSize,
-        preferred_az: Option<AvailabilityZone>,
-        node_id: NodeId,
-    ) -> Self {
+    fn new(tenant_shard_id: TenantShardId, stripe_size: ShardStripeSize, node_id: NodeId) -> Self {
         if tenant_shard_id.shard_count.count() > 1 {
             Self::Sharded(ShardedComputeHookTenant {
                 shards: vec![(tenant_shard_id.shard_number, node_id)],
                 stripe_size,
                 shard_count: tenant_shard_id.shard_count,
-                preferred_az,
                 send_lock: Arc::default(),
             })
         } else {
             Self::Unsharded(UnshardedComputeHookTenant {
                 node_id,
-                preferred_az,
                 send_lock: Arc::default(),
             })
         }
@@ -135,20 +120,15 @@ impl ComputeHookTenant {
 
     /// Set one shard's location.  If stripe size or shard count have changed, Self is reset
     /// and drops existing content.
-    fn update(&mut self, shard_update: ShardUpdate) {
-        let tenant_shard_id = shard_update.tenant_shard_id;
-        let node_id = shard_update.node_id;
-        let stripe_size = shard_update.stripe_size;
-        let preferred_az = shard_update.preferred_az;
-
+    fn update(
+        &mut self,
+        tenant_shard_id: TenantShardId,
+        stripe_size: ShardStripeSize,
+        node_id: NodeId,
+    ) {
         match self {
             Self::Unsharded(unsharded_tenant) if tenant_shard_id.shard_count.count() == 1 => {
-                unsharded_tenant.node_id = node_id;
-                if unsharded_tenant.preferred_az.as_ref()
-                    != preferred_az.as_ref().map(|az| az.as_ref())
-                {
-                    unsharded_tenant.preferred_az = preferred_az.map(|az| az.as_ref().clone());
-                }
+                unsharded_tenant.node_id = node_id
             }
             Self::Sharded(sharded_tenant)
                 if sharded_tenant.stripe_size == stripe_size
@@ -166,21 +146,10 @@ impl ComputeHookTenant {
                         .push((tenant_shard_id.shard_number, node_id));
                     sharded_tenant.shards.sort_by_key(|s| s.0)
                 }
-
-                if sharded_tenant.preferred_az.as_ref()
-                    != preferred_az.as_ref().map(|az| az.as_ref())
-                {
-                    sharded_tenant.preferred_az = preferred_az.map(|az| az.as_ref().clone());
-                }
             }
             _ => {
                 // Shard count changed: reset struct.
-                *self = Self::new(
-                    tenant_shard_id,
-                    stripe_size,
-                    preferred_az.map(|az| az.into_owned()),
-                    node_id,
-                );
+                *self = Self::new(tenant_shard_id, stripe_size, node_id);
             }
         }
     }
@@ -196,7 +165,6 @@ struct ComputeHookNotifyRequestShard {
 #[derive(Serialize, Deserialize, Debug, Eq, PartialEq)]
 struct ComputeHookNotifyRequest {
     tenant_id: TenantId,
-    preferred_az: Option<String>,
     stripe_size: Option<ShardStripeSize>,
     shards: Vec<ComputeHookNotifyRequestShard>,
 }
@@ -270,10 +238,6 @@ impl ComputeHookTenant {
                     node_id: unsharded_tenant.node_id,
                 }],
                 stripe_size: None,
-                preferred_az: unsharded_tenant
-                    .preferred_az
-                    .as_ref()
-                    .map(|az| az.0.clone()),
             }),
             Self::Sharded(sharded_tenant)
                 if sharded_tenant.shards.len() == sharded_tenant.shard_count.count() as usize =>
@@ -289,7 +253,6 @@ impl ComputeHookTenant {
                         })
                         .collect(),
                     stripe_size: Some(sharded_tenant.stripe_size),
-                    preferred_az: sharded_tenant.preferred_az.as_ref().map(|az| az.0.clone()),
                 })
             }
             Self::Sharded(sharded_tenant) => {
@@ -350,17 +313,6 @@ pub(super) struct ComputeHook {
     client: reqwest::Client,
 }
 
-/// Callers may give us a list of these when asking us to send a bulk batch
-/// of notifications in the background.  This is a 'notification' in the sense of
-/// other code notifying us of a shard's status, rather than being the final notification
-/// that we send upwards to the control plane for the whole tenant.
-pub(crate) struct ShardUpdate<'a> {
-    pub(crate) tenant_shard_id: TenantShardId,
-    pub(crate) node_id: NodeId,
-    pub(crate) stripe_size: ShardStripeSize,
-    pub(crate) preferred_az: Option<Cow<'a, AvailabilityZone>>,
-}
-
 impl ComputeHook {
     pub(super) fn new(config: Config) -> Self {
         let authorization_header = config
@@ -411,7 +363,6 @@ impl ComputeHook {
             tenant_id,
             shards,
             stripe_size,
-            preferred_az: _preferred_az,
         } = reconfigure_request;
 
         let compute_pageservers = shards
@@ -552,30 +503,24 @@ impl ComputeHook {
     }
 
     /// Synchronous phase: update the per-tenant state for the next intended notification
-    fn notify_prepare(&self, shard_update: ShardUpdate) -> MaybeSendResult {
+    fn notify_prepare(
+        &self,
+        tenant_shard_id: TenantShardId,
+        node_id: NodeId,
+        stripe_size: ShardStripeSize,
+    ) -> MaybeSendResult {
         let mut state_locked = self.state.lock().unwrap();
 
         use std::collections::hash_map::Entry;
-        let tenant_shard_id = shard_update.tenant_shard_id;
-
         let tenant = match state_locked.entry(tenant_shard_id.tenant_id) {
-            Entry::Vacant(e) => {
-                let ShardUpdate {
-                    tenant_shard_id,
-                    node_id,
-                    stripe_size,
-                    preferred_az,
-                } = shard_update;
-                e.insert(ComputeHookTenant::new(
-                    tenant_shard_id,
-                    stripe_size,
-                    preferred_az.map(|az| az.into_owned()),
-                    node_id,
-                ))
-            }
+            Entry::Vacant(e) => e.insert(ComputeHookTenant::new(
+                tenant_shard_id,
+                stripe_size,
+                node_id,
+            )),
             Entry::Occupied(e) => {
                 let tenant = e.into_mut();
-                tenant.update(shard_update);
+                tenant.update(tenant_shard_id, stripe_size, node_id);
                 tenant
             }
         };
@@ -663,14 +608,13 @@ impl ComputeHook {
     /// if something failed.
     pub(super) fn notify_background(
         self: &Arc<Self>,
-        notifications: Vec<ShardUpdate>,
+        notifications: Vec<(TenantShardId, NodeId, ShardStripeSize)>,
         result_tx: tokio::sync::mpsc::Sender<Result<(), (TenantShardId, NotifyError)>>,
         cancel: &CancellationToken,
     ) {
         let mut maybe_sends = Vec::new();
-        for shard_update in notifications {
-            let tenant_shard_id = shard_update.tenant_shard_id;
-            let maybe_send_result = self.notify_prepare(shard_update);
+        for (tenant_shard_id, node_id, stripe_size) in notifications {
+            let maybe_send_result = self.notify_prepare(tenant_shard_id, node_id, stripe_size);
             maybe_sends.push((tenant_shard_id, maybe_send_result))
         }
 
@@ -734,14 +678,15 @@ impl ComputeHook {
     /// periods, but we don't retry forever.  The **caller** is responsible for handling failures and
     /// ensuring that they eventually call again to ensure that the compute is eventually notified of
     /// the proper pageserver nodes for a tenant.
-    #[tracing::instrument(skip_all, fields(tenant_id=%shard_update.tenant_shard_id.tenant_id, shard_id=%shard_update.tenant_shard_id.shard_slug(), node_id))]
-    pub(super) async fn notify<'a>(
+    #[tracing::instrument(skip_all, fields(tenant_id=%tenant_shard_id.tenant_id, shard_id=%tenant_shard_id.shard_slug(), node_id))]
+    pub(super) async fn notify(
         &self,
-        shard_update: ShardUpdate<'a>,
+        tenant_shard_id: TenantShardId,
+        node_id: NodeId,
+        stripe_size: ShardStripeSize,
         cancel: &CancellationToken,
     ) -> Result<(), NotifyError> {
-        let tenant_shard_id = shard_update.tenant_shard_id;
-        let maybe_send_result = self.notify_prepare(shard_update);
+        let maybe_send_result = self.notify_prepare(tenant_shard_id, node_id, stripe_size);
         self.notify_execute(maybe_send_result, tenant_shard_id, cancel)
             .await
     }
@@ -794,7 +739,6 @@ pub(crate) mod tests {
                 shard_number: ShardNumber(0),
             },
             ShardStripeSize(12345),
-            None,
             NodeId(1),
         );
 
@@ -821,32 +765,30 @@ pub(crate) mod tests {
         // Writing the first shard of a multi-sharded situation (i.e. in a split)
         // resets the tenant state and puts it in an non-notifying state (need to
         // see all shards)
-        tenant_state.update(ShardUpdate {
-            tenant_shard_id: TenantShardId {
+        tenant_state.update(
+            TenantShardId {
                 tenant_id,
                 shard_count: ShardCount::new(2),
                 shard_number: ShardNumber(1),
             },
-            stripe_size: ShardStripeSize(32768),
-            preferred_az: None,
-            node_id: NodeId(1),
-        });
+            ShardStripeSize(32768),
+            NodeId(1),
+        );
         assert!(matches!(
             tenant_state.maybe_send(tenant_id, None),
             MaybeSendResult::Noop
         ));
 
         // Writing the second shard makes it ready to notify
-        tenant_state.update(ShardUpdate {
-            tenant_shard_id: TenantShardId {
+        tenant_state.update(
+            TenantShardId {
                 tenant_id,
                 shard_count: ShardCount::new(2),
                 shard_number: ShardNumber(0),
             },
-            stripe_size: ShardStripeSize(32768),
-            preferred_az: None,
-            node_id: NodeId(1),
-        });
+            ShardStripeSize(32768),
+            NodeId(1),
+        );
 
         let send_result = tenant_state.maybe_send(tenant_id, None);
         let MaybeSendResult::Transmit((request, mut guard)) = send_result else {

storage_controller/src/persistence.rs

@@ -11,7 +11,6 @@ use diesel::Connection;
 use itertools::Itertools;
 use pageserver_api::controller_api::AvailabilityZone;
 use pageserver_api::controller_api::MetadataHealthRecord;
-use pageserver_api::controller_api::SafekeeperDescribeResponse;
 use pageserver_api::controller_api::ShardSchedulingPolicy;
 use pageserver_api::controller_api::{NodeSchedulingPolicy, PlacementPolicy};
 use pageserver_api::models::TenantConfig;
@@ -1242,18 +1241,6 @@ impl SafekeeperPersistence {
             availability_zone_id: &self.availability_zone_id,
         }
     }
-    pub(crate) fn as_describe_response(&self) -> SafekeeperDescribeResponse {
-        // omit the `active` flag on purpose: it is deprecated.
-        SafekeeperDescribeResponse {
-            id: NodeId(self.id as u64),
-            region_id: self.region_id.clone(),
-            version: self.version,
-            host: self.host.clone(),
-            port: self.port,
-            http_port: self.http_port,
-            availability_zone_id: self.availability_zone_id.clone(),
-        }
-    }
 }
 
 #[derive(Insertable, AsChangeset)]

storage_controller/src/reconciler.rs

@@ -1,14 +1,13 @@
 use crate::pageserver_client::PageserverClient;
 use crate::persistence::Persistence;
-use crate::{compute_hook, service};
-use pageserver_api::controller_api::{AvailabilityZone, PlacementPolicy};
+use crate::service;
+use pageserver_api::controller_api::PlacementPolicy;
 use pageserver_api::models::{
     LocationConfig, LocationConfigMode, LocationConfigSecondary, TenantConfig,
 };
 use pageserver_api::shard::{ShardIdentity, TenantShardId};
 use pageserver_client::mgmt_api;
 use reqwest::StatusCode;
-use std::borrow::Cow;
 use std::collections::HashMap;
 use std::sync::Arc;
 use std::time::{Duration, Instant};
@@ -46,7 +45,6 @@ pub(super) struct Reconciler {
     pub(crate) reconciler_config: ReconcilerConfig,
 
     pub(crate) config: TenantConfig,
-    pub(crate) preferred_az: Option<AvailabilityZone>,
 
     /// Observed state from the point of view of the reconciler.
     /// This gets updated as the reconciliation makes progress.
@@ -836,12 +834,9 @@ impl Reconciler {
             let result = self
                 .compute_hook
                 .notify(
-                    compute_hook::ShardUpdate {
-                        tenant_shard_id: self.tenant_shard_id,
-                        node_id: node.get_id(),
-                        stripe_size: self.shard.stripe_size,
-                        preferred_az: self.preferred_az.as_ref().map(Cow::Borrowed),
-                    },
+                    self.tenant_shard_id,
+                    node.get_id(),
+                    self.shard.stripe_size,
                     &self.cancel,
                 )
                 .await;

storage_controller/src/service.rs

@@ -18,7 +18,7 @@ use crate::{
     background_node_operations::{
         Drain, Fill, Operation, OperationError, OperationHandler, MAX_RECONCILES_PER_OPERATION,
     },
-    compute_hook::{self, NotifyError},
+    compute_hook::NotifyError,
     drain_utils::{self, TenantShardDrain, TenantShardIterator},
     id_lock_map::{trace_exclusive_lock, trace_shared_lock, IdLockMap, TracingExclusiveGuard},
     leadership::Leadership,
@@ -46,11 +46,10 @@ use pageserver_api::{
     controller_api::{
         AvailabilityZone, MetadataHealthRecord, MetadataHealthUpdateRequest, NodeAvailability,
         NodeRegisterRequest, NodeSchedulingPolicy, NodeShard, NodeShardResponse, PlacementPolicy,
-        SafekeeperDescribeResponse, ShardSchedulingPolicy, ShardsPreferredAzsRequest,
-        ShardsPreferredAzsResponse, TenantCreateRequest, TenantCreateResponse,
-        TenantCreateResponseShard, TenantDescribeResponse, TenantDescribeResponseShard,
-        TenantLocateResponse, TenantPolicyRequest, TenantShardMigrateRequest,
-        TenantShardMigrateResponse,
+        ShardSchedulingPolicy, ShardsPreferredAzsRequest, ShardsPreferredAzsResponse,
+        TenantCreateRequest, TenantCreateResponse, TenantCreateResponseShard,
+        TenantDescribeResponse, TenantDescribeResponseShard, TenantLocateResponse,
+        TenantPolicyRequest, TenantShardMigrateRequest, TenantShardMigrateResponse,
     },
     models::{
         SecondaryProgress, TenantConfigPatchRequest, TenantConfigRequest,
@@ -657,14 +656,11 @@ impl Service {
                     // emit a compute notification for this. In the case where our observed state does not
                     // yet match our intent, we will eventually reconcile, and that will emit a compute notification.
                     if let Some(attached_at) = tenant_shard.stably_attached() {
-                        compute_notifications.push(compute_hook::ShardUpdate {
-                            tenant_shard_id: *tenant_shard_id,
-                            node_id: attached_at,
-                            stripe_size: tenant_shard.shard.stripe_size,
-                            preferred_az: tenant_shard
-                                .preferred_az()
-                                .map(|az| Cow::Owned(az.clone())),
-                        });
+                        compute_notifications.push((
+                            *tenant_shard_id,
+                            attached_at,
+                            tenant_shard.shard.stripe_size,
+                        ));
                     }
                 }
             }
@@ -4790,15 +4786,7 @@ impl Service {
         for (child_id, child_ps, stripe_size) in child_locations {
             if let Err(e) = self
                 .compute_hook
-                .notify(
-                    compute_hook::ShardUpdate {
-                        tenant_shard_id: child_id,
-                        node_id: child_ps,
-                        stripe_size,
-                        preferred_az: preferred_az_id.as_ref().map(Cow::Borrowed),
-                    },
-                    &self.cancel,
-                )
+                .notify(child_id, child_ps, stripe_size, &self.cancel)
                 .await
             {
                 tracing::warn!("Failed to update compute of {}->{} during split, proceeding anyway to complete split ({e})",
@@ -7170,24 +7158,15 @@ impl Service {
 
     pub(crate) async fn safekeepers_list(
         &self,
-    ) -> Result<Vec<SafekeeperDescribeResponse>, DatabaseError> {
-        Ok(self
-            .persistence
-            .list_safekeepers()
-            .await?
-            .into_iter()
-            .map(|v| v.as_describe_response())
-            .collect::<Vec<_>>())
+    ) -> Result<Vec<crate::persistence::SafekeeperPersistence>, DatabaseError> {
+        self.persistence.list_safekeepers().await
     }
 
     pub(crate) async fn get_safekeeper(
         &self,
         id: i64,
-    ) -> Result<SafekeeperDescribeResponse, DatabaseError> {
-        self.persistence
-            .safekeeper_get(id)
-            .await
-            .map(|v| v.as_describe_response())
+    ) -> Result<crate::persistence::SafekeeperPersistence, DatabaseError> {
+        self.persistence.safekeeper_get(id).await
     }
 
     pub(crate) async fn upsert_safekeeper(

storage_controller/src/tenant_shard.rs

@@ -1198,7 +1198,6 @@ impl TenantShard {
             detach,
             reconciler_config,
             config: self.config.clone(),
-            preferred_az: self.preferred_az_id.clone(),
             observed: self.observed.clone(),
             original_observed: self.observed.clone(),
             compute_hook: compute_hook.clone(),

storage_scrubber/src/checks.rs

@@ -310,7 +310,7 @@ pub(crate) enum BlobDataParseResult {
         index_part_generation: Generation,
         s3_layers: HashSet<(LayerName, Generation)>,
     },
-    /// The remains of an uncleanly deleted Timeline or aborted timeline creation(e.g. an initdb archive only, or some layer without an index)
+    /// The remains of a deleted Timeline (i.e. an initdb archive only)
     Relic,
     Incorrect {
         errors: Vec<String>,
@@ -346,7 +346,7 @@ pub(crate) async fn list_timeline_blobs(
     match res {
         ListTimelineBlobsResult::Ready(data) => Ok(data),
         ListTimelineBlobsResult::MissingIndexPart(_) => {
-            // Retry if listing raced with removal of an index
+            // Retry if index is missing.
             let data = list_timeline_blobs_impl(remote_client, id, root_target)
                 .await?
                 .into_data();
@@ -358,7 +358,7 @@ pub(crate) async fn list_timeline_blobs(
 enum ListTimelineBlobsResult {
     /// Blob data is ready to be intepreted.
     Ready(RemoteTimelineBlobData),
-    /// The listing contained an index but when we tried to fetch it, we couldn't
+    /// List timeline blobs has layer files but is missing [`IndexPart`].
     MissingIndexPart(RemoteTimelineBlobData),
 }
 
@@ -467,19 +467,19 @@ async fn list_timeline_blobs_impl(
     match index_part_object.as_ref() {
         Some(selected) => index_part_keys.retain(|k| k != selected),
         None => {
-            // This case does not indicate corruption, but it should be very unusual.  It can
-            // happen if:
-            // - timeline creation is in progress (first layer is written before index is written)
-            // - timeline deletion happened while a stale pageserver was still attached, it might upload
-            //   a layer after the deletion is done.
-            tracing::info!(
+            // It is possible that the branch gets deleted after we got some layer files listed
+            // and we no longer have the index file in the listing.
+            errors.push(
                 "S3 list response got no index_part.json file but still has layer files"
+                    .to_string(),
             );
-            return Ok(ListTimelineBlobsResult::Ready(RemoteTimelineBlobData {
-                blob_data: BlobDataParseResult::Relic,
-                unused_index_keys: index_part_keys,
-                unknown_keys,
-            }));
+            return Ok(ListTimelineBlobsResult::MissingIndexPart(
+                RemoteTimelineBlobData {
+                    blob_data: BlobDataParseResult::Incorrect { errors, s3_layers },
+                    unused_index_keys: index_part_keys,
+                    unknown_keys,
+                },
+            ));
         }
     }
 

test_runner/fixtures/neon_fixtures.py

@@ -134,9 +134,6 @@ DEFAULT_BRANCH_NAME: str = "main"
 
 BASE_PORT: int = 15000
 
-# By default we create pageservers with this phony AZ
-DEFAULT_AZ_ID: str = "us-east-2a"
-
 
 @pytest.fixture(scope="session")
 def neon_api_key() -> str:
@@ -1096,7 +1093,7 @@ class NeonEnv:
                 "pg_auth_type": pg_auth_type,
                 "http_auth_type": http_auth_type,
                 # Default which can be overriden with `NeonEnvBuilder.pageserver_config_override`
-                "availability_zone": DEFAULT_AZ_ID,
+                "availability_zone": "us-east-2a",
                 # Disable pageserver disk syncs in tests: when running tests concurrently, this avoids
                 # the pageserver taking a long time to start up due to syncfs flushing other tests' data
                 "no_sync": True,
@@ -3222,6 +3219,7 @@ class NeonProxy(PgProtocol):
                 # Link auth backend params
                 *["--auth-backend", "link"],
                 *["--uri", NeonProxy.link_auth_uri],
+                *["--allow-self-signed-compute", "true"],
             ]
 
     class ProxyV1(AuthBackend):
@@ -3276,7 +3274,7 @@ class NeonProxy(PgProtocol):
         metric_collection_interval: str | None = None,
     ):
         host = "127.0.0.1"
-        domain = "proxy.localtest.me"  # resolves to 127.0.0.1
+        domain = "ep-test-endpoint.localtest.me"  # resolves to 127.0.0.1
         super().__init__(dsn=auth_backend.default_conn_url, host=domain, port=proxy_port)
 
         self.domain = domain
@@ -3641,7 +3639,19 @@ def static_proxy(
     vanilla_pg.safe_psql("create user proxy with login superuser password 'password'")
     vanilla_pg.safe_psql("CREATE SCHEMA IF NOT EXISTS neon_control_plane")
     vanilla_pg.safe_psql(
-        "CREATE TABLE neon_control_plane.endpoints (endpoint_id VARCHAR(255) PRIMARY KEY, allowed_ips VARCHAR(255))"
+        f"""
+        CREATE TABLE neon_control_plane.endpoints (
+            endpoint_id text PRIMARY KEY,
+            allowed_ips text,
+            host text not null default '{host}',
+            port integer not null default {port}
+        )
+        """
+    )
+    vanilla_pg.safe_psql(
+        """
+        insert into neon_control_plane.endpoints (endpoint_id) VALUES ('ep-test-endpoint');
+        """
     )
 
     proxy_port = port_distributor.get_port()

test_runner/fixtures/pageserver/http.py

@@ -738,18 +738,6 @@ class PageserverHttpClient(requests.Session, MetricsGetter):
         res_json = res.json()
         assert res_json is None
 
-    def timeline_compact_info(
-        self,
-        tenant_id: TenantId | TenantShardId,
-        timeline_id: TimelineId,
-    ) -> Any:
-        res = self.get(
-            f"http://localhost:{self.port}/v1/tenant/{tenant_id}/timeline/{timeline_id}/compact",
-        )
-        self.verbose_error(res)
-        res_json = res.json()
-        return res_json
-
     def timeline_compact(
         self,
         tenant_id: TenantId | TenantShardId,
@@ -761,6 +749,7 @@ class PageserverHttpClient(requests.Session, MetricsGetter):
         enhanced_gc_bottom_most_compaction=False,
         body: dict[str, Any] | None = None,
     ):
+        self.is_testing_enabled_or_skip()
         query = {}
         if force_repartition:
             query["force_repartition"] = "true"

test_runner/performance/many_relations/create_many_relations.sql

@@ -1,199 +0,0 @@
--- create a schema that simulates Neon control plane operations table
--- however use partitioned operations tables with many (e.g. 500) child partition tables per table
--- in summary we create multiple of these partitioned operations tables (with 500 childs each) - until we reach the requested number of tables
-
-
--- first we need some other tables that can be referenced by the operations table
-
---  Table for branches
-CREATE TABLE public.branches (
-    id text PRIMARY KEY
-);
-
--- Table for endpoints
-CREATE TABLE public.endpoints (
-    id text PRIMARY KEY
-);
-
--- Table for projects
-CREATE TABLE public.projects (
-    id text PRIMARY KEY
-);
-
-INSERT INTO public.branches (id)
-VALUES ('branch_1');
-
--- Insert one row into endpoints
-INSERT INTO public.endpoints (id)
-VALUES ('endpoint_1');
-
--- Insert one row into projects
-INSERT INTO public.projects (id)
-VALUES ('project_1');
-
--- now we create a procedure that can create n operations tables
--- we do that in a procedure to save roundtrip latency when scaling the test to many tables
--- prefix is the base table name, e.g. 'operations_scale_1000' if we create 1000 tables
-CREATE OR REPLACE PROCEDURE create_partitioned_tables(prefix text, n INT)
-LANGUAGE plpgsql AS $$
-DECLARE
-    table_name TEXT;  -- Variable to hold table names dynamically
-    i INT;            -- Counter for the loop
-BEGIN
-    -- Loop to create n partitioned tables
-    FOR i IN 1..n LOOP
-        table_name := format('%s_%s', prefix, i);
-
-        -- Create the partitioned table
-        EXECUTE format(
-            'CREATE TABLE public.%s (
-                project_id character varying NOT NULL,
-                id uuid NOT NULL,
-                status integer,
-                action character varying NOT NULL,
-                error character varying,
-                created_at timestamp with time zone NOT NULL DEFAULT now(),
-                updated_at timestamp with time zone NOT NULL DEFAULT now(),
-                spec jsonb,
-                retry_at timestamp with time zone,
-                failures_count integer DEFAULT 0,
-                metadata jsonb NOT NULL DEFAULT ''{}''::jsonb,
-                executor_id text NOT NULL,
-                attempt_duration_ms integer,
-                metrics jsonb DEFAULT ''{}''::jsonb,
-                branch_id text,
-                endpoint_id text,
-                next_operation_id uuid,
-                compute_id text,
-                connection_attempt_at timestamp with time zone,
-                concurrency_key text,
-                queue_id text,
-                CONSTRAINT %s_pkey PRIMARY KEY (id, created_at),
-                CONSTRAINT %s_branch_id_fk FOREIGN KEY (branch_id) REFERENCES branches(id) ON DELETE CASCADE,
-                CONSTRAINT %s_endpoint_id_fk FOREIGN KEY (endpoint_id) REFERENCES endpoints(id) ON DELETE CASCADE,
-                CONSTRAINT %s_next_operation_id_fk FOREIGN KEY (next_operation_id, created_at) REFERENCES %s(id, created_at),
-                CONSTRAINT %s_project_id_fk FOREIGN KEY (project_id) REFERENCES projects(id) ON DELETE CASCADE
-            ) PARTITION BY RANGE (created_at)',
-            table_name, table_name, table_name, table_name, table_name, table_name, table_name
-        );
-
-        -- Add indexes for the partitioned table
-        EXECUTE format('CREATE INDEX index_%s_on_next_operation_id ON public.%s (next_operation_id)', table_name, table_name);
-        EXECUTE format('CREATE INDEX index_%s_on_project_id ON public.%s (project_id)', table_name, table_name);
-        EXECUTE format('CREATE INDEX %s_branch_id ON public.%s (branch_id)', table_name, table_name);
-        EXECUTE format('CREATE INDEX %s_branch_id_created_idx ON public.%s (branch_id, created_at)', table_name, table_name);
-        EXECUTE format('CREATE INDEX %s_created_at_idx ON public.%s (created_at)', table_name, table_name);
-        EXECUTE format('CREATE INDEX %s_created_at_project_id_id_cond_idx ON public.%s (created_at, project_id, id)', table_name, table_name);
-        EXECUTE format('CREATE INDEX %s_endpoint_id ON public.%s (endpoint_id)', table_name, table_name);
-        EXECUTE format(
-            'CREATE INDEX %s_for_redo_worker_idx ON public.%s (executor_id) WHERE status <> 1',
-            table_name, table_name
-        );
-        EXECUTE format(
-            'CREATE INDEX %s_project_id_status_index ON public.%s ((project_id::text), status)',
-            table_name, table_name
-        );
-        EXECUTE format(
-            'CREATE INDEX %s_status_not_finished ON public.%s (status) WHERE status <> 1',
-            table_name, table_name
-        );
-        EXECUTE format('CREATE INDEX %s_updated_at_desc_idx ON public.%s (updated_at DESC)', table_name, table_name);
-        EXECUTE format(
-            'CREATE INDEX %s_with_failures ON public.%s (failures_count) WHERE failures_count > 0',
-            table_name, table_name
-        );
-    END LOOP;
-END;
-$$;
-
--- next we create a procedure that can add the child partitions (one per day) to each of the operations tables
-CREATE OR REPLACE PROCEDURE create_operations_partitions(
-    table_name TEXT, 
-    start_date DATE,
-    end_date DATE
-)
-LANGUAGE plpgsql AS $$
-DECLARE
-    partition_date DATE;
-    partition_name TEXT;
-    counter INT := 0;  -- Counter to track the number of tables created in the current transaction
-BEGIN
-    partition_date := start_date;
-
-    -- Create partitions in batches
-    WHILE partition_date < end_date LOOP
-        partition_name := format('%s_%s', table_name, to_char(partition_date,'YYYY_MM_DD'));
-
-        EXECUTE format(
-            'CREATE TABLE IF NOT EXISTS public.%s PARTITION OF public.%s
-             FOR VALUES FROM (''%s'') TO (''%s'')',
-            partition_name,
-            table_name,
-            partition_date,
-            partition_date + INTERVAL '1 day'
-        );
-
-        counter := counter + 1;
-
-        -- Commit and reset counter after every 100 partitions
-        IF counter >= 100 THEN
-            COMMIT;
-            counter := 0;  -- Reset the counter
-        END IF;
-
-        -- Advance to the next day
-        partition_date := partition_date + INTERVAL '1 day';
-    END LOOP;
-
-    -- Final commit for remaining partitions
-    IF counter > 0 THEN
-        COMMIT;
-    END IF;
-
-    -- Insert synthetic rows into each partition
-    EXECUTE format(
-        'INSERT INTO %I (
-            project_id,
-            branch_id,
-            endpoint_id,
-            id,
-            status,
-            action,
-            created_at,
-            updated_at,
-            spec,
-            metadata,
-            executor_id,
-            failures_count
-        )
-        SELECT 
-            ''project_1'',                                   -- project_id
-            ''branch_1'',                                    -- branch_id
-            ''endpoint_1'',                                  -- endpoint_id
-            ''e8bba687-0df9-4291-bfcd-7d5f6aa7c158'',          -- unique id
-            1,                                               -- status
-            ''SYNTHETIC_ACTION'',                            -- action
-            gs::timestamp + interval ''0 ms'',               -- created_at
-            gs::timestamp + interval ''1 minute'',           -- updated_at
-            ''{"key": "value"}'',                            -- spec (JSONB)
-            ''{"metadata_key": "metadata_value"}'',          -- metadata (JSONB)
-            ''executor_1'',                                  -- executor_id
-            0                                                -- failures_count
-        FROM generate_series(%L, %L::DATE - INTERVAL ''1 day'', INTERVAL ''1 day'') AS gs',
-        table_name, start_date, end_date
-    );
-    
-    -- Commit the inserted rows
-    COMMIT;
-END;
-$$;
-
--- we can now create partitioned tables using something like
--- CALL create_partitioned_tables('operations_scale_1000' ,10);
-
--- and we can create the child partitions for a table using something like
--- CALL create_operations_partitions(
---     'operations_scale_1000_1',
---     '2000-01-01',            -- Start date
---     ('2000-01-01'::DATE + INTERVAL '1 day' * 500)::DATE  -- End date (start date + number of days)
--- );

test_runner/performance/test_gc_feedback.py

@@ -22,7 +22,7 @@ def gc_feedback_impl(neon_env_builder: NeonEnvBuilder, zenbenchmark: NeonBenchma
             "checkpoint_distance": f"{1024 ** 2}",
             "compaction_target_size": f"{1024 ** 2}",
             # set PITR interval to be small, so we can do GC
-            "pitr_interval": "10 s",
+            "pitr_interval": "60 s",
             # "compaction_threshold": "3",
             # "image_creation_threshold": "2",
         }
@@ -32,7 +32,6 @@ def gc_feedback_impl(neon_env_builder: NeonEnvBuilder, zenbenchmark: NeonBenchma
     n_steps = 10
     n_update_iters = 100
     step_size = 10000
-    branch_created = 0
     with endpoint.cursor() as cur:
         cur.execute("SET statement_timeout='1000s'")
         cur.execute(
@@ -67,7 +66,6 @@ def gc_feedback_impl(neon_env_builder: NeonEnvBuilder, zenbenchmark: NeonBenchma
         if mode == "with_snapshots":
             if step == n_steps / 2:
                 env.create_branch("child")
-                branch_created += 1
 
     max_num_of_deltas_above_image = 0
     max_total_num_of_deltas = 0
@@ -144,15 +142,6 @@ def gc_feedback_impl(neon_env_builder: NeonEnvBuilder, zenbenchmark: NeonBenchma
     with layer_map_path.open("w") as f:
         f.write(json.dumps(client.timeline_layer_map_info(tenant_id, timeline_id)))
 
-    # We should have collected all garbage
-    if mode == "normal":
-        # in theory we should get physical size ~= logical size, but given that gc interval is 10s,
-        # and the layer has indexes that might contribute to the fluctuation, we allow a small margin
-        # of 1 here, and the end ratio we are asserting is 1 (margin) + 1 (expected) = 2.
-        assert physical_size / logical_size < 2
-    elif mode == "with_snapshots":
-        assert physical_size / logical_size < (2 + branch_created)
-
 
 @pytest.mark.timeout(10000)
 def test_gc_feedback(neon_env_builder: NeonEnvBuilder, zenbenchmark: NeonBenchmarker):

test_runner/performance/test_perf_many_relations.py

@@ -1,66 +0,0 @@
-import os
-from pathlib import Path
-
-import pytest
-from fixtures.compare_fixtures import RemoteCompare
-from fixtures.log_helper import log
-
-
-def get_num_relations(default: int = 1000) -> list[int]:
-    # We parametrize each run with scale specifying the number of wanted child partitions.
-    # Databases are pre-created and passed through BENCHMARK_CONNSTR env variable.
-    scales = os.getenv("TEST_NUM_RELATIONS", default=str(default))
-    rv = []
-    for s in scales.split(","):
-        scale = int(s)
-        rv.append(scale)
-    return rv
-
-
-@pytest.mark.parametrize("num_relations", get_num_relations())
-@pytest.mark.remote_cluster
-def test_perf_many_relations(remote_compare: RemoteCompare, num_relations: int):
-    """
-    Test creating many relations in a single database.
-    We use partitioned tables with child tables, indexes and constraints to have a realistic schema.
-    Also we include some common data types like text, uuid, timestamp, JSONB, etc.
-
-    see many_relations/create_many_relations.sql
-    """
-    env = remote_compare
-
-    # prepare some base tables and the plpgsql procedures that we use to create the tables
-    sql_file = Path(__file__).parent / "many_relations" / "create_many_relations.sql"
-    env.pg_bin.run_capture(["psql", env.pg.connstr(), "-f", str(sql_file)])
-
-    num_parent_tables = num_relations // 500 + 1
-    log.info(f"Creating {num_relations} relations in {num_parent_tables} parent tables")
-
-    log.info(f"Creating {num_parent_tables} parent tables")
-    sql = f"CALL create_partitioned_tables('operations_scale_{num_relations}', {num_parent_tables})"
-    log.info(sql)
-    env.pg_bin.run_capture(["psql", env.pg.connstr(), "-c", sql])
-
-    current_table = 0
-    num_relations_remaining = num_relations
-
-    # now run and measure the actual relation creation
-    while num_relations_remaining > 0:
-        current_table += 1
-        parent_table_name = f"operations_scale_{num_relations}_{current_table}"
-        if num_relations_remaining > 500:
-            num_relations_to_create = 500
-        else:
-            num_relations_to_create = num_relations_remaining
-        num_relations_remaining -= num_relations_to_create
-        log.info(
-            f"Creating {num_relations_to_create} child tables in partitioned parent table '{parent_table_name}'"
-        )
-        sql = f"CALL create_operations_partitions( '{parent_table_name}', '2000-01-01', ('2000-01-01'::DATE + INTERVAL '1 day' * {num_relations_to_create})::DATE)"
-        log.info(sql)
-        with env.zenbenchmark.record_duration(
-            f"CREATE_TABLE/{current_table}/{num_relations_to_create}"
-        ):
-            env.pg_bin.run_capture(
-                ["psql", env.pg.connstr(options="-cstatement_timeout=1000s "), "-c", sql]
-            )

test_runner/regress/test_compaction.py

@@ -134,10 +134,6 @@ def test_pageserver_gc_compaction_smoke(neon_env_builder: NeonEnvBuilder):
     }
 
     env = neon_env_builder.init_start(initial_tenant_conf=SMOKE_CONF)
-    env.pageserver.allowed_errors.append(
-        r".*failed to acquire partition lock during gc-compaction.*"
-    )
-    env.pageserver.allowed_errors.append(r".*repartition() called concurrently.*")
 
     tenant_id = env.initial_tenant
     timeline_id = env.initial_timeline
@@ -176,12 +172,6 @@ def test_pageserver_gc_compaction_smoke(neon_env_builder: NeonEnvBuilder):
 
         workload.churn_rows(row_count, env.pageserver.id)
 
-    def compaction_finished():
-        queue_depth = len(ps_http.timeline_compact_info(tenant_id, timeline_id))
-        assert queue_depth == 0
-
-    wait_until(compaction_finished, timeout=60)
-
     # ensure gc_compaction is scheduled and it's actually running (instead of skipping due to no layers picked)
     env.pageserver.assert_log_contains(
         "scheduled_compact_timeline.*picked .* layers for compaction"

test_runner/regress/test_import_pgdata.py

@@ -84,8 +84,6 @@ def test_pgdata_import_smoke(
     elif rel_block_size == RelBlockSize.TWO_STRPES_PER_SHARD:
         target_relblock_size = (shard_count or 1) * stripe_size * 8192 * 2
     elif rel_block_size == RelBlockSize.MULTIPLE_RELATION_SEGMENTS:
-        # Postgres uses a 1GiB segment size, fixed at compile time, so we must use >2GB of data
-        # to exercise multiple segments.
         target_relblock_size = int(((2.333 * 1024 * 1024 * 1024) // 8192) * 8192)
     else:
         raise ValueError
@@ -113,15 +111,9 @@ def test_pgdata_import_smoke(
 
     def validate_vanilla_equivalence(ep):
         # TODO: would be nicer to just compare pgdump
-
-        # Enable IO concurrency for batching on large sequential scan, to avoid making
-        # this test unnecessarily onerous on CPU
-        assert ep.safe_psql_many(
-            [
-                "set effective_io_concurrency=32;",
-                "select count(*), sum(data::bigint)::bigint from t",
-            ]
-        ) == [[], [(expect_nrows, expect_sum)]]
+        assert ep.safe_psql("select count(*), sum(data::bigint)::bigint from t") == [
+            (expect_nrows, expect_sum)
+        ]
 
     validate_vanilla_equivalence(vanilla_pg)
 

test_runner/regress/test_pageserver_layer_rolling.py

@@ -22,10 +22,7 @@ CHECKPOINT_TIMEOUT_SECONDS = 60
 
 
 async def run_worker_for_tenant(
-    env: NeonEnv,
-    entries: int,
-    tenant: TenantId,
-    offset: int | None = None,
+    env: NeonEnv, entries: int, tenant: TenantId, offset: int | None = None
 ) -> Lsn:
     if offset is None:
         offset = 0
@@ -40,20 +37,12 @@ async def run_worker_for_tenant(
         finally:
             await conn.close(timeout=10)
 
-        loop = asyncio.get_running_loop()
-        sql = await loop.run_in_executor(
-            None, lambda ep: ep.safe_psql("SELECT pg_current_wal_flush_lsn()"), ep
-        )
-        last_flush_lsn = Lsn(sql[0][0])
+        last_flush_lsn = Lsn(ep.safe_psql("SELECT pg_current_wal_flush_lsn()")[0][0])
         return last_flush_lsn
 
 
 async def run_worker(env: NeonEnv, tenant_conf, entries: int) -> tuple[TenantId, TimelineId, Lsn]:
-    loop = asyncio.get_running_loop()
-    # capture tenant_conf by specifying `tenant_conf=tenant_conf`, otherwise it will be evaluated to some random value
-    tenant, timeline = await loop.run_in_executor(
-        None, lambda tenant_conf, env: env.create_tenant(conf=tenant_conf), tenant_conf, env
-    )
+    tenant, timeline = env.create_tenant(conf=tenant_conf)
     last_flush_lsn = await run_worker_for_tenant(env, entries, tenant)
     return tenant, timeline, last_flush_lsn
 

test_runner/regress/test_physical_and_logical_replicaiton.py

@@ -2,7 +2,7 @@ from __future__ import annotations
 
 import time
 
-from fixtures.neon_fixtures import NeonEnv, logical_replication_sync, wait_replica_caughtup
+from fixtures.neon_fixtures import NeonEnv, logical_replication_sync
 
 
 def test_physical_and_logical_replication_slot_not_copied(neon_simple_env: NeonEnv, vanilla_pg):
@@ -38,8 +38,6 @@ def test_physical_and_logical_replication_slot_not_copied(neon_simple_env: NeonE
     for pk in range(n_records):
         p_cur.execute("insert into t (pk) values (%s)", (pk,))
 
-    wait_replica_caughtup(primary, secondary)
-
     s_cur.execute("select count(*) from t")
     assert s_cur.fetchall()[0][0] == n_records
 

test_runner/regress/test_proxy.py

@@ -43,11 +43,16 @@ async def test_http_pool_begin_1(static_proxy: NeonProxy):
     print(results)
 
 
-def test_proxy_select_1(static_proxy: NeonProxy):
+def test_proxy_select_1(static_proxy: NeonProxy, vanilla_pg: VanillaPostgres):
     """
     A simplest smoke test: check proxy against a local postgres instance.
     """
 
+    # Establish the default compute for this endpoint
+    vanilla_pg.safe_psql(
+        "INSERT INTO neon_control_plane.endpoints (endpoint_id) VALUES ('generic-project-name')"
+    )
+
     # no SNI, deprecated `options=project` syntax (before we had several endpoint in project)
     out = static_proxy.safe_psql("select 1", sslsni=0, options="project=generic-project-name")
     assert out[0][0] == 1
@@ -61,12 +66,17 @@ def test_proxy_select_1(static_proxy: NeonProxy):
     assert out[0][0] == 42
 
 
-def test_password_hack(static_proxy: NeonProxy):
+def test_password_hack(static_proxy: NeonProxy, vanilla_pg: VanillaPostgres):
     """
     Check the PasswordHack auth flow: an alternative to SCRAM auth for
     clients which can't provide the project/endpoint name via SNI or `options`.
     """
 
+    # Establish the default compute for this endpoint
+    vanilla_pg.safe_psql(
+        "INSERT INTO neon_control_plane.endpoints (endpoint_id) VALUES ('irrelevant')"
+    )
+
     user = "borat"
     password = "password"
     static_proxy.safe_psql(f"create role {user} with login password '{password}'")
@@ -107,7 +117,7 @@ async def test_link_auth(vanilla_pg: VanillaPostgres, link_proxy: NeonProxy):
 
 
 @pytest.mark.parametrize("option_name", ["project", "endpoint"])
-def test_proxy_options(static_proxy: NeonProxy, option_name: str):
+def test_proxy_options(static_proxy: NeonProxy, vanilla_pg: VanillaPostgres, option_name: str):
     """
     Check that we pass extra `options` to the PostgreSQL server:
     * `project=...` and `endpoint=...` shouldn't be passed at all
@@ -115,6 +125,11 @@ def test_proxy_options(static_proxy: NeonProxy, option_name: str):
     * everything else should be passed as-is.
     """
 
+    # Establish the default compute for this endpoint
+    vanilla_pg.safe_psql(
+        "INSERT INTO neon_control_plane.endpoints (endpoint_id) VALUES ('irrelevant')"
+    )
+
     options = f"{option_name}=irrelevant -cproxytest.option=value"
     out = static_proxy.safe_psql("show proxytest.option", options=options, sslsni=0)
     assert out[0][0] == "value"

test_runner/regress/test_sharding.py

@@ -11,7 +11,6 @@ from fixtures.common_types import Lsn, TenantId, TenantShardId, TimelineArchival
 from fixtures.compute_reconfigure import ComputeReconfigure
 from fixtures.log_helper import log
 from fixtures.neon_fixtures import (
-    DEFAULT_AZ_ID,
     NeonEnv,
     NeonEnvBuilder,
     StorageControllerApiException,
@@ -794,7 +793,6 @@ def test_sharding_split_stripe_size(
         "tenant_id": str(env.initial_tenant),
         "stripe_size": None,
         "shards": [{"node_id": int(env.pageservers[0].id), "shard_number": 0}],
-        "preferred_az": DEFAULT_AZ_ID,
     }
     assert notifications[0] == expect
 
@@ -814,7 +812,6 @@ def test_sharding_split_stripe_size(
             {"node_id": int(env.pageservers[0].id), "shard_number": 0},
             {"node_id": int(env.pageservers[0].id), "shard_number": 1},
         ],
-        "preferred_az": DEFAULT_AZ_ID,
     }
     log.info(f"Got notification: {notifications[1]}")
     assert notifications[1] == expect_after

test_runner/regress/test_storage_controller.py

@@ -16,7 +16,6 @@ from fixtures.common_types import TenantId, TenantShardId, TimelineId
 from fixtures.compute_reconfigure import ComputeReconfigure
 from fixtures.log_helper import log
 from fixtures.neon_fixtures import (
-    DEFAULT_AZ_ID,
     NeonEnv,
     NeonEnvBuilder,
     NeonPageserver,
@@ -600,7 +599,6 @@ def test_storage_controller_compute_hook(
         "tenant_id": str(env.initial_tenant),
         "stripe_size": None,
         "shards": [{"node_id": int(env.pageservers[0].id), "shard_number": 0}],
-        "preferred_az": DEFAULT_AZ_ID,
     }
     assert notifications[0] == expect
 
@@ -618,7 +616,6 @@ def test_storage_controller_compute_hook(
         "tenant_id": str(env.initial_tenant),
         "stripe_size": None,
         "shards": [{"node_id": int(env.pageservers[1].id), "shard_number": 0}],
-        "preferred_az": DEFAULT_AZ_ID,
     }
 
     def received_migration_notification():
@@ -646,7 +643,6 @@ def test_storage_controller_compute_hook(
             {"node_id": int(env.pageservers[1].id), "shard_number": 0},
             {"node_id": int(env.pageservers[1].id), "shard_number": 1},
         ],
-        "preferred_az": DEFAULT_AZ_ID,
     }
 
     def received_split_notification():
@@ -718,7 +714,6 @@ def test_storage_controller_stuck_compute_hook(
         "tenant_id": str(env.initial_tenant),
         "stripe_size": None,
         "shards": [{"node_id": int(env.pageservers[0].id), "shard_number": 0}],
-        "preferred_az": DEFAULT_AZ_ID,
     }
     assert notifications[0] == expect
 
@@ -3009,7 +3004,7 @@ def test_safekeeper_deployment_time_update(neon_env_builder: NeonEnvBuilder):
 def eq_safekeeper_records(a: dict[str, Any], b: dict[str, Any]) -> bool:
     compared = [dict(a), dict(b)]
 
-    masked_keys = ["created_at", "updated_at", "active"]
+    masked_keys = ["created_at", "updated_at"]
 
     for d in compared:
         # keep deleting these in case we are comparing the body as it will be uploaded by real scripts

vendor/revisions.json

@@ -1,7 +1,7 @@
 {
   "v17": [
     "17.2",
-    "7e3f3974bc8895938308f94d0e96879ffae638cd"
+    "65c4e46baf56ec05412c7dd63d62faff0b33dcfb"
   ],
   "v16": [
     "16.6",