pageserver_api: implement keyspace range iterator (#6435)

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

@@ -348,10 +348,6 @@ jobs:
           rerun_failed: true
           pg_version: ${{ matrix.pg_version }}
           aws-oicd-role-arn: ${{ vars.DEV_AWS_OIDC_ROLE_ARN }}
-          # `--session-timeout` is equal to (timeout-minutes - 10 minutes) * 60 seconds.
-          # Attempt to stop tests gracefully to generate test reports
-          # until they are forcibly stopped by the stricter `timeout-minutes` limit.
-          extra_params: --session-timeout=${{ inputs.sanitizers != 'enabled' && 3000 || 10200 }}
         env:
           TEST_RESULT_CONNSTR: ${{ secrets.REGRESS_TEST_RESULT_CONNSTR_NEW }}
           CHECK_ONDISK_DATA_COMPATIBILITY: nonempty

.github/workflows/build_and_test.yml

@@ -263,9 +263,8 @@ jobs:
           echo "json=$(jq --compact-output '.' /tmp/benchmark_durations.json)" >> $GITHUB_OUTPUT
 
   benchmarks:
-    # `!failure() && !cancelled()` is required because the workflow depends on the job that can be skipped: `deploy` in PRs
-    if: github.ref_name == 'main' || (contains(github.event.pull_request.labels.*.name, 'run-benchmarks') && !failure() && !cancelled())
-    needs: [ check-permissions, build-build-tools-image, get-benchmarks-durations, deploy ]
+    if: github.ref_name == 'main' || contains(github.event.pull_request.labels.*.name, 'run-benchmarks')
+    needs: [ check-permissions, build-and-test-locally, build-build-tools-image, get-benchmarks-durations ]
     permissions:
       id-token: write # aws-actions/configure-aws-credentials
       statuses: write
@@ -893,7 +892,7 @@ jobs:
       docker-hub-password: ${{ secrets.NEON_DOCKERHUB_PASSWORD }}
 
   push-neon-image-prod:
-    if: github.ref_name == 'release' || github.ref_name == 'release-proxy' || github.ref_name == 'release-compute'
+    if: github.ref_name == 'main' || github.ref_name == 'release' || github.ref_name == 'release-proxy' || github.ref_name == 'release-compute'
     needs: [ generate-image-maps, neon-image, test-images ]
     uses: ./.github/workflows/_push-to-container-registry.yml
     with:
@@ -910,7 +909,7 @@ jobs:
       docker-hub-password: ${{ secrets.NEON_DOCKERHUB_PASSWORD }}
 
   push-compute-image-prod:
-    if: github.ref_name == 'release' || github.ref_name == 'release-proxy' || github.ref_name == 'release-compute'
+    if: github.ref_name == 'main' || github.ref_name == 'release' || github.ref_name == 'release-proxy' || github.ref_name == 'release-compute'
     needs: [ generate-image-maps, vm-compute-node-image, test-images ]
     uses: ./.github/workflows/_push-to-container-registry.yml
     with:

.github/workflows/force-test-extensions-upgrade.yml

@@ -1,76 +0,0 @@
-name: Force Test Upgrading of Extension
-on:
-  schedule:
-    # * is a special character in YAML so you have to quote this string
-    #          ┌───────────── minute (0 - 59)
-    #          │ ┌───────────── hour (0 - 23)
-    #          │ │ ┌───────────── day of the month (1 - 31)
-    #          │ │ │ ┌───────────── month (1 - 12 or JAN-DEC)
-    #          │ │ │ │ ┌───────────── day of the week (0 - 6 or SUN-SAT)
-    - cron:  '45 2 * * *' # run once a day, timezone is utc
-  workflow_dispatch: # adds ability to run this manually
-
-defaults:
-  run:
-    shell: bash -euxo pipefail {0}
-
-concurrency:
-  # Allow only one workflow
-  group: ${{ github.workflow }}
-  cancel-in-progress: true
-
-permissions:
-  id-token: write # aws-actions/configure-aws-credentials
-  statuses: write
-  contents: read
-
-jobs:
-  regress:
-    strategy:
-      fail-fast: false
-      matrix:
-        pg-version: [16, 17]
-
-    runs-on: small
-
-    steps:
-      - uses: actions/checkout@v4
-        with:
-          submodules: false
-
-      - name: Get the last compute release tag
-        id: get-last-compute-release-tag
-        env:
-          GH_TOKEN: ${{ secrets.GITHUB_TOKEN }}
-        run: |
-          tag=$(gh api -q '[.[].tag_name | select(startswith("release-compute"))][0]'\
-            -H "Accept: application/vnd.github+json" \
-            -H "X-GitHub-Api-Version: 2022-11-28" \
-            "/repos/${GITHUB_REPOSITORY}/releases")
-          echo tag=${tag} >> ${GITHUB_OUTPUT}
-
-      - name: Test extension upgrade
-        timeout-minutes: 20
-        env:
-          NEWTAG: latest
-          OLDTAG: ${{ steps.get-last-compute-release-tag.outputs.tag }}
-          PG_VERSION: ${{ matrix.pg-version }}
-          FORCE_ALL_UPGRADE_TESTS: true
-        run: ./docker-compose/test_extensions_upgrade.sh
-
-      - name: Print logs and clean up
-        if: always()
-        run: |
-          docker compose --profile test-extensions -f ./docker-compose/docker-compose.yml logs || true
-          docker compose --profile test-extensions -f ./docker-compose/docker-compose.yml down
-
-      - name: Post to the Slack channel
-        if: ${{ github.event.schedule && failure() }}
-        uses: slackapi/slack-github-action@v1
-        with:
-          channel-id: ${{ vars.SLACK_ON_CALL_QA_STAGING_STREAM }}
-          slack-message: |
-            Test upgrading of extensions: ${{ job.status }}
-            <${{ github.server_url }}/${{ github.repository }}/actions/runs/${{ github.run_id }}|GitHub Run>
-        env:
-          SLACK_BOT_TOKEN: ${{ secrets.SLACK_BOT_TOKEN }}

.github/workflows/regenerate-pg-setting.yml

@@ -1,41 +0,0 @@
-name: Regenerate Postgres Settings
-
-on:
-  pull_request:
-    types:
-      - opened
-      - synchronize
-      - reopened
-    paths:
-      - pgxn/neon/**.c
-      - vendor/postgres-v*
-      - vendor/revisions.json
-
-concurrency:
-  group: ${{ github.workflow }}-${{ github.head_ref }}
-  cancel-in-progress: true
-
-permissions:
-  pull-requests: write
-
-jobs:
-  regenerate-pg-settings:
-    runs-on: ubuntu-22.04
-
-    steps:
-      - name: Add comment
-        uses: thollander/actions-comment-pull-request@v3
-        with:
-          comment-tag: ${{ github.job }}
-          pr-number: ${{ github.event.number }}
-          message: |
-            If this PR added a GUC in the Postgres fork or `neon` extension,
-            please regenerate the Postgres settings in the `cloud` repo:
-
-            ```
-            make NEON_WORKDIR=path/to/neon/checkout \
-              -C goapp/internal/shareddomain/postgres generate
-            ```
-
-            If you're an external contributor, a Neon employee will assist in
-            making sure this step is done.

.github/workflows/trigger-e2e-tests.yml

@@ -88,7 +88,7 @@ jobs:
           BUILD_AND_TEST_RUN_ID=${TAG}
           while true; do
             gh run --repo ${GITHUB_REPOSITORY} view ${BUILD_AND_TEST_RUN_ID} --json jobs --jq '[.jobs[] | select((.name | startswith("push-neon-image-dev")) or (.name | startswith("push-compute-image-dev"))) | {"name": .name, "conclusion": .conclusion, "url": .url}]' > jobs.json
-            if [ $(jq '[.[] | select(.conclusion == "success")] | length' jobs.json) -eq 2 ]; then
+            if [ $(jq '[.[] | select(.conclusion == "success")]' jobs.json) -eq 2 ]; then
               break
             fi
             jq -c '.[]' jobs.json | while read -r job; do

Cargo.lock

@@ -786,7 +786,7 @@ dependencies = [
 [[package]]
 name = "azure_core"
 version = "0.21.0"
-source = "git+https://github.com/neondatabase/azure-sdk-for-rust.git?branch=neon#f64bd57262ced51afce5d8909c06dcb11a6dd85a"
+source = "git+https://github.com/neondatabase/azure-sdk-for-rust.git?branch=neon#c36ed4c039bb3d59b5a1705f2cc337636c73b541"
 dependencies = [
  "async-trait",
  "base64 0.22.1",
@@ -815,7 +815,7 @@ dependencies = [
 [[package]]
 name = "azure_identity"
 version = "0.21.0"
-source = "git+https://github.com/neondatabase/azure-sdk-for-rust.git?branch=neon#f64bd57262ced51afce5d8909c06dcb11a6dd85a"
+source = "git+https://github.com/neondatabase/azure-sdk-for-rust.git?branch=neon#c36ed4c039bb3d59b5a1705f2cc337636c73b541"
 dependencies = [
  "async-lock",
  "async-trait",
@@ -834,7 +834,7 @@ dependencies = [
 [[package]]
 name = "azure_storage"
 version = "0.21.0"
-source = "git+https://github.com/neondatabase/azure-sdk-for-rust.git?branch=neon#f64bd57262ced51afce5d8909c06dcb11a6dd85a"
+source = "git+https://github.com/neondatabase/azure-sdk-for-rust.git?branch=neon#c36ed4c039bb3d59b5a1705f2cc337636c73b541"
 dependencies = [
  "RustyXML",
  "async-lock",
@@ -852,7 +852,7 @@ dependencies = [
 [[package]]
 name = "azure_storage_blobs"
 version = "0.21.0"
-source = "git+https://github.com/neondatabase/azure-sdk-for-rust.git?branch=neon#f64bd57262ced51afce5d8909c06dcb11a6dd85a"
+source = "git+https://github.com/neondatabase/azure-sdk-for-rust.git?branch=neon#c36ed4c039bb3d59b5a1705f2cc337636c73b541"
 dependencies = [
  "RustyXML",
  "azure_core",
@@ -872,7 +872,7 @@ dependencies = [
 [[package]]
 name = "azure_svc_blobstorage"
 version = "0.21.0"
-source = "git+https://github.com/neondatabase/azure-sdk-for-rust.git?branch=neon#f64bd57262ced51afce5d8909c06dcb11a6dd85a"
+source = "git+https://github.com/neondatabase/azure-sdk-for-rust.git?branch=neon#c36ed4c039bb3d59b5a1705f2cc337636c73b541"
 dependencies = [
  "azure_core",
  "bytes",
@@ -1029,6 +1029,12 @@ dependencies = [
  "generic-array",
 ]
 
+[[package]]
+name = "boxcar"
+version = "0.2.8"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "2721c3c5a6f0e7f7e607125d963fedeb765f545f67adc9d71ed934693881eb42"
+
 [[package]]
 name = "bstr"
 version = "1.5.0"
@@ -1287,7 +1293,6 @@ version = "0.1.0"
 dependencies = [
  "anyhow",
  "chrono",
- "jsonwebtoken",
  "regex",
  "remote_storage",
  "serde",
@@ -1303,7 +1308,6 @@ dependencies = [
  "aws-config",
  "aws-sdk-kms",
  "aws-sdk-s3",
- "aws-smithy-types",
  "axum",
  "base64 0.13.1",
  "bytes",
@@ -1316,7 +1320,6 @@ dependencies = [
  "flate2",
  "futures",
  "http 1.1.0",
- "jsonwebtoken",
  "metrics",
  "nix 0.27.1",
  "notify",
@@ -1352,7 +1355,6 @@ dependencies = [
  "utils",
  "uuid",
  "vm_monitor",
- "walkdir",
  "workspace_hack",
  "zstd",
 ]
@@ -4925,6 +4927,7 @@ dependencies = [
  "aws-sdk-iam",
  "aws-sigv4",
  "base64 0.13.1",
+ "boxcar",
  "bstr",
  "bytes",
  "camino",
@@ -4976,6 +4979,7 @@ dependencies = [
  "postgres-protocol2",
  "postgres_backend",
  "pq_proto",
+ "prometheus",
  "rand 0.8.5",
  "rand_distr",
  "rcgen",
@@ -5000,6 +5004,7 @@ dependencies = [
  "smallvec",
  "smol_str",
  "socket2",
+ "strum",
  "strum_macros",
  "subtle",
  "thiserror 1.0.69",
@@ -5014,6 +5019,7 @@ dependencies = [
  "tracing",
  "tracing-log",
  "tracing-opentelemetry",
+ "tracing-serde",
  "tracing-subscriber",
  "tracing-utils",
  "try-lock",
@@ -6454,13 +6460,10 @@ dependencies = [
  "pageserver_client",
  "postgres_connection",
  "rand 0.8.5",
- "regex",
  "reqwest",
  "routerify",
  "rustls 0.23.18",
  "rustls-native-certs 0.8.0",
- "safekeeper_api",
- "safekeeper_client",
  "scoped-futures",
  "scopeguard",
  "serde",

Dockerfile

@@ -50,14 +50,6 @@ RUN set -e \
     && rm -rf pg_install/build \
     && tar -C pg_install -czf /home/nonroot/postgres_install.tar.gz .
 
-# Prepare cargo-chef recipe
-FROM $REPOSITORY/$IMAGE:$TAG AS plan
-WORKDIR /home/nonroot
-
-COPY --chown=nonroot . .
-
-RUN cargo chef prepare --recipe-path recipe.json
-
 # Build neon binaries
 FROM $REPOSITORY/$IMAGE:$TAG AS build
 WORKDIR /home/nonroot
@@ -71,15 +63,9 @@ COPY --from=pg-build /home/nonroot/pg_install/v16/include/postgresql/server pg_i
 COPY --from=pg-build /home/nonroot/pg_install/v17/include/postgresql/server pg_install/v17/include/postgresql/server
 COPY --from=pg-build /home/nonroot/pg_install/v16/lib                       pg_install/v16/lib
 COPY --from=pg-build /home/nonroot/pg_install/v17/lib                       pg_install/v17/lib
-COPY --from=plan     /home/nonroot/recipe.json                              recipe.json
-
-ARG ADDITIONAL_RUSTFLAGS=""
-
-RUN set -e \
-    && RUSTFLAGS="-Clinker=clang -Clink-arg=-fuse-ld=mold -Clink-arg=-Wl,--no-rosegment -Cforce-frame-pointers=yes ${ADDITIONAL_RUSTFLAGS}" cargo chef cook --locked --release --recipe-path recipe.json
-
 COPY --chown=nonroot . .
 
+ARG ADDITIONAL_RUSTFLAGS
 RUN set -e \
     && RUSTFLAGS="-Clinker=clang -Clink-arg=-fuse-ld=mold -Clink-arg=-Wl,--no-rosegment -Cforce-frame-pointers=yes ${ADDITIONAL_RUSTFLAGS}" cargo build \
       --bin pg_sni_router  \

build-tools.Dockerfile

@@ -300,7 +300,6 @@ ARG CARGO_HAKARI_VERSION=0.9.33
 ARG CARGO_DENY_VERSION=0.16.2
 ARG CARGO_HACK_VERSION=0.6.33
 ARG CARGO_NEXTEST_VERSION=0.9.85
-ARG CARGO_CHEF_VERSION=0.1.71
 ARG CARGO_DIESEL_CLI_VERSION=2.2.6
 RUN curl -sSO https://static.rust-lang.org/rustup/dist/$(uname -m)-unknown-linux-gnu/rustup-init && whoami && \
 	chmod +x rustup-init && \
@@ -315,7 +314,6 @@ RUN curl -sSO https://static.rust-lang.org/rustup/dist/$(uname -m)-unknown-linux
     cargo install cargo-deny --locked --version ${CARGO_DENY_VERSION} && \
     cargo install cargo-hack          --version ${CARGO_HACK_VERSION} && \
     cargo install cargo-nextest       --version ${CARGO_NEXTEST_VERSION} && \
-    cargo install cargo-chef --locked --version ${CARGO_CHEF_VERSION} && \
     cargo install diesel_cli          --version ${CARGO_DIESEL_CLI_VERSION} \
                                       --features postgres-bundled --no-default-features && \
     rm -rf /home/nonroot/.cargo/registry && \

compute/compute-node.Dockerfile

@@ -148,7 +148,7 @@ RUN case $DEBIAN_VERSION in \
     apt install --no-install-recommends --no-install-suggests -y \
     ninja-build git autoconf automake libtool build-essential bison flex libreadline-dev \
     zlib1g-dev libxml2-dev libcurl4-openssl-dev libossp-uuid-dev wget ca-certificates pkg-config libssl-dev \
-    libicu-dev libxslt1-dev liblz4-dev libzstd-dev zstd curl unzip g++ \
+    libicu-dev libxslt1-dev liblz4-dev libzstd-dev zstd curl unzip \
     $VERSION_INSTALLS \
     && apt clean && rm -rf /var/lib/apt/lists/*
 
@@ -1464,31 +1464,6 @@ RUN make release -j $(getconf _NPROCESSORS_ONLN) && \
     make install -j $(getconf _NPROCESSORS_ONLN) && \
     echo 'trusted = true' >> /usr/local/pgsql/share/extension/pg_mooncake.control
 
-#########################################################################################
-#
-# Layer "pg-duckdb-pg-build"
-# compile pg_duckdb extension
-#
-#########################################################################################
-FROM build-deps AS pg_duckdb-src
-WORKDIR /ext-src
-COPY compute/patches/pg_duckdb_v031.patch .
-# pg_duckdb build requires source dir to be a git repo to get submodules
-# allow neon_superuser to execute some functions that in pg_duckdb are available to superuser only: 
-# - extension management function duckdb.install_extension()
-# - access to duckdb.extensions table and its sequence
-RUN git clone --depth 1 --branch v0.3.1 https://github.com/duckdb/pg_duckdb.git pg_duckdb-src && \
-    cd pg_duckdb-src && \
-    git submodule update --init --recursive && \
-    patch -p1 < /ext-src/pg_duckdb_v031.patch
-
-FROM pg-build AS pg_duckdb-build
-ARG PG_VERSION
-COPY --from=pg_duckdb-src /ext-src/ /ext-src/
-WORKDIR /ext-src/pg_duckdb-src
-RUN make install -j $(getconf _NPROCESSORS_ONLN) && \
-    echo 'trusted = true' >> /usr/local/pgsql/share/extension/pg_duckdb.control 
-        
 #########################################################################################
 #
 # Layer "pg_repack"
@@ -1602,7 +1577,6 @@ COPY --from=pg_anon-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_duckdb-build /usr/local/pgsql/ /usr/local/pgsql/
 COPY --from=pg_repack-build /usr/local/pgsql/ /usr/local/pgsql/
 
 #########################################################################################
@@ -1695,6 +1669,29 @@ RUN if [ "$TARGETARCH" = "amd64" ]; then\
     && echo "${pgbouncer_exporter_sha256} pgbouncer_exporter" | sha256sum -c -\
     && echo "${sql_exporter_sha256} sql_exporter" | sha256sum -c -
 
+#########################################################################################
+#
+# Layer "awscli"
+#
+#########################################################################################
+FROM build-deps AS awscli
+ARG TARGETARCH
+RUN set -ex; \
+    if [ "${TARGETARCH}" = "amd64" ]; then \
+        TARGETARCH_ALT="x86_64"; \
+        CHECKSUM="c9a9df3770a3ff9259cb469b6179e02829687a464e0824d5c32d378820b53a00"; \
+    elif [ "${TARGETARCH}" = "arm64" ]; then \
+        TARGETARCH_ALT="aarch64"; \
+        CHECKSUM="8181730be7891582b38b028112e81b4899ca817e8c616aad807c9e9d1289223a"; \
+    else \
+        echo "Unsupported architecture: ${TARGETARCH}"; exit 1; \
+    fi; \
+    curl --retry 5 -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
+
 #########################################################################################
 #
 # Clean up postgres folder before inclusion
@@ -1753,7 +1750,7 @@ COPY --from=pg_graphql-src /ext-src/ /ext-src/
 COPY --from=hypopg-src /ext-src/ /ext-src/
 COPY --from=pg_hashids-src /ext-src/ /ext-src/
 COPY --from=rum-src /ext-src/ /ext-src/
-COPY --from=pgtap-src /ext-src/ /ext-src/
+#COPY --from=pgtap-src /ext-src/ /ext-src/
 COPY --from=ip4r-src /ext-src/ /ext-src/
 COPY --from=prefix-src /ext-src/ /ext-src/
 COPY --from=hll-src /ext-src/ /ext-src/
@@ -1864,6 +1861,9 @@ RUN mkdir /var/db && useradd -m -d /var/db/postgres postgres && \
     mkdir /usr/local/download_extensions && \
     chown -R postgres:postgres /usr/local/download_extensions
 
+# aws cli is used by fast_import
+COPY --from=awscli /usr/local/aws-cli /usr/local/aws-cli
+
 # pgbouncer and its config
 COPY --from=pgbouncer         /usr/local/pgbouncer/bin/pgbouncer /usr/local/bin/pgbouncer
 COPY --chmod=0666 --chown=postgres compute/etc/pgbouncer.ini /etc/pgbouncer.ini

compute/patches/pg_duckdb_v031.patch

@@ -1,11 +0,0 @@
-diff --git a/sql/pg_duckdb--0.2.0--0.3.0.sql b/sql/pg_duckdb--0.2.0--0.3.0.sql
-index d777d76..af60106 100644
---- a/sql/pg_duckdb--0.2.0--0.3.0.sql
-+++ b/sql/pg_duckdb--0.2.0--0.3.0.sql
-@@ -1056,3 +1056,6 @@ GRANT ALL ON FUNCTION duckdb.cache(TEXT, TEXT) TO PUBLIC;
- GRANT ALL ON FUNCTION duckdb.cache_info() TO PUBLIC;
- GRANT ALL ON FUNCTION duckdb.cache_delete(TEXT) TO PUBLIC;
- GRANT ALL ON PROCEDURE duckdb.recycle_ddb() TO PUBLIC;
-+GRANT ALL ON FUNCTION duckdb.install_extension(TEXT) TO neon_superuser;
-+GRANT ALL ON TABLE duckdb.extensions TO neon_superuser;
-+GRANT ALL ON SEQUENCE duckdb.extensions_table_seq TO neon_superuser;

compute/vm-image-spec-bullseye.yaml

@@ -47,9 +47,7 @@ files:
       # Allow postgres user (which is what compute_ctl runs as) to run /neonvm/bin/resize-swap
       # and /neonvm/bin/set-disk-quota as root without requiring entering a password (NOPASSWD),
       # regardless of hostname (ALL)
-      #
-      # Also allow it to shut down the VM. The fast_import job does that when it's finished.
-      postgres ALL=(root) NOPASSWD: /neonvm/bin/resize-swap, /neonvm/bin/set-disk-quota, /neonvm/bin/poweroff
+      postgres ALL=(root) NOPASSWD: /neonvm/bin/resize-swap, /neonvm/bin/set-disk-quota
   - filename: cgconfig.conf
     content: |
       # Configuration for cgroups in VM compute nodes

compute_tools/Cargo.toml

@@ -14,7 +14,6 @@ base64.workspace = true
 aws-config.workspace = true
 aws-sdk-s3.workspace = true
 aws-sdk-kms.workspace = true
-aws-smithy-types.workspace = true
 anyhow.workspace = true
 axum = { workspace = true, features = [] }
 camino.workspace = true
@@ -25,7 +24,6 @@ fail.workspace = true
 flate2.workspace = true
 futures.workspace = true
 http.workspace = true
-jsonwebtoken.workspace = true
 metrics.workspace = true
 nix.workspace = true
 notify.workspace = true
@@ -55,7 +53,6 @@ thiserror.workspace = true
 url.workspace = true
 uuid.workspace = true
 prometheus.workspace = true
-walkdir.workspace = true
 
 postgres_initdb.workspace = true
 compute_api.workspace = true

compute_tools/src/bin/compute_ctl.rs

@@ -55,7 +55,7 @@ use signal_hook::{consts::SIGINT, iterator::Signals};
 use tracing::{error, info, warn};
 use url::Url;
 
-use compute_api::responses::{ComputeCtlConfig, ComputeStatus};
+use compute_api::responses::ComputeStatus;
 use compute_api::spec::ComputeSpec;
 
 use compute_tools::compute::{
@@ -281,7 +281,6 @@ fn try_spec_from_cli(cli: &Cli) -> Result<CliSpecParams> {
         info!("got spec from cli argument {}", spec_json);
         return Ok(CliSpecParams {
             spec: Some(serde_json::from_str(spec_json)?),
-            compute_ctl_config: ComputeCtlConfig::default(),
             live_config_allowed: false,
         });
     }
@@ -291,7 +290,6 @@ fn try_spec_from_cli(cli: &Cli) -> Result<CliSpecParams> {
         let file = File::open(Path::new(spec_path))?;
         return Ok(CliSpecParams {
             spec: Some(serde_json::from_reader(file)?),
-            compute_ctl_config: ComputeCtlConfig::default(),
             live_config_allowed: true,
         });
     }
@@ -301,9 +299,8 @@ fn try_spec_from_cli(cli: &Cli) -> Result<CliSpecParams> {
     };
 
     match get_spec_from_control_plane(cli.control_plane_uri.as_ref().unwrap(), &cli.compute_id) {
-        Ok(resp) => Ok(CliSpecParams {
-            spec: resp.0,
-            compute_ctl_config: resp.1,
+        Ok(spec) => Ok(CliSpecParams {
+            spec,
             live_config_allowed: true,
         }),
         Err(e) => {
@@ -320,8 +317,6 @@ fn try_spec_from_cli(cli: &Cli) -> Result<CliSpecParams> {
 struct CliSpecParams {
     /// If a spec was provided via CLI or file, the [`ComputeSpec`]
     spec: Option<ComputeSpec>,
-    #[allow(dead_code)]
-    compute_ctl_config: ComputeCtlConfig,
     live_config_allowed: bool,
 }
 
@@ -331,7 +326,6 @@ fn wait_spec(
     CliSpecParams {
         spec,
         live_config_allowed,
-        compute_ctl_config: _,
     }: CliSpecParams,
 ) -> Result<Arc<ComputeNode>> {
     let mut new_state = ComputeState::new();

compute_tools/src/bin/fast_import.rs

@@ -25,10 +25,10 @@
 //! docker push localhost:3030/localregistry/compute-node-v14:latest
 //! ```
 
-use anyhow::{bail, Context};
+use anyhow::Context;
 use aws_config::BehaviorVersion;
 use camino::{Utf8Path, Utf8PathBuf};
-use clap::{Parser, Subcommand};
+use clap::Parser;
 use compute_tools::extension_server::{get_pg_version, PostgresMajorVersion};
 use nix::unistd::Pid;
 use tracing::{error, info, info_span, warn, Instrument};
@@ -44,59 +44,32 @@ mod s3_uri;
 const PG_WAIT_TIMEOUT: std::time::Duration = std::time::Duration::from_secs(600);
 const PG_WAIT_RETRY_INTERVAL: std::time::Duration = std::time::Duration::from_millis(300);
 
-#[derive(Subcommand, Debug)]
-enum Command {
-    /// Runs local postgres (neon binary), restores into it,
-    /// uploads pgdata to s3 to be consumed by pageservers
-    Pgdata {
-        /// Raw connection string to the source database. Used only in tests,
-        /// real scenario uses encrypted connection string in spec.json from s3.
-        #[clap(long)]
-        source_connection_string: Option<String>,
-        /// If specified, will not shut down the local postgres after the import. Used in local testing
-        #[clap(short, long)]
-        interactive: bool,
-        /// Port to run postgres on. Default is 5432.
-        #[clap(long, default_value_t = 5432)]
-        pg_port: u16, // port to run postgres on, 5432 is default
-
-        /// Number of CPUs in the system. This is used to configure # of
-        /// parallel worker processes, for index creation.
-        #[clap(long, env = "NEON_IMPORTER_NUM_CPUS")]
-        num_cpus: Option<usize>,
-
-        /// Amount of RAM in the system. This is used to configure shared_buffers
-        /// and maintenance_work_mem.
-        #[clap(long, env = "NEON_IMPORTER_MEMORY_MB")]
-        memory_mb: Option<usize>,
-    },
-
-    /// Runs pg_dump-pg_restore from source to destination without running local postgres.
-    DumpRestore {
-        /// Raw connection string to the source database. Used only in tests,
-        /// real scenario uses encrypted connection string in spec.json from s3.
-        #[clap(long)]
-        source_connection_string: Option<String>,
-        /// Raw connection string to the destination database. Used only in tests,
-        /// real scenario uses encrypted connection string in spec.json from s3.
-        #[clap(long)]
-        destination_connection_string: Option<String>,
-    },
-}
-
 #[derive(clap::Parser)]
 struct Args {
-    #[clap(long, env = "NEON_IMPORTER_WORKDIR")]
+    #[clap(long)]
     working_directory: Utf8PathBuf,
     #[clap(long, env = "NEON_IMPORTER_S3_PREFIX")]
     s3_prefix: Option<s3_uri::S3Uri>,
-    #[clap(long, env = "NEON_IMPORTER_PG_BIN_DIR")]
+    #[clap(long)]
+    source_connection_string: Option<String>,
+    #[clap(short, long)]
+    interactive: bool,
+    #[clap(long)]
     pg_bin_dir: Utf8PathBuf,
-    #[clap(long, env = "NEON_IMPORTER_PG_LIB_DIR")]
+    #[clap(long)]
     pg_lib_dir: Utf8PathBuf,
+    #[clap(long)]
+    pg_port: Option<u16>, // port to run postgres on, 5432 is default
 
-    #[clap(subcommand)]
-    command: Command,
+    /// Number of CPUs in the system. This is used to configure # of
+    /// parallel worker processes, for index creation.
+    #[clap(long, env = "NEON_IMPORTER_NUM_CPUS")]
+    num_cpus: Option<usize>,
+
+    /// Amount of RAM in the system. This is used to configure shared_buffers
+    /// and maintenance_work_mem.
+    #[clap(long, env = "NEON_IMPORTER_MEMORY_MB")]
+    memory_mb: Option<usize>,
 }
 
 #[serde_with::serde_as]
@@ -105,8 +78,6 @@ struct Spec {
     encryption_secret: EncryptionSecret,
     #[serde_as(as = "serde_with::base64::Base64")]
     source_connstring_ciphertext_base64: Vec<u8>,
-    #[serde_as(as = "Option<serde_with::base64::Base64>")]
-    destination_connstring_ciphertext_base64: Option<Vec<u8>>,
 }
 
 #[derive(serde::Deserialize)]
@@ -122,150 +93,192 @@ const DEFAULT_LOCALE: &str = if cfg!(target_os = "macos") {
     "C.UTF-8"
 };
 
-async fn decode_connstring(
-    kms_client: &aws_sdk_kms::Client,
-    key_id: &String,
-    connstring_ciphertext_base64: Vec<u8>,
-) -> Result<String, anyhow::Error> {
-    let mut output = kms_client
-        .decrypt()
-        .key_id(key_id)
-        .ciphertext_blob(aws_sdk_s3::primitives::Blob::new(
-            connstring_ciphertext_base64,
-        ))
-        .send()
-        .await
-        .context("decrypt connection string")?;
+#[tokio::main]
+pub(crate) async fn main() -> anyhow::Result<()> {
+    utils::logging::init(
+        utils::logging::LogFormat::Plain,
+        utils::logging::TracingErrorLayerEnablement::EnableWithRustLogFilter,
+        utils::logging::Output::Stdout,
+    )?;
 
-    let plaintext = output
-        .plaintext
-        .take()
-        .context("get plaintext connection string")?;
+    info!("starting");
 
-    String::from_utf8(plaintext.into_inner()).context("parse connection string as utf8")
-}
+    let args = Args::parse();
 
-struct PostgresProcess {
-    pgdata_dir: Utf8PathBuf,
-    pg_bin_dir: Utf8PathBuf,
-    pgbin: Utf8PathBuf,
-    pg_lib_dir: Utf8PathBuf,
-    postgres_proc: Option<tokio::process::Child>,
-}
-
-impl PostgresProcess {
-    fn new(pgdata_dir: Utf8PathBuf, pg_bin_dir: Utf8PathBuf, pg_lib_dir: Utf8PathBuf) -> Self {
-        Self {
-            pgdata_dir,
-            pgbin: pg_bin_dir.join("postgres"),
-            pg_bin_dir,
-            pg_lib_dir,
-            postgres_proc: None,
-        }
+    // Validate arguments
+    if args.s3_prefix.is_none() && args.source_connection_string.is_none() {
+        anyhow::bail!("either s3_prefix or source_connection_string must be specified");
+    }
+    if args.s3_prefix.is_some() && args.source_connection_string.is_some() {
+        anyhow::bail!("only one of s3_prefix or source_connection_string can be specified");
     }
 
-    async fn prepare(&self, initdb_user: &str) -> Result<(), anyhow::Error> {
-        tokio::fs::create_dir(&self.pgdata_dir)
-            .await
-            .context("create pgdata directory")?;
+    let working_directory = args.working_directory;
+    let pg_bin_dir = args.pg_bin_dir;
+    let pg_lib_dir = args.pg_lib_dir;
+    let pg_port = args.pg_port.unwrap_or_else(|| {
+        info!("pg_port not specified, using default 5432");
+        5432
+    });
 
-        let pg_version = match get_pg_version(self.pgbin.as_ref()) {
-            PostgresMajorVersion::V14 => 14,
-            PostgresMajorVersion::V15 => 15,
-            PostgresMajorVersion::V16 => 16,
-            PostgresMajorVersion::V17 => 17,
+    // Initialize AWS clients only if s3_prefix is specified
+    let (aws_config, kms_client) = if args.s3_prefix.is_some() {
+        let config = aws_config::load_defaults(BehaviorVersion::v2024_03_28()).await;
+        let kms = aws_sdk_kms::Client::new(&config);
+        (Some(config), Some(kms))
+    } else {
+        (None, None)
+    };
+
+    // Get source connection string either from S3 spec or direct argument
+    let source_connection_string = if let Some(s3_prefix) = &args.s3_prefix {
+        let spec: Spec = {
+            let spec_key = s3_prefix.append("/spec.json");
+            let s3_client = aws_sdk_s3::Client::new(aws_config.as_ref().unwrap());
+            let object = s3_client
+                .get_object()
+                .bucket(&spec_key.bucket)
+                .key(spec_key.key)
+                .send()
+                .await
+                .context("get spec from s3")?
+                .body
+                .collect()
+                .await
+                .context("download spec body")?;
+            serde_json::from_slice(&object.into_bytes()).context("parse spec as json")?
         };
-        postgres_initdb::do_run_initdb(postgres_initdb::RunInitdbArgs {
-            superuser: initdb_user,
-            locale: DEFAULT_LOCALE, // XXX: this shouldn't be hard-coded,
-            pg_version,
-            initdb_bin: self.pg_bin_dir.join("initdb").as_ref(),
-            library_search_path: &self.pg_lib_dir, // TODO: is this right? Prob works in compute image, not sure about neon_local.
-            pgdata: &self.pgdata_dir,
-        })
+
+        match spec.encryption_secret {
+            EncryptionSecret::KMS { key_id } => {
+                let mut output = kms_client
+                    .unwrap()
+                    .decrypt()
+                    .key_id(key_id)
+                    .ciphertext_blob(aws_sdk_s3::primitives::Blob::new(
+                        spec.source_connstring_ciphertext_base64,
+                    ))
+                    .send()
+                    .await
+                    .context("decrypt source connection string")?;
+                let plaintext = output
+                    .plaintext
+                    .take()
+                    .context("get plaintext source connection string")?;
+                String::from_utf8(plaintext.into_inner())
+                    .context("parse source connection string as utf8")?
+            }
+        }
+    } else {
+        args.source_connection_string.unwrap()
+    };
+
+    match tokio::fs::create_dir(&working_directory).await {
+        Ok(()) => {}
+        Err(e) if e.kind() == std::io::ErrorKind::AlreadyExists => {
+            if !is_directory_empty(&working_directory)
+                .await
+                .context("check if working directory is empty")?
+            {
+                anyhow::bail!("working directory is not empty");
+            } else {
+                // ok
+            }
+        }
+        Err(e) => return Err(anyhow::Error::new(e).context("create working directory")),
+    }
+
+    let pgdata_dir = working_directory.join("pgdata");
+    tokio::fs::create_dir(&pgdata_dir)
         .await
-        .context("initdb")
-    }
+        .context("create pgdata directory")?;
 
-    async fn start(
-        &mut self,
-        initdb_user: &str,
-        port: u16,
-        nproc: usize,
-        memory_mb: usize,
-    ) -> Result<&tokio::process::Child, anyhow::Error> {
-        self.prepare(initdb_user).await?;
+    let pgbin = pg_bin_dir.join("postgres");
+    let pg_version = match get_pg_version(pgbin.as_ref()) {
+        PostgresMajorVersion::V14 => 14,
+        PostgresMajorVersion::V15 => 15,
+        PostgresMajorVersion::V16 => 16,
+        PostgresMajorVersion::V17 => 17,
+    };
+    let superuser = "cloud_admin"; // XXX: this shouldn't be hard-coded
+    postgres_initdb::do_run_initdb(postgres_initdb::RunInitdbArgs {
+        superuser,
+        locale: DEFAULT_LOCALE, // XXX: this shouldn't be hard-coded,
+        pg_version,
+        initdb_bin: pg_bin_dir.join("initdb").as_ref(),
+        library_search_path: &pg_lib_dir, // TODO: is this right? Prob works in compute image, not sure about neon_local.
+        pgdata: &pgdata_dir,
+    })
+    .await
+    .context("initdb")?;
 
-        // Somewhat arbitrarily, use 10 % of memory for shared buffer cache, 70% for
-        // maintenance_work_mem (i.e. for sorting during index creation), and leave the rest
-        // available for misc other stuff that PostgreSQL uses memory for.
-        let shared_buffers_mb = ((memory_mb as f32) * 0.10) as usize;
-        let maintenance_work_mem_mb = ((memory_mb as f32) * 0.70) as usize;
+    // If the caller didn't specify CPU / RAM to use for sizing, default to
+    // number of CPUs in the system, and pretty arbitrarily, 256 MB of RAM.
+    let nproc = args.num_cpus.unwrap_or_else(num_cpus::get);
+    let memory_mb = args.memory_mb.unwrap_or(256);
 
-        //
-        // Launch postgres process
-        //
-        let mut proc = tokio::process::Command::new(&self.pgbin)
-            .arg("-D")
-            .arg(&self.pgdata_dir)
-            .args(["-p", &format!("{port}")])
-            .args(["-c", "wal_level=minimal"])
-            .args(["-c", &format!("shared_buffers={shared_buffers_mb}MB")])
-            .args(["-c", "max_wal_senders=0"])
-            .args(["-c", "fsync=off"])
-            .args(["-c", "full_page_writes=off"])
-            .args(["-c", "synchronous_commit=off"])
-            .args([
-                "-c",
-                &format!("maintenance_work_mem={maintenance_work_mem_mb}MB"),
-            ])
-            .args(["-c", &format!("max_parallel_maintenance_workers={nproc}")])
-            .args(["-c", &format!("max_parallel_workers={nproc}")])
-            .args(["-c", &format!("max_parallel_workers_per_gather={nproc}")])
-            .args(["-c", &format!("max_worker_processes={nproc}")])
-            .args(["-c", "effective_io_concurrency=100"])
-            .env_clear()
-            .env("LD_LIBRARY_PATH", &self.pg_lib_dir)
-            .env(
-                "ASAN_OPTIONS",
-                std::env::var("ASAN_OPTIONS").unwrap_or_default(),
-            )
-            .env(
-                "UBSAN_OPTIONS",
-                std::env::var("UBSAN_OPTIONS").unwrap_or_default(),
-            )
-            .stdout(std::process::Stdio::piped())
-            .stderr(std::process::Stdio::piped())
-            .spawn()
-            .context("spawn postgres")?;
+    // Somewhat arbitrarily, use 10 % of memory for shared buffer cache, 70% for
+    // maintenance_work_mem (i.e. for sorting during index creation), and leave the rest
+    // available for misc other stuff that PostgreSQL uses memory for.
+    let shared_buffers_mb = ((memory_mb as f32) * 0.10) as usize;
+    let maintenance_work_mem_mb = ((memory_mb as f32) * 0.70) as usize;
 
-        info!("spawned postgres, waiting for it to become ready");
-        tokio::spawn(
-            child_stdio_to_log::relay_process_output(proc.stdout.take(), proc.stderr.take())
-                .instrument(info_span!("postgres")),
-        );
-
-        self.postgres_proc = Some(proc);
-        Ok(self.postgres_proc.as_ref().unwrap())
-    }
-
-    async fn shutdown(&mut self) -> Result<(), anyhow::Error> {
-        let proc: &mut tokio::process::Child = self.postgres_proc.as_mut().unwrap();
-        info!("shutdown postgres");
-        nix::sys::signal::kill(
-            Pid::from_raw(i32::try_from(proc.id().unwrap()).expect("convert child pid to i32")),
-            nix::sys::signal::SIGTERM,
+    //
+    // Launch postgres process
+    //
+    let mut postgres_proc = tokio::process::Command::new(pgbin)
+        .arg("-D")
+        .arg(&pgdata_dir)
+        .args(["-p", &format!("{pg_port}")])
+        .args(["-c", "wal_level=minimal"])
+        .args(["-c", &format!("shared_buffers={shared_buffers_mb}MB")])
+        .args(["-c", "max_wal_senders=0"])
+        .args(["-c", "fsync=off"])
+        .args(["-c", "full_page_writes=off"])
+        .args(["-c", "synchronous_commit=off"])
+        .args([
+            "-c",
+            &format!("maintenance_work_mem={maintenance_work_mem_mb}MB"),
+        ])
+        .args(["-c", &format!("max_parallel_maintenance_workers={nproc}")])
+        .args(["-c", &format!("max_parallel_workers={nproc}")])
+        .args(["-c", &format!("max_parallel_workers_per_gather={nproc}")])
+        .args(["-c", &format!("max_worker_processes={nproc}")])
+        .args([
+            "-c",
+            &format!(
+                "effective_io_concurrency={}",
+                if cfg!(target_os = "macos") { 0 } else { 100 }
+            ),
+        ])
+        .env_clear()
+        .env("LD_LIBRARY_PATH", &pg_lib_dir)
+        .env(
+            "ASAN_OPTIONS",
+            std::env::var("ASAN_OPTIONS").unwrap_or_default(),
         )
-        .context("signal postgres to shut down")?;
-        proc.wait()
-            .await
-            .context("wait for postgres to shut down")
-            .map(|_| ())
-    }
-}
+        .env(
+            "UBSAN_OPTIONS",
+            std::env::var("UBSAN_OPTIONS").unwrap_or_default(),
+        )
+        .stdout(std::process::Stdio::piped())
+        .stderr(std::process::Stdio::piped())
+        .spawn()
+        .context("spawn postgres")?;
+
+    info!("spawned postgres, waiting for it to become ready");
+    tokio::spawn(
+        child_stdio_to_log::relay_process_output(
+            postgres_proc.stdout.take(),
+            postgres_proc.stderr.take(),
+        )
+        .instrument(info_span!("postgres")),
+    );
 
-async fn wait_until_ready(connstring: String, create_dbname: String) {
     // Create neondb database in the running postgres
+    let restore_pg_connstring =
+        format!("host=localhost port={pg_port} user={superuser} dbname=postgres");
+
     let start_time = std::time::Instant::now();
 
     loop {
@@ -276,12 +289,7 @@ async fn wait_until_ready(connstring: String, create_dbname: String) {
             std::process::exit(1);
         }
 
-        match tokio_postgres::connect(
-            &connstring.replace("dbname=neondb", "dbname=postgres"),
-            tokio_postgres::NoTls,
-        )
-        .await
-        {
+        match tokio_postgres::connect(&restore_pg_connstring, tokio_postgres::NoTls).await {
             Ok((client, connection)) => {
                 // Spawn the connection handling task to maintain the connection
                 tokio::spawn(async move {
@@ -290,12 +298,9 @@ async fn wait_until_ready(connstring: String, create_dbname: String) {
                     }
                 });
 
-                match client
-                    .simple_query(format!("CREATE DATABASE {create_dbname};").as_str())
-                    .await
-                {
+                match client.simple_query("CREATE DATABASE neondb;").await {
                     Ok(_) => {
-                        info!("created {} database", create_dbname);
+                        info!("created neondb database");
                         break;
                     }
                     Err(e) => {
@@ -319,16 +324,10 @@ async fn wait_until_ready(connstring: String, create_dbname: String) {
             }
         }
     }
-}
 
-async fn run_dump_restore(
-    workdir: Utf8PathBuf,
-    pg_bin_dir: Utf8PathBuf,
-    pg_lib_dir: Utf8PathBuf,
-    source_connstring: String,
-    destination_connstring: String,
-) -> Result<(), anyhow::Error> {
-    let dumpdir = workdir.join("dumpdir");
+    let restore_pg_connstring = restore_pg_connstring.replace("dbname=postgres", "dbname=neondb");
+
+    let dumpdir = working_directory.join("dumpdir");
 
     let common_args = [
         // schema mapping (prob suffices to specify them on one side)
@@ -357,7 +356,7 @@ async fn run_dump_restore(
             .arg("--no-sync")
             // POSITIONAL args
             // source db (db name included in connection string)
-            .arg(&source_connstring)
+            .arg(&source_connection_string)
             // how we run it
             .env_clear()
             .env("LD_LIBRARY_PATH", &pg_lib_dir)
@@ -377,18 +376,19 @@ async fn run_dump_restore(
         let st = pg_dump.wait().await.context("wait for pg_dump")?;
         info!(status=?st, "pg_dump exited");
         if !st.success() {
-            error!(status=%st, "pg_dump failed, restore will likely fail as well");
-            bail!("pg_dump failed");
+            warn!(status=%st, "pg_dump failed, restore will likely fail as well");
         }
     }
 
-    // TODO: maybe do it in a streaming way, plenty of internal research done on this already
+    // TODO: do it in a streaming way, plenty of internal research done on this already
     // TODO: do the unlogged table trick
+
+    info!("restore from working directory into vanilla postgres");
     {
         let mut pg_restore = tokio::process::Command::new(pg_bin_dir.join("pg_restore"))
             .args(&common_args)
             .arg("-d")
-            .arg(&destination_connstring)
+            .arg(&restore_pg_connstring)
             // POSITIONAL args
             .arg(&dumpdir)
             // how we run it
@@ -411,259 +411,48 @@ async fn run_dump_restore(
         let st = pg_restore.wait().await.context("wait for pg_restore")?;
         info!(status=?st, "pg_restore exited");
         if !st.success() {
-            error!(status=%st, "pg_restore failed, restore will likely fail as well");
-            bail!("pg_restore failed");
+            warn!(status=%st, "pg_restore failed, restore will likely fail as well");
         }
     }
 
-    Ok(())
-}
-
-#[allow(clippy::too_many_arguments)]
-async fn cmd_pgdata(
-    s3_client: Option<aws_sdk_s3::Client>,
-    kms_client: Option<aws_sdk_kms::Client>,
-    maybe_s3_prefix: Option<s3_uri::S3Uri>,
-    maybe_spec: Option<Spec>,
-    source_connection_string: Option<String>,
-    interactive: bool,
-    pg_port: u16,
-    workdir: Utf8PathBuf,
-    pg_bin_dir: Utf8PathBuf,
-    pg_lib_dir: Utf8PathBuf,
-    num_cpus: Option<usize>,
-    memory_mb: Option<usize>,
-) -> Result<(), anyhow::Error> {
-    if maybe_spec.is_none() && source_connection_string.is_none() {
-        bail!("spec must be provided for pgdata command");
-    }
-    if maybe_spec.is_some() && source_connection_string.is_some() {
-        bail!("only one of spec or source_connection_string can be provided");
-    }
-
-    let source_connection_string = if let Some(spec) = maybe_spec {
-        match spec.encryption_secret {
-            EncryptionSecret::KMS { key_id } => {
-                decode_connstring(
-                    kms_client.as_ref().unwrap(),
-                    &key_id,
-                    spec.source_connstring_ciphertext_base64,
-                )
-                .await?
-            }
-        }
-    } else {
-        source_connection_string.unwrap()
-    };
-
-    let superuser = "cloud_admin";
-    let destination_connstring = format!(
-        "host=localhost port={} user={} dbname=neondb",
-        pg_port, superuser
-    );
-
-    let pgdata_dir = workdir.join("pgdata");
-    let mut proc = PostgresProcess::new(pgdata_dir.clone(), pg_bin_dir.clone(), pg_lib_dir.clone());
-    let nproc = num_cpus.unwrap_or_else(num_cpus::get);
-    let memory_mb = memory_mb.unwrap_or(256);
-    proc.start(superuser, pg_port, nproc, memory_mb).await?;
-    wait_until_ready(destination_connstring.clone(), "neondb".to_string()).await;
-
-    run_dump_restore(
-        workdir.clone(),
-        pg_bin_dir,
-        pg_lib_dir,
-        source_connection_string,
-        destination_connstring,
-    )
-    .await?;
-
     // If interactive mode, wait for Ctrl+C
-    if interactive {
+    if args.interactive {
         info!("Running in interactive mode. Press Ctrl+C to shut down.");
         tokio::signal::ctrl_c().await.context("wait for ctrl-c")?;
     }
 
-    proc.shutdown().await?;
+    info!("shutdown postgres");
+    {
+        nix::sys::signal::kill(
+            Pid::from_raw(
+                i32::try_from(postgres_proc.id().unwrap()).expect("convert child pid to i32"),
+            ),
+            nix::sys::signal::SIGTERM,
+        )
+        .context("signal postgres to shut down")?;
+        postgres_proc
+            .wait()
+            .await
+            .context("wait for postgres to shut down")?;
+    }
 
     // Only sync if s3_prefix was specified
-    if let Some(s3_prefix) = maybe_s3_prefix {
+    if let Some(s3_prefix) = args.s3_prefix {
         info!("upload pgdata");
-        aws_s3_sync::upload_dir_recursive(
-            s3_client.as_ref().unwrap(),
-            Utf8Path::new(&pgdata_dir),
-            &s3_prefix.append("/pgdata/"),
-        )
-        .await
-        .context("sync dump directory to destination")?;
+        aws_s3_sync::sync(Utf8Path::new(&pgdata_dir), &s3_prefix.append("/pgdata/"))
+            .await
+            .context("sync dump directory to destination")?;
 
         info!("write status");
         {
-            let status_dir = workdir.join("status");
+            let status_dir = working_directory.join("status");
             std::fs::create_dir(&status_dir).context("create status directory")?;
             let status_file = status_dir.join("pgdata");
             std::fs::write(&status_file, serde_json::json!({"done": true}).to_string())
                 .context("write status file")?;
-            aws_s3_sync::upload_dir_recursive(
-                s3_client.as_ref().unwrap(),
-                &status_dir,
-                &s3_prefix.append("/status/"),
-            )
-            .await
-            .context("sync status directory to destination")?;
-        }
-    }
-
-    Ok(())
-}
-
-async fn cmd_dumprestore(
-    kms_client: Option<aws_sdk_kms::Client>,
-    maybe_spec: Option<Spec>,
-    source_connection_string: Option<String>,
-    destination_connection_string: Option<String>,
-    workdir: Utf8PathBuf,
-    pg_bin_dir: Utf8PathBuf,
-    pg_lib_dir: Utf8PathBuf,
-) -> Result<(), anyhow::Error> {
-    let (source_connstring, destination_connstring) = if let Some(spec) = maybe_spec {
-        match spec.encryption_secret {
-            EncryptionSecret::KMS { key_id } => {
-                let source = decode_connstring(
-                    kms_client.as_ref().unwrap(),
-                    &key_id,
-                    spec.source_connstring_ciphertext_base64,
-                )
-                .await?;
-
-                let dest = if let Some(dest_ciphertext) =
-                    spec.destination_connstring_ciphertext_base64
-                {
-                    decode_connstring(kms_client.as_ref().unwrap(), &key_id, dest_ciphertext)
-                        .await?
-                } else {
-                    bail!("destination connection string must be provided in spec for dump_restore command");
-                };
-
-                (source, dest)
-            }
-        }
-    } else {
-        (
-            source_connection_string.unwrap(),
-            if let Some(val) = destination_connection_string {
-                val
-            } else {
-                bail!("destination connection string must be provided for dump_restore command");
-            },
-        )
-    };
-
-    run_dump_restore(
-        workdir,
-        pg_bin_dir,
-        pg_lib_dir,
-        source_connstring,
-        destination_connstring,
-    )
-    .await
-}
-
-#[tokio::main]
-pub(crate) async fn main() -> anyhow::Result<()> {
-    utils::logging::init(
-        utils::logging::LogFormat::Json,
-        utils::logging::TracingErrorLayerEnablement::EnableWithRustLogFilter,
-        utils::logging::Output::Stdout,
-    )?;
-
-    info!("starting");
-
-    let args = Args::parse();
-
-    // Initialize AWS clients only if s3_prefix is specified
-    let (s3_client, kms_client) = if args.s3_prefix.is_some() {
-        let config = aws_config::load_defaults(BehaviorVersion::v2024_03_28()).await;
-        let s3_client = aws_sdk_s3::Client::new(&config);
-        let kms = aws_sdk_kms::Client::new(&config);
-        (Some(s3_client), Some(kms))
-    } else {
-        (None, None)
-    };
-
-    let spec: Option<Spec> = if let Some(s3_prefix) = &args.s3_prefix {
-        let spec_key = s3_prefix.append("/spec.json");
-        let object = s3_client
-            .as_ref()
-            .unwrap()
-            .get_object()
-            .bucket(&spec_key.bucket)
-            .key(spec_key.key)
-            .send()
-            .await
-            .context("get spec from s3")?
-            .body
-            .collect()
-            .await
-            .context("download spec body")?;
-        serde_json::from_slice(&object.into_bytes()).context("parse spec as json")?
-    } else {
-        None
-    };
-
-    match tokio::fs::create_dir(&args.working_directory).await {
-        Ok(()) => {}
-        Err(e) if e.kind() == std::io::ErrorKind::AlreadyExists => {
-            if !is_directory_empty(&args.working_directory)
+            aws_s3_sync::sync(&status_dir, &s3_prefix.append("/status/"))
                 .await
-                .context("check if working directory is empty")?
-            {
-                bail!("working directory is not empty");
-            } else {
-                // ok
-            }
-        }
-        Err(e) => return Err(anyhow::Error::new(e).context("create working directory")),
-    }
-
-    match args.command {
-        Command::Pgdata {
-            source_connection_string,
-            interactive,
-            pg_port,
-            num_cpus,
-            memory_mb,
-        } => {
-            cmd_pgdata(
-                s3_client,
-                kms_client,
-                args.s3_prefix,
-                spec,
-                source_connection_string,
-                interactive,
-                pg_port,
-                args.working_directory,
-                args.pg_bin_dir,
-                args.pg_lib_dir,
-                num_cpus,
-                memory_mb,
-            )
-            .await?;
-        }
-        Command::DumpRestore {
-            source_connection_string,
-            destination_connection_string,
-        } => {
-            cmd_dumprestore(
-                kms_client,
-                spec,
-                source_connection_string,
-                destination_connection_string,
-                args.working_directory,
-                args.pg_bin_dir,
-                args.pg_lib_dir,
-            )
-            .await?;
+                .context("sync status directory to destination")?;
         }
     }
 

compute_tools/src/bin/fast_import/aws_s3_sync.rs

@@ -1,102 +1,24 @@
-use camino::{Utf8Path, Utf8PathBuf};
-use tokio::task::JoinSet;
-use walkdir::WalkDir;
+use anyhow::Context;
+use camino::Utf8Path;
 
 use super::s3_uri::S3Uri;
 
-use tracing::{info, warn};
-
-const MAX_PARALLEL_UPLOADS: usize = 10;
-
-/// Upload all files from 'local' to 'remote'
-pub(crate) async fn upload_dir_recursive(
-    s3_client: &aws_sdk_s3::Client,
-    local: &Utf8Path,
-    remote: &S3Uri,
-) -> anyhow::Result<()> {
-    // Recursively scan directory
-    let mut dirwalker = WalkDir::new(local)
-        .into_iter()
-        .map(|entry| {
-            let entry = entry?;
-            let file_type = entry.file_type();
-            let path = <&Utf8Path>::try_from(entry.path())?.to_path_buf();
-            Ok((file_type, path))
-        })
-        .filter_map(|e: anyhow::Result<(std::fs::FileType, Utf8PathBuf)>| {
-            match e {
-                Ok((file_type, path)) if file_type.is_file() => Some(Ok(path)),
-                Ok((file_type, _path)) if file_type.is_dir() => {
-                    // The WalkDir iterator will recurse into directories, but we don't want
-                    // to do anything with directories as such. There's no concept of uploading
-                    // an empty directory to S3.
-                    None
-                }
-                Ok((file_type, path)) if file_type.is_symlink() => {
-                    // huh, didn't expect a symlink. Can't upload that to S3. Warn and skip.
-                    warn!("cannot upload symlink ({})", path);
-                    None
-                }
-                Ok((_file_type, path)) => {
-                    // should not happen
-                    warn!("directory entry has unexpected type ({})", path);
-                    None
-                }
-                Err(e) => Some(Err(e)),
-            }
-        });
-
-    // Spawn upload tasks for each file, keeping MAX_PARALLEL_UPLOADS active in
-    // parallel.
-    let mut joinset = JoinSet::new();
-    loop {
-        // Could we upload more?
-        while joinset.len() < MAX_PARALLEL_UPLOADS {
-            if let Some(full_local_path) = dirwalker.next() {
-                let full_local_path = full_local_path?;
-                let relative_local_path = full_local_path
-                    .strip_prefix(local)
-                    .expect("all paths start from the walkdir root");
-                let remote_path = remote.append(relative_local_path.as_str());
-                info!(
-                    "starting upload of {} to {}",
-                    &full_local_path, &remote_path
-                );
-                let upload_task = upload_file(s3_client.clone(), full_local_path, remote_path);
-                joinset.spawn(upload_task);
-            } else {
-                info!("draining upload tasks");
-                break;
-            }
-        }
-
-        // Wait for an upload to complete
-        if let Some(res) = joinset.join_next().await {
-            let _ = res?;
-        } else {
-            // all done!
-            break;
-        }
+pub(crate) async fn sync(local: &Utf8Path, remote: &S3Uri) -> anyhow::Result<()> {
+    let mut builder = tokio::process::Command::new("aws");
+    builder
+        .arg("s3")
+        .arg("sync")
+        .arg(local.as_str())
+        .arg(remote.to_string());
+    let st = builder
+        .spawn()
+        .context("spawn aws s3 sync")?
+        .wait()
+        .await
+        .context("wait for aws s3 sync")?;
+    if st.success() {
+        Ok(())
+    } else {
+        Err(anyhow::anyhow!("aws s3 sync failed"))
     }
-    Ok(())
-}
-
-pub(crate) async fn upload_file(
-    s3_client: aws_sdk_s3::Client,
-    local_path: Utf8PathBuf,
-    remote: S3Uri,
-) -> anyhow::Result<()> {
-    use aws_smithy_types::byte_stream::ByteStream;
-    let stream = ByteStream::from_path(&local_path).await?;
-
-    let _result = s3_client
-        .put_object()
-        .bucket(remote.bucket)
-        .key(&remote.key)
-        .body(stream)
-        .send()
-        .await?;
-    info!("upload of {} to {} finished", &local_path, &remote.key);
-
-    Ok(())
 }

compute_tools/src/http/server.rs

@@ -7,7 +7,6 @@ use std::{
 
 use anyhow::Result;
 use axum::{
-    body::Body,
     extract::Request,
     middleware::{self, Next},
     response::{IntoResponse, Response},
@@ -17,7 +16,6 @@ use axum::{
 use http::StatusCode;
 use tokio::net::TcpListener;
 use tower::ServiceBuilder;
-use tower_http::classify::ServerErrorsFailureClass;
 use tower_http::{request_id::PropagateRequestIdLayer, trace::TraceLayer};
 use tracing::{debug, error, info, Span};
 use uuid::Uuid;
@@ -86,85 +84,46 @@ impl From<Server> for Router<Arc<ComputeNode>> {
                 .route("/terminate", post(terminate::terminate)),
         };
 
-        router
-            .fallback(Server::handle_404)
-            .method_not_allowed_fallback(Server::handle_405)
-            .layer(
-                ServiceBuilder::new()
-                    // Add this middleware since we assume the request ID exists
-                    .layer(middleware::from_fn(maybe_add_request_id_header))
-                    .layer(
-                        TraceLayer::new_for_http()
-                            .make_span_with(|request: &Request<Body>| {
-                                let request_id = request
+        router.fallback(Server::handle_404).method_not_allowed_fallback(Server::handle_405).layer(
+            ServiceBuilder::new()
+                // Add this middleware since we assume the request ID exists
+                .layer(middleware::from_fn(maybe_add_request_id_header))
+                .layer(
+                    TraceLayer::new_for_http()
+                        .on_request(|request: &http::Request<_>, _span: &Span| {
+                            let request_id = request
+                                .headers()
+                                .get(X_REQUEST_ID)
+                                .unwrap()
+                                .to_str()
+                                .unwrap();
+
+                            match request.uri().path() {
+                                "/metrics" => {
+                                    debug!(%request_id, "{} {}", request.method(), request.uri())
+                                }
+                                _ => info!(%request_id, "{} {}", request.method(), request.uri()),
+                            };
+                        })
+                        .on_response(
+                            |response: &http::Response<_>, latency: Duration, _span: &Span| {
+                                let request_id = response
                                     .headers()
                                     .get(X_REQUEST_ID)
                                     .unwrap()
                                     .to_str()
                                     .unwrap();
 
-                                match request.uri().path() {
-                                    "/metrics" => {
-                                        tracing::span!(
-                                            tracing::Level::DEBUG,
-                                            "",
-                                            method = tracing::field::display(request.method()),
-                                            uri = tracing::field::display(request.uri()),
-                                            request_id = tracing::field::display(request_id)
-                                        )
-                                    }
-                                    _ => tracing::span!(
-                                        tracing::Level::INFO,
-                                        "",
-                                        method = tracing::field::display(request.method()),
-                                        uri = tracing::field::display(request.uri()),
-                                        request_id = tracing::field::display(request_id)
-                                    ),
-                                }
-                            })
-                            .on_request(|request: &http::Request<_>, _span: &Span| {
-                                match request.uri().path() {
-                                    "/metrics" => debug!("incoming request"),
-                                    _ => info!("incoming request"),
-                                };
-                            })
-                            .on_response(
-                                |response: &http::Response<_>, latency: Duration, _span: &Span| {
-                                    // All errors will be logged in the on_failure handler
-                                    if let 200..=399 = response.status().as_u16() {
-                                        info!(
-                                            message = "request finished",
-                                            code = %response.status().as_u16(),
-                                            latency_ms = %latency.as_millis()
-                                        )
-                                    }
-                                },
-                            )
-                            .on_failure(
-                                |error: ServerErrorsFailureClass,
-                                 latency: Duration,
-                                 _span: &Span| {
-                                    match error {
-                                        ServerErrorsFailureClass::StatusCode(code) => {
-                                            error!(
-                                                message = "request failed",
-                                                code = %code,
-                                                latency_ms = %latency.as_millis()
-                                            );
-                                        }
-                                        ServerErrorsFailureClass::Error(error) => {
-                                            error!(
-                                                message = "request failed unexpectedly",
-                                                error = %error,
-                                                latency_ms = %latency.as_millis()
-                                            );
-                                        }
-                                    }
-                                },
-                            ),
-                    )
-                    .layer(PropagateRequestIdLayer::x_request_id()),
-            )
+                                info!(
+                                    %request_id,
+                                    code = response.status().as_u16(),
+                                    latency = latency.as_millis()
+                                )
+                            },
+                        ),
+                )
+                .layer(PropagateRequestIdLayer::x_request_id()),
+        )
     }
 }
 

compute_tools/src/spec.rs

@@ -11,9 +11,7 @@ use crate::migration::MigrationRunner;
 use crate::params::PG_HBA_ALL_MD5;
 use crate::pg_helpers::*;
 
-use compute_api::responses::{
-    ComputeCtlConfig, ControlPlaneComputeStatus, ControlPlaneSpecResponse,
-};
+use compute_api::responses::{ControlPlaneComputeStatus, ControlPlaneSpecResponse};
 use compute_api::spec::ComputeSpec;
 
 // Do control plane request and return response if any. In case of error it
@@ -75,13 +73,14 @@ fn do_control_plane_request(
 pub fn get_spec_from_control_plane(
     base_uri: &str,
     compute_id: &str,
-) -> Result<(Option<ComputeSpec>, ComputeCtlConfig)> {
+) -> Result<Option<ComputeSpec>> {
     let cp_uri = format!("{base_uri}/compute/api/v2/computes/{compute_id}/spec");
     let jwt: String = match std::env::var("NEON_CONTROL_PLANE_TOKEN") {
         Ok(v) => v,
         Err(_) => "".to_string(),
     };
     let mut attempt = 1;
+    let mut spec: Result<Option<ComputeSpec>> = Ok(None);
 
     info!("getting spec from control plane: {}", cp_uri);
 
@@ -91,7 +90,7 @@ pub fn get_spec_from_control_plane(
     // - no spec for compute yet (Empty state) -> return Ok(None)
     // - got spec -> return Ok(Some(spec))
     while attempt < 4 {
-        let result = match do_control_plane_request(&cp_uri, &jwt) {
+        spec = match do_control_plane_request(&cp_uri, &jwt) {
             Ok(spec_resp) => {
                 CPLANE_REQUESTS_TOTAL
                     .with_label_values(&[
@@ -100,10 +99,10 @@ pub fn get_spec_from_control_plane(
                     ])
                     .inc();
                 match spec_resp.status {
-                    ControlPlaneComputeStatus::Empty => Ok((None, spec_resp.compute_ctl_config)),
+                    ControlPlaneComputeStatus::Empty => Ok(None),
                     ControlPlaneComputeStatus::Attached => {
                         if let Some(spec) = spec_resp.spec {
-                            Ok((Some(spec), spec_resp.compute_ctl_config))
+                            Ok(Some(spec))
                         } else {
                             bail!("compute is attached, but spec is empty")
                         }
@@ -122,10 +121,10 @@ pub fn get_spec_from_control_plane(
             }
         };
 
-        if let Err(e) = &result {
+        if let Err(e) = &spec {
             error!("attempt {} to get spec failed with: {}", attempt, e);
         } else {
-            return result;
+            return spec;
         }
 
         attempt += 1;
@@ -133,9 +132,7 @@ pub fn get_spec_from_control_plane(
     }
 
     // All attempts failed, return error.
-    Err(anyhow::anyhow!(
-        "Exhausted all attempts to retrieve the spec from the control plane"
-    ))
+    spec
 }
 
 /// Check `pg_hba.conf` and update if needed to allow external connections.

control_plane/src/endpoint.rs

@@ -48,8 +48,6 @@ use std::sync::Arc;
 use std::time::Duration;
 
 use anyhow::{anyhow, bail, Context, Result};
-use compute_api::requests::ConfigurationRequest;
-use compute_api::responses::ComputeCtlConfig;
 use compute_api::spec::Database;
 use compute_api::spec::PgIdent;
 use compute_api::spec::RemoteExtSpec;
@@ -882,13 +880,10 @@ impl Endpoint {
                 self.external_http_address.port()
             ))
             .header(CONTENT_TYPE.as_str(), "application/json")
-            .body(
-                serde_json::to_string(&ConfigurationRequest {
-                    spec,
-                    compute_ctl_config: ComputeCtlConfig::default(),
-                })
-                .unwrap(),
-            )
+            .body(format!(
+                "{{\"spec\":{}}}",
+                serde_json::to_string_pretty(&spec)?
+            ))
             .send()
             .await?;
 

control_plane/src/storage_controller.rs

@@ -838,10 +838,7 @@ impl StorageController {
         self.dispatch(
             Method::PUT,
             format!("control/v1/tenant/{tenant_shard_id}/migrate"),
-            Some(TenantShardMigrateRequest {
-                node_id,
-                migration_config: None,
-            }),
+            Some(TenantShardMigrateRequest { node_id }),
         )
         .await
     }

control_plane/storcon_cli/src/main.rs

@@ -609,10 +609,7 @@ async fn main() -> anyhow::Result<()> {
             tenant_shard_id,
             node,
         } => {
-            let req = TenantShardMigrateRequest {
-                node_id: node,
-                migration_config: None,
-            };
+            let req = TenantShardMigrateRequest { node_id: node };
 
             storcon_client
                 .dispatch::<TenantShardMigrateRequest, TenantShardMigrateResponse>(
@@ -626,10 +623,7 @@ async fn main() -> anyhow::Result<()> {
             tenant_shard_id,
             node,
         } => {
-            let req = TenantShardMigrateRequest {
-                node_id: node,
-                migration_config: None,
-            };
+            let req = TenantShardMigrateRequest { node_id: node };
 
             storcon_client
                 .dispatch::<TenantShardMigrateRequest, TenantShardMigrateResponse>(
@@ -1088,10 +1082,7 @@ async fn main() -> anyhow::Result<()> {
                             .dispatch::<TenantShardMigrateRequest, TenantShardMigrateResponse>(
                                 Method::PUT,
                                 format!("control/v1/tenant/{}/migrate", mv.tenant_shard_id),
-                                Some(TenantShardMigrateRequest {
-                                    node_id: mv.to,
-                                    migration_config: None,
-                                }),
+                                Some(TenantShardMigrateRequest { node_id: mv.to }),
                             )
                             .await
                             .map_err(|e| (mv.tenant_shard_id, mv.from, mv.to, e))

docker-compose/docker_compose_test.sh

@@ -71,7 +71,7 @@ for pg_version in ${TEST_VERSION_ONLY-14 15 16 17}; do
         cat ../compute/patches/contrib_pg${pg_version}.patch | docker exec -i $TEST_CONTAINER_NAME bash -c "(cd /postgres && patch -p1)"
         # We are running tests now
         rm -f testout.txt testout_contrib.txt
-        docker exec -e USE_PGXS=1 -e SKIP=timescaledb-src,rdkit-src,postgis-src,pgx_ulid-src,pg_tiktoken-src,pg_jsonschema-src,kq_imcx-src,wal2json_2_5-src \
+        docker exec -e USE_PGXS=1 -e SKIP=timescaledb-src,rdkit-src,postgis-src,pgx_ulid-src,pgtap-src,pg_tiktoken-src,pg_jsonschema-src,kq_imcx-src,wal2json_2_5-src \
         $TEST_CONTAINER_NAME /run-tests.sh /ext-src | tee testout.txt && EXT_SUCCESS=1 || EXT_SUCCESS=0
         docker exec -e SKIP=start-scripts,postgres_fdw,ltree_plpython,jsonb_plpython,jsonb_plperl,hstore_plpython,hstore_plperl,dblink,bool_plperl \
         $TEST_CONTAINER_NAME /run-tests.sh /postgres/contrib | tee testout_contrib.txt && CONTRIB_SUCCESS=1 || CONTRIB_SUCCESS=0

docker-compose/ext-src/pgtap-src/test-upgrade.patch

@@ -1,15 +0,0 @@
-diff --git a/test/schedule/create.sql b/test/schedule/create.sql
-index ba355ed..7e250f5 100644
---- a/test/schedule/create.sql
-+++ b/test/schedule/create.sql
-@@ -1,3 +1,2 @@
- \unset ECHO
- \i test/psql.sql
--CREATE EXTENSION pgtap;
-diff --git a/test/schedule/main.sch b/test/schedule/main.sch
-index a8a5fbc..0463fc4 100644
---- a/test/schedule/main.sch
-+++ b/test/schedule/main.sch
-@@ -1,2 +1 @@
--test: build
- test: create

docker-compose/ext-src/pgtap-src/test-upgrade.sh

@@ -1,6 +0,0 @@
-#!/bin/sh
-set -ex
-cd "$(dirname ${0})"
-patch -p1 <test-upgrade.patch
-PG_REGRESS=$(dirname "$(pg_config --pgxs)")/../test/regress/pg_regress
-${PG_REGRESS} --inputdir=./ --bindir='/usr/local/pgsql/bin'    --inputdir=test --max-connections=86 --schedule test/schedule/main.sch   --schedule test/build/run.sch --dbname contrib_regression --use-existing

docker-compose/test_extensions_upgrade.sh

@@ -11,7 +11,6 @@ if [ -z ${OLDTAG+x} ] || [ -z ${NEWTAG+x} ] || [ -z "${OLDTAG}" ] || [ -z "${NEW
   exit 1
 fi
 export PG_VERSION=${PG_VERSION:-16}
-export PG_TEST_VERSION=${PG_VERSION}
 function wait_for_ready {
   TIME=0
   while ! docker compose logs compute_is_ready | grep -q "accepting connections" && [ ${TIME} -le 300 ] ; do
@@ -42,8 +41,7 @@ EXTENSIONS='[
 {"extname": "roaringbitmap", "extdir": "pg_roaringbitmap-src"},
 {"extname": "semver", "extdir": "pg_semver-src"},
 {"extname": "pg_ivm", "extdir": "pg_ivm-src"},
-{"extname": "pgjwt", "extdir": "pgjwt-src"},
-{"extname": "pgtap", "extdir": "pgtap-src"}
+{"extname": "pgjwt", "extdir": "pgjwt-src"}
 ]'
 EXTNAMES=$(echo ${EXTENSIONS} | jq -r '.[].extname' | paste -sd ' ' -)
 TAG=${NEWTAG} docker compose --profile test-extensions up --quiet-pull --build -d
@@ -60,12 +58,8 @@ docker compose cp  ext-src neon-test-extensions:/
 docker compose exec neon-test-extensions psql -c "DROP DATABASE IF EXISTS contrib_regression"
 docker compose exec neon-test-extensions psql -c "CREATE DATABASE contrib_regression"
 create_extensions "${EXTNAMES}"
-if [ "${FORCE_ALL_UPGRADE_TESTS:-false}" = true ]; then
-  exts="${EXTNAMES}"
-else
-  query="select pge.extname from pg_extension pge join (select key as extname, value as extversion from json_each_text('${new_vers}')) x on pge.extname=x.extname and pge.extversion <> x.extversion"
-  exts=$(docker compose exec neon-test-extensions psql -Aqt -d contrib_regression -c "$query")
-fi
+query="select pge.extname from pg_extension pge join (select key as extname, value as extversion from json_each_text('${new_vers}')) x on pge.extname=x.extname and pge.extversion <> x.extversion"
+exts=$(docker compose exec neon-test-extensions psql -Aqt -d contrib_regression -c "$query")
 if [ -z "${exts}" ]; then
   echo "No extensions were upgraded"
 else
@@ -93,10 +87,7 @@ else
       exit 1
     fi
     docker compose exec neon-test-extensions psql -d contrib_regression -c "\dx ${ext}"
-    if ! docker compose exec neon-test-extensions sh -c /ext-src/${EXTDIR}/test-upgrade.sh; then
-      docker  compose exec neon-test-extensions  cat /ext-src/${EXTDIR}/regression.diffs
-      exit 1
-    fi
+    docker compose exec neon-test-extensions sh -c /ext-src/${EXTDIR}/test-upgrade.sh
     docker compose exec neon-test-extensions psql -d contrib_regression -c "alter extension ${ext} update"
     docker compose exec neon-test-extensions psql -d contrib_regression -c "\dx ${ext}"
   done

libs/compute_api/Cargo.toml

@@ -7,7 +7,6 @@ license.workspace = true
 [dependencies]
 anyhow.workspace = true
 chrono.workspace = true
-jsonwebtoken.workspace = true
 serde.workspace = true
 serde_json.workspace = true
 regex.workspace = true

libs/compute_api/src/requests.rs

@@ -1,20 +1,18 @@
 //! Structs representing the JSON formats used in the compute_ctl's HTTP API.
 use crate::{
     privilege::Privilege,
-    responses::ComputeCtlConfig,
     spec::{ComputeSpec, ExtVersion, PgIdent},
 };
-use serde::{Deserialize, Serialize};
+use serde::Deserialize;
 
 /// Request of the /configure API
 ///
 /// We now pass only `spec` in the configuration request, but later we can
 /// extend it and something like `restart: bool` or something else. So put
 /// `spec` into a struct initially to be more flexible in the future.
-#[derive(Debug, Deserialize, Serialize)]
+#[derive(Deserialize, Debug)]
 pub struct ConfigurationRequest {
     pub spec: ComputeSpec,
-    pub compute_ctl_config: ComputeCtlConfig,
 }
 
 #[derive(Deserialize, Debug)]

libs/compute_api/src/responses.rs

@@ -3,7 +3,6 @@
 use std::fmt::Display;
 
 use chrono::{DateTime, Utc};
-use jsonwebtoken::jwk::JwkSet;
 use serde::{Deserialize, Serialize, Serializer};
 
 use crate::{
@@ -136,27 +135,13 @@ pub struct CatalogObjects {
     pub databases: Vec<Database>,
 }
 
-#[derive(Debug, Deserialize, Serialize)]
-pub struct ComputeCtlConfig {
-    pub jwks: JwkSet,
-}
-
-impl Default for ComputeCtlConfig {
-    fn default() -> Self {
-        Self {
-            jwks: JwkSet {
-                keys: Vec::default(),
-            },
-        }
-    }
-}
-
 /// Response of the `/computes/{compute_id}/spec` control-plane API.
+/// This is not actually a compute API response, so consider moving
+/// to a different place.
 #[derive(Deserialize, Debug)]
 pub struct ControlPlaneSpecResponse {
     pub spec: Option<ComputeSpec>,
     pub status: ControlPlaneComputeStatus,
-    pub compute_ctl_config: ComputeCtlConfig,
 }
 
 #[derive(Deserialize, Clone, Copy, Debug, PartialEq, Eq)]

libs/pageserver_api/src/config.rs

@@ -351,7 +351,7 @@ pub struct TenantConfigToml {
 
     /// Enable rel_size_v2 for this tenant. Once enabled, the tenant will persist this information into
     /// `index_part.json`, and it cannot be reversed.
-    pub rel_size_v2_enabled: bool,
+    pub rel_size_v2_enabled: Option<bool>,
 
     // gc-compaction related configs
     /// Enable automatic gc-compaction trigger on this tenant.
@@ -633,7 +633,7 @@ impl Default for TenantConfigToml {
             lsn_lease_length_for_ts: LsnLease::DEFAULT_LENGTH_FOR_TS,
             timeline_offloading: true,
             wal_receiver_protocol_override: None,
-            rel_size_v2_enabled: false,
+            rel_size_v2_enabled: None,
             gc_compaction_enabled: DEFAULT_GC_COMPACTION_ENABLED,
             gc_compaction_initial_threshold_kb: DEFAULT_GC_COMPACTION_INITIAL_THRESHOLD_KB,
             gc_compaction_ratio_percent: DEFAULT_GC_COMPACTION_RATIO_PERCENT,

libs/pageserver_api/src/controller_api.rs

@@ -182,18 +182,6 @@ pub struct TenantDescribeResponseShard {
 #[derive(Serialize, Deserialize, Debug)]
 pub struct TenantShardMigrateRequest {
     pub node_id: NodeId,
-    #[serde(default)]
-    pub migration_config: Option<MigrationConfig>,
-}
-
-#[derive(Serialize, Deserialize, Debug)]
-pub struct MigrationConfig {
-    #[serde(default)]
-    #[serde(with = "humantime_serde")]
-    pub secondary_warmup_timeout: Option<Duration>,
-    #[serde(default)]
-    #[serde(with = "humantime_serde")]
-    pub secondary_download_request_timeout: Option<Duration>,
 }
 
 #[derive(Serialize, Clone, Debug)]

libs/pageserver_api/src/key.rs

@@ -1,12 +1,10 @@
 use anyhow::{bail, Result};
 use byteorder::{ByteOrder, BE};
-use bytes::Bytes;
 use postgres_ffi::relfile_utils::{FSM_FORKNUM, VISIBILITYMAP_FORKNUM};
 use postgres_ffi::Oid;
 use postgres_ffi::RepOriginId;
 use serde::{Deserialize, Serialize};
 use std::{fmt, ops::Range};
-use utils::const_assert;
 
 use crate::reltag::{BlockNumber, RelTag, SlruKind};
 
@@ -51,64 +49,6 @@ pub const AUX_KEY_PREFIX: u8 = 0x62;
 /// The key prefix of ReplOrigin keys.
 pub const REPL_ORIGIN_KEY_PREFIX: u8 = 0x63;
 
-/// The key prefix of db directory keys.
-pub const DB_DIR_KEY_PREFIX: u8 = 0x64;
-
-/// The key prefix of rel directory keys.
-pub const REL_DIR_KEY_PREFIX: u8 = 0x65;
-
-#[derive(Debug, Clone, Copy, Hash, PartialEq, Eq)]
-pub enum RelDirExists {
-    Exists,
-    Removed,
-}
-
-#[derive(Debug)]
-pub struct DecodeError;
-
-impl fmt::Display for DecodeError {
-    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
-        write!(f, "invalid marker")
-    }
-}
-
-impl std::error::Error for DecodeError {}
-
-impl RelDirExists {
-    /// The value of the rel directory keys that indicates the existence of a relation.
-    const REL_EXISTS_MARKER: Bytes = Bytes::from_static(b"r");
-
-    pub fn encode(&self) -> Bytes {
-        match self {
-            Self::Exists => Self::REL_EXISTS_MARKER.clone(),
-            Self::Removed => SPARSE_TOMBSTONE_MARKER.clone(),
-        }
-    }
-
-    pub fn decode_option(data: Option<impl AsRef<[u8]>>) -> Result<Self, DecodeError> {
-        match data {
-            Some(marker) if marker.as_ref() == Self::REL_EXISTS_MARKER => Ok(Self::Exists),
-            // Any other marker is invalid
-            Some(_) => Err(DecodeError),
-            None => Ok(Self::Removed),
-        }
-    }
-
-    pub fn decode(data: impl AsRef<[u8]>) -> Result<Self, DecodeError> {
-        let data = data.as_ref();
-        if data == Self::REL_EXISTS_MARKER {
-            Ok(Self::Exists)
-        } else if data == SPARSE_TOMBSTONE_MARKER {
-            Ok(Self::Removed)
-        } else {
-            Err(DecodeError)
-        }
-    }
-}
-
-/// A tombstone in the sparse keyspace, which is an empty buffer.
-pub const SPARSE_TOMBSTONE_MARKER: Bytes = Bytes::from_static(b"");
-
 /// Check if the key falls in the range of metadata keys.
 pub const fn is_metadata_key_slice(key: &[u8]) -> bool {
     key[0] >= METADATA_KEY_BEGIN_PREFIX && key[0] < METADATA_KEY_END_PREFIX
@@ -170,24 +110,6 @@ impl Key {
         }
     }
 
-    pub fn rel_dir_sparse_key_range() -> Range<Self> {
-        Key {
-            field1: REL_DIR_KEY_PREFIX,
-            field2: 0,
-            field3: 0,
-            field4: 0,
-            field5: 0,
-            field6: 0,
-        }..Key {
-            field1: REL_DIR_KEY_PREFIX + 1,
-            field2: 0,
-            field3: 0,
-            field4: 0,
-            field5: 0,
-            field6: 0,
-        }
-    }
-
     /// This function checks more extensively what keys we can take on the write path.
     /// If a key beginning with 00 does not have a global/default tablespace OID, it
     /// will be rejected on the write path.
@@ -518,36 +440,6 @@ pub fn rel_dir_to_key(spcnode: Oid, dbnode: Oid) -> Key {
     }
 }
 
-#[inline(always)]
-pub fn rel_tag_sparse_key(spcnode: Oid, dbnode: Oid, relnode: Oid, forknum: u8) -> Key {
-    Key {
-        field1: REL_DIR_KEY_PREFIX,
-        field2: spcnode,
-        field3: dbnode,
-        field4: relnode,
-        field5: forknum,
-        field6: 1,
-    }
-}
-
-pub fn rel_tag_sparse_key_range(spcnode: Oid, dbnode: Oid) -> Range<Key> {
-    Key {
-        field1: REL_DIR_KEY_PREFIX,
-        field2: spcnode,
-        field3: dbnode,
-        field4: 0,
-        field5: 0,
-        field6: 0,
-    }..Key {
-        field1: REL_DIR_KEY_PREFIX,
-        field2: spcnode,
-        field3: dbnode,
-        field4: u32::MAX,
-        field5: u8::MAX,
-        field6: u32::MAX,
-    } // it's fine to exclude the last key b/c we only use field6 == 1
-}
-
 #[inline(always)]
 pub fn rel_block_to_key(rel: RelTag, blknum: BlockNumber) -> Key {
     Key {
@@ -842,9 +734,9 @@ impl Key {
         self.field1 == RELATION_SIZE_PREFIX
     }
 
-    pub const fn sparse_non_inherited_keyspace() -> Range<Key> {
+    pub fn sparse_non_inherited_keyspace() -> Range<Key> {
         // The two keys are adjacent; if we will have non-adjancent keys in the future, we should return a keyspace
-        const_assert!(AUX_KEY_PREFIX + 1 == REPL_ORIGIN_KEY_PREFIX);
+        debug_assert_eq!(AUX_KEY_PREFIX + 1, REPL_ORIGIN_KEY_PREFIX);
         Key {
             field1: AUX_KEY_PREFIX,
             field2: 0,

libs/pageserver_api/src/keyspace.rs

@@ -27,6 +27,55 @@ impl std::fmt::Display for KeySpace {
     }
 }
 
+pub struct KeySpaceIter<'a> {
+    ranges: &'a [Range<Key>],
+    current_range: usize,
+    current_key: Option<Key>,
+}
+
+impl<'a> Iterator for KeySpaceIter<'a> {
+    // (current_key, range_end)
+    type Item = (Key, Key);
+
+    fn next(&mut self) -> Option<Self::Item> {
+        // if we've gone through all ranges, stop iteration
+        if self.current_range >= self.ranges.len() {
+            return None;
+        }
+
+        let range = &self.ranges[self.current_range];
+
+        // if current_key is None, initialize it with range.start
+        if self.current_key.is_none() {
+            self.current_key = Some(range.start);
+        }
+
+        if let Some(current_key) = self.current_key {
+            // check if the current_key is still within the range
+            if current_key < range.end {
+                // get the next key
+                let next_key = current_key.next();
+                // move current_key forward
+                self.current_key = Some(next_key);
+                // return the range from current_key to the current range_end
+                return Some((current_key, range.end));
+            }
+        }
+
+        // move to the next range
+        self.current_range += 1;
+
+        // if there are more ranges, initialize the next key from the start of the next range
+        if self.current_range < self.ranges.len() {
+            self.current_key = Some(self.ranges[self.current_range].start);
+            self.next() // Recurse to continue iterating with the new range
+        } else {
+            // no more ranges, end iteration
+            None
+        }
+    }
+}
+
 /// A wrapper type for sparse keyspaces.
 #[derive(Clone, Debug, Default, PartialEq, Eq)]
 pub struct SparseKeySpace(pub KeySpace);
@@ -435,6 +484,14 @@ impl KeySpace {
     pub fn contains(&self, key: &Key) -> bool {
         self.overlaps(&(*key..key.next()))
     }
+
+    pub fn iter(&self) -> KeySpaceIter {
+        KeySpaceIter {
+            ranges: &self.ranges,
+            current_range: 0,
+            current_key: self.ranges.first().map(|r| r.start),
+        }
+    }
 }
 
 ///
@@ -619,6 +676,92 @@ mod tests {
     use super::*;
     use std::fmt::Write;
 
+    fn key(field1: u8, field2: u32, field6: u32) -> Key {
+        Key {
+            field1,
+            field2,
+            field3: 0,
+            field4: 0,
+            field5: 0,
+            field6,
+        }
+    }
+
+    #[test]
+    fn test_iter_single_range() {
+        let key_range = Range {
+            start: key(1, 0, 0),
+            end: key(1, 0, 5),
+        };
+        let keyspace = KeySpace {
+            ranges: vec![key_range],
+        };
+        let collected_ranges: Vec<_> = keyspace.iter().collect();
+
+        assert_eq!(
+            collected_ranges,
+            vec![
+                (key(1, 0, 0), key(1, 0, 5)),
+                (key(1, 0, 1), key(1, 0, 5)),
+                (key(1, 0, 2), key(1, 0, 5)),
+                (key(1, 0, 3), key(1, 0, 5)),
+                (key(1, 0, 4), key(1, 0, 5)), // Stops at field6 = 5
+            ]
+        );
+    }
+
+    #[test]
+    fn test_iter_multiple_ranges() {
+        let ranges = vec![
+            Range {
+                start: key(1, 0, 0),
+                end: key(1, 0, 3),
+            },
+            Range {
+                start: key(2, 0, 0),
+                end: key(2, 0, 2),
+            },
+        ];
+
+        let keyspace = KeySpace { ranges };
+        let collected_ranges: Vec<_> = keyspace.iter().collect();
+
+        assert_eq!(
+            collected_ranges,
+            vec![
+                (key(1, 0, 0), key(1, 0, 3)),
+                (key(1, 0, 1), key(1, 0, 3)),
+                (key(1, 0, 2), key(1, 0, 3)), // End of first range
+                (key(2, 0, 0), key(2, 0, 2)),
+                (key(2, 0, 1), key(2, 0, 2)), // End of second range
+            ]
+        );
+    }
+
+    #[test]
+    fn test_iter_empty_keyspace() {
+        let keyspace = KeySpace { ranges: vec![] };
+        let mut iter = keyspace.iter();
+
+        assert_eq!(iter.next(), None);
+    }
+
+    #[test]
+    fn test_iter_range_with_single_key() {
+        let key_range = Range {
+            start: key(1, 42, 0),
+            end: key(1, 42, 1), // Only one step
+        };
+
+        let keyspace = KeySpace {
+            ranges: vec![key_range],
+        };
+
+        let mut iter = keyspace.iter();
+        assert_eq!(iter.next(), Some((key(1, 42, 0), key(1, 42, 1))));
+        assert_eq!(iter.next(), None);
+    }
+
     // Helper function to create a key range.
     //
     // Make the tests below less verbose.

libs/pageserver_api/src/models.rs

@@ -1136,26 +1136,7 @@ pub struct TimelineInfo {
     pub ancestor_lsn: Option<Lsn>,
     pub last_record_lsn: Lsn,
     pub prev_record_lsn: Option<Lsn>,
-
-    /// Legacy field for compat with control plane.  Synonym of `min_readable_lsn`.
-    /// TODO: remove once control plane no longer reads it.
     pub latest_gc_cutoff_lsn: Lsn,
-
-    /// The LSN up to which GC has advanced: older data may still exist but it is not available for clients.
-    /// This LSN is not suitable for deciding where to create branches etc: use [`TimelineInfo::min_readable_lsn`] instead,
-    /// as it is easier to reason about.
-    #[serde(default)]
-    pub applied_gc_cutoff_lsn: Lsn,
-
-    /// The upper bound of data which is either already GC'ed, or elegible to be GC'ed at any time based on PITR interval.
-    /// This LSN represents the "end of history" for this timeline, and callers should use it to figure out the oldest
-    /// LSN at which it is legal to create a branch or ephemeral endpoint.
-    ///
-    /// Note that holders of valid LSN leases may be able to create branches and read pages earlier
-    /// than this LSN, but new leases may not be taken out earlier than this LSN.
-    #[serde(default)]
-    pub min_readable_lsn: Lsn,
-
     pub disk_consistent_lsn: Lsn,
 
     /// The LSN that we have succesfully uploaded to remote storage

libs/postgres_backend/src/lib.rs

@@ -9,8 +9,6 @@ use bytes::Bytes;
 use serde::{Deserialize, Serialize};
 use std::io::ErrorKind;
 use std::net::SocketAddr;
-use std::os::fd::AsRawFd;
-use std::os::fd::RawFd;
 use std::pin::Pin;
 use std::sync::Arc;
 use std::task::{ready, Poll};
@@ -270,7 +268,6 @@ impl<IO: AsyncRead + AsyncWrite + Unpin> MaybeWriteOnly<IO> {
 }
 
 pub struct PostgresBackend<IO> {
-    pub socket_fd: RawFd,
     framed: MaybeWriteOnly<IO>,
 
     pub state: ProtoState,
@@ -296,11 +293,9 @@ impl PostgresBackend<tokio::net::TcpStream> {
         tls_config: Option<Arc<rustls::ServerConfig>>,
     ) -> io::Result<Self> {
         let peer_addr = socket.peer_addr()?;
-        let socket_fd = socket.as_raw_fd();
         let stream = MaybeTlsStream::Unencrypted(socket);
 
         Ok(Self {
-            socket_fd,
             framed: MaybeWriteOnly::Full(Framed::new(stream)),
             state: ProtoState::Initialization,
             auth_type,
@@ -312,7 +307,6 @@ impl PostgresBackend<tokio::net::TcpStream> {
 
 impl<IO: AsyncRead + AsyncWrite + Unpin> PostgresBackend<IO> {
     pub fn new_from_io(
-        socket_fd: RawFd,
         socket: IO,
         peer_addr: SocketAddr,
         auth_type: AuthType,
@@ -321,7 +315,6 @@ impl<IO: AsyncRead + AsyncWrite + Unpin> PostgresBackend<IO> {
         let stream = MaybeTlsStream::Unencrypted(socket);
 
         Ok(Self {
-            socket_fd,
             framed: MaybeWriteOnly::Full(Framed::new(stream)),
             state: ProtoState::Initialization,
             auth_type,

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

@@ -10,8 +10,8 @@ use crate::simple_query::SimpleQueryStream;
 use crate::types::{Oid, ToSql, Type};
 
 use crate::{
-    query, simple_query, slice_iter, CancelToken, Error, ReadyForQueryStatus, Row,
-    SimpleQueryMessage, Statement, Transaction, TransactionBuilder,
+    prepare, query, simple_query, slice_iter, CancelToken, Error, ReadyForQueryStatus, Row,
+    SimpleQueryMessage, Statement, ToStatement, Transaction, TransactionBuilder,
 };
 use bytes::BytesMut;
 use fallible_iterator::FallibleIterator;
@@ -54,18 +54,18 @@ impl Responses {
 }
 
 /// A cache of type info and prepared statements for fetching type info
-/// (corresponding to the queries in the [crate::prepare] module).
+/// (corresponding to the queries in the [prepare] module).
 #[derive(Default)]
 struct CachedTypeInfo {
     /// A statement for basic information for a type from its
-    /// OID. Corresponds to [TYPEINFO_QUERY](crate::prepare::TYPEINFO_QUERY) (or its
+    /// OID. Corresponds to [TYPEINFO_QUERY](prepare::TYPEINFO_QUERY) (or its
     /// fallback).
     typeinfo: Option<Statement>,
     /// A statement for getting information for a composite type from its OID.
-    /// Corresponds to [TYPEINFO_QUERY](crate::prepare::TYPEINFO_COMPOSITE_QUERY).
+    /// Corresponds to [TYPEINFO_QUERY](prepare::TYPEINFO_COMPOSITE_QUERY).
     typeinfo_composite: Option<Statement>,
     /// A statement for getting information for a composite type from its OID.
-    /// Corresponds to [TYPEINFO_QUERY](crate::prepare::TYPEINFO_COMPOSITE_QUERY) (or
+    /// Corresponds to [TYPEINFO_QUERY](prepare::TYPEINFO_COMPOSITE_QUERY) (or
     /// its fallback).
     typeinfo_enum: Option<Statement>,
 
@@ -190,6 +190,26 @@ impl Client {
         &self.inner
     }
 
+    /// Creates a new prepared statement.
+    ///
+    /// Prepared statements can be executed repeatedly, and may contain query parameters (indicated by `$1`, `$2`, etc),
+    /// which are set when executed. Prepared statements can only be used with the connection that created them.
+    pub async fn prepare(&self, query: &str) -> Result<Statement, Error> {
+        self.prepare_typed(query, &[]).await
+    }
+
+    /// Like `prepare`, but allows the types of query parameters to be explicitly specified.
+    ///
+    /// The list of types may be smaller than the number of parameters - the types of the remaining parameters will be
+    /// inferred. For example, `client.prepare_typed(query, &[])` is equivalent to `client.prepare(query)`.
+    pub async fn prepare_typed(
+        &self,
+        query: &str,
+        parameter_types: &[Type],
+    ) -> Result<Statement, Error> {
+        prepare::prepare(&self.inner, query, parameter_types).await
+    }
+
     /// Executes a statement, returning a vector of the resulting rows.
     ///
     /// A statement may contain parameters, specified by `$n`, where `n` is the index of the parameter of the list
@@ -202,11 +222,14 @@ impl Client {
     /// # Panics
     ///
     /// Panics if the number of parameters provided does not match the number expected.
-    pub async fn query(
+    pub async fn query<T>(
         &self,
-        statement: Statement,
+        statement: &T,
         params: &[&(dyn ToSql + Sync)],
-    ) -> Result<Vec<Row>, Error> {
+    ) -> Result<Vec<Row>, Error>
+    where
+        T: ?Sized + ToStatement,
+    {
         self.query_raw(statement, slice_iter(params))
             .await?
             .try_collect()
@@ -227,15 +250,13 @@ impl Client {
     /// Panics if the number of parameters provided does not match the number expected.
     ///
     /// [`query`]: #method.query
-    pub async fn query_raw<'a, I>(
-        &self,
-        statement: Statement,
-        params: I,
-    ) -> Result<RowStream, Error>
+    pub async fn query_raw<'a, T, I>(&self, statement: &T, params: I) -> Result<RowStream, Error>
     where
+        T: ?Sized + ToStatement,
         I: IntoIterator<Item = &'a (dyn ToSql + Sync)>,
         I::IntoIter: ExactSizeIterator,
     {
+        let statement = statement.__convert().into_statement(self).await?;
         query::query(&self.inner, statement, params).await
     }
 
@@ -250,6 +271,55 @@ impl Client {
         query::query_txt(&self.inner, statement, params).await
     }
 
+    /// Executes a statement, returning the number of rows modified.
+    ///
+    /// A statement may contain parameters, specified by `$n`, where `n` is the index of the parameter of the list
+    /// provided, 1-indexed.
+    ///
+    /// The `statement` argument can either be a `Statement`, or a raw query string. If the same statement will be
+    /// repeatedly executed (perhaps with different query parameters), consider preparing the statement up front
+    /// with the `prepare` method.
+    ///
+    /// If the statement does not modify any rows (e.g. `SELECT`), 0 is returned.
+    ///
+    /// # Panics
+    ///
+    /// Panics if the number of parameters provided does not match the number expected.
+    pub async fn execute<T>(
+        &self,
+        statement: &T,
+        params: &[&(dyn ToSql + Sync)],
+    ) -> Result<u64, Error>
+    where
+        T: ?Sized + ToStatement,
+    {
+        self.execute_raw(statement, slice_iter(params)).await
+    }
+
+    /// The maximally flexible version of [`execute`].
+    ///
+    /// A statement may contain parameters, specified by `$n`, where `n` is the index of the parameter of the list
+    /// provided, 1-indexed.
+    ///
+    /// The `statement` argument can either be a `Statement`, or a raw query string. If the same statement will be
+    /// repeatedly executed (perhaps with different query parameters), consider preparing the statement up front
+    /// with the `prepare` method.
+    ///
+    /// # Panics
+    ///
+    /// Panics if the number of parameters provided does not match the number expected.
+    ///
+    /// [`execute`]: #method.execute
+    pub async fn execute_raw<'a, T, I>(&self, statement: &T, params: I) -> Result<u64, Error>
+    where
+        T: ?Sized + ToStatement,
+        I: IntoIterator<Item = &'a (dyn ToSql + Sync)>,
+        I::IntoIter: ExactSizeIterator,
+    {
+        let statement = statement.__convert().into_statement(self).await?;
+        query::execute(self.inner(), statement, params).await
+    }
+
     /// Executes a sequence of SQL statements using the simple query protocol, returning the resulting rows.
     ///
     /// Statements should be separated by semicolons. If an error occurs, execution of the sequence will stop at that

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

@@ -1,8 +1,7 @@
-#![allow(async_fn_in_trait)]
-
 use crate::query::RowStream;
 use crate::types::Type;
 use crate::{Client, Error, Transaction};
+use async_trait::async_trait;
 use postgres_protocol2::Oid;
 
 mod private {
@@ -12,6 +11,7 @@ mod private {
 /// A trait allowing abstraction over connections and transactions.
 ///
 /// This trait is "sealed", and cannot be implemented outside of this crate.
+#[async_trait]
 pub trait GenericClient: private::Sealed {
     /// Like `Client::query_raw_txt`.
     async fn query_raw_txt<S, I>(&self, statement: &str, params: I) -> Result<RowStream, Error>
@@ -26,6 +26,7 @@ pub trait GenericClient: private::Sealed {
 
 impl private::Sealed for Client {}
 
+#[async_trait]
 impl GenericClient for Client {
     async fn query_raw_txt<S, I>(&self, statement: &str, params: I) -> Result<RowStream, Error>
     where
@@ -38,12 +39,14 @@ impl GenericClient for Client {
 
     /// Query for type information
     async fn get_type(&self, oid: Oid) -> Result<Type, Error> {
-        crate::prepare::get_type(self.inner(), oid).await
+        self.get_type(oid).await
     }
 }
 
 impl private::Sealed for Transaction<'_> {}
 
+#[async_trait]
+#[allow(clippy::needless_lifetimes)]
 impl GenericClient for Transaction<'_> {
     async fn query_raw_txt<S, I>(&self, statement: &str, params: I) -> Result<RowStream, Error>
     where

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

@@ -14,6 +14,7 @@ pub use crate::row::{Row, SimpleQueryRow};
 pub use crate::simple_query::SimpleQueryStream;
 pub use crate::statement::{Column, Statement};
 pub use crate::tls::NoTls;
+pub use crate::to_statement::ToStatement;
 pub use crate::transaction::Transaction;
 pub use crate::transaction_builder::{IsolationLevel, TransactionBuilder};
 use crate::types::ToSql;
@@ -64,6 +65,7 @@ pub mod row;
 mod simple_query;
 mod statement;
 pub mod tls;
+mod to_statement;
 mod transaction;
 mod transaction_builder;
 pub mod types;

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

@@ -1,6 +1,7 @@
 use crate::client::InnerClient;
 use crate::codec::FrontendMessage;
 use crate::connection::RequestMessages;
+use crate::error::SqlState;
 use crate::types::{Field, Kind, Oid, Type};
 use crate::{query, slice_iter};
 use crate::{Column, Error, Statement};
@@ -12,6 +13,7 @@ use postgres_protocol2::message::backend::Message;
 use postgres_protocol2::message::frontend;
 use std::future::Future;
 use std::pin::Pin;
+use std::sync::atomic::{AtomicUsize, Ordering};
 use std::sync::Arc;
 
 pub(crate) const TYPEINFO_QUERY: &str = "\
@@ -22,6 +24,14 @@ INNER JOIN pg_catalog.pg_namespace n ON t.typnamespace = n.oid
 WHERE t.oid = $1
 ";
 
+// Range types weren't added until Postgres 9.2, so pg_range may not exist
+const TYPEINFO_FALLBACK_QUERY: &str = "\
+SELECT t.typname, t.typtype, t.typelem, NULL::OID, t.typbasetype, n.nspname, t.typrelid
+FROM pg_catalog.pg_type t
+INNER JOIN pg_catalog.pg_namespace n ON t.typnamespace = n.oid
+WHERE t.oid = $1
+";
+
 const TYPEINFO_ENUM_QUERY: &str = "\
 SELECT enumlabel
 FROM pg_catalog.pg_enum
@@ -29,6 +39,14 @@ WHERE enumtypid = $1
 ORDER BY enumsortorder
 ";
 
+// Postgres 9.0 didn't have enumsortorder
+const TYPEINFO_ENUM_FALLBACK_QUERY: &str = "\
+SELECT enumlabel
+FROM pg_catalog.pg_enum
+WHERE enumtypid = $1
+ORDER BY oid
+";
+
 pub(crate) const TYPEINFO_COMPOSITE_QUERY: &str = "\
 SELECT attname, atttypid
 FROM pg_catalog.pg_attribute
@@ -38,13 +56,15 @@ AND attnum > 0
 ORDER BY attnum
 ";
 
+static NEXT_ID: AtomicUsize = AtomicUsize::new(0);
+
 pub async fn prepare(
     client: &Arc<InnerClient>,
-    name: &'static str,
     query: &str,
     types: &[Type],
 ) -> Result<Statement, Error> {
-    let buf = encode(client, name, query, types)?;
+    let name = format!("s{}", NEXT_ID.fetch_add(1, Ordering::SeqCst));
+    let buf = encode(client, &name, query, types)?;
     let mut responses = client.send(RequestMessages::Single(FrontendMessage::Raw(buf)))?;
 
     match responses.next().await? {
@@ -85,11 +105,10 @@ pub async fn prepare(
 
 fn prepare_rec<'a>(
     client: &'a Arc<InnerClient>,
-    name: &'static str,
     query: &'a str,
     types: &'a [Type],
 ) -> Pin<Box<dyn Future<Output = Result<Statement, Error>> + 'a + Send>> {
-    Box::pin(prepare(client, name, query, types))
+    Box::pin(prepare(client, query, types))
 }
 
 fn encode(client: &InnerClient, name: &str, query: &str, types: &[Type]) -> Result<Bytes, Error> {
@@ -173,8 +192,13 @@ async fn typeinfo_statement(client: &Arc<InnerClient>) -> Result<Statement, Erro
         return Ok(stmt);
     }
 
-    let typeinfo = "neon_proxy_typeinfo";
-    let stmt = prepare_rec(client, typeinfo, TYPEINFO_QUERY, &[]).await?;
+    let stmt = match prepare_rec(client, TYPEINFO_QUERY, &[]).await {
+        Ok(stmt) => stmt,
+        Err(ref e) if e.code() == Some(&SqlState::UNDEFINED_TABLE) => {
+            prepare_rec(client, TYPEINFO_FALLBACK_QUERY, &[]).await?
+        }
+        Err(e) => return Err(e),
+    };
 
     client.set_typeinfo(&stmt);
     Ok(stmt)
@@ -195,8 +219,13 @@ async fn typeinfo_enum_statement(client: &Arc<InnerClient>) -> Result<Statement,
         return Ok(stmt);
     }
 
-    let typeinfo = "neon_proxy_typeinfo_enum";
-    let stmt = prepare_rec(client, typeinfo, TYPEINFO_ENUM_QUERY, &[]).await?;
+    let stmt = match prepare_rec(client, TYPEINFO_ENUM_QUERY, &[]).await {
+        Ok(stmt) => stmt,
+        Err(ref e) if e.code() == Some(&SqlState::UNDEFINED_COLUMN) => {
+            prepare_rec(client, TYPEINFO_ENUM_FALLBACK_QUERY, &[]).await?
+        }
+        Err(e) => return Err(e),
+    };
 
     client.set_typeinfo_enum(&stmt);
     Ok(stmt)
@@ -226,8 +255,7 @@ async fn typeinfo_composite_statement(client: &Arc<InnerClient>) -> Result<State
         return Ok(stmt);
     }
 
-    let typeinfo = "neon_proxy_typeinfo_composite";
-    let stmt = prepare_rec(client, typeinfo, TYPEINFO_COMPOSITE_QUERY, &[]).await?;
+    let stmt = prepare_rec(client, TYPEINFO_COMPOSITE_QUERY, &[]).await?;
 
     client.set_typeinfo_composite(&stmt);
     Ok(stmt)

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

@@ -157,6 +157,49 @@ where
     })
 }
 
+pub async fn execute<'a, I>(
+    client: &InnerClient,
+    statement: Statement,
+    params: I,
+) -> Result<u64, Error>
+where
+    I: IntoIterator<Item = &'a (dyn ToSql + Sync)>,
+    I::IntoIter: ExactSizeIterator,
+{
+    let buf = if log_enabled!(Level::Debug) {
+        let params = params.into_iter().collect::<Vec<_>>();
+        debug!(
+            "executing statement {} with parameters: {:?}",
+            statement.name(),
+            BorrowToSqlParamsDebug(params.as_slice()),
+        );
+        encode(client, &statement, params)?
+    } else {
+        encode(client, &statement, params)?
+    };
+    let mut responses = start(client, buf).await?;
+
+    let mut rows = 0;
+    loop {
+        match responses.next().await? {
+            Message::DataRow(_) => {}
+            Message::CommandComplete(body) => {
+                rows = body
+                    .tag()
+                    .map_err(Error::parse)?
+                    .rsplit(' ')
+                    .next()
+                    .unwrap()
+                    .parse()
+                    .unwrap_or(0);
+            }
+            Message::EmptyQueryResponse => rows = 0,
+            Message::ReadyForQuery(_) => return Ok(rows),
+            _ => return Err(Error::unexpected_message()),
+        }
+    }
+}
+
 async fn start(client: &InnerClient, buf: Bytes) -> Result<Responses, Error> {
     let mut responses = client.send(RequestMessages::Single(FrontendMessage::Raw(buf)))?;
 

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

@@ -13,7 +13,7 @@ use std::{
 
 struct StatementInner {
     client: Weak<InnerClient>,
-    name: &'static str,
+    name: String,
     params: Vec<Type>,
     columns: Vec<Column>,
 }
@@ -22,7 +22,7 @@ impl Drop for StatementInner {
     fn drop(&mut self) {
         if let Some(client) = self.client.upgrade() {
             let buf = client.with_buf(|buf| {
-                frontend::close(b'S', self.name, buf).unwrap();
+                frontend::close(b'S', &self.name, buf).unwrap();
                 frontend::sync(buf);
                 buf.split().freeze()
             });
@@ -40,7 +40,7 @@ pub struct Statement(Arc<StatementInner>);
 impl Statement {
     pub(crate) fn new(
         inner: &Arc<InnerClient>,
-        name: &'static str,
+        name: String,
         params: Vec<Type>,
         columns: Vec<Column>,
     ) -> Statement {
@@ -55,14 +55,14 @@ impl Statement {
     pub(crate) fn new_anonymous(params: Vec<Type>, columns: Vec<Column>) -> Statement {
         Statement(Arc::new(StatementInner {
             client: Weak::new(),
-            name: "<anonymous>",
+            name: String::new(),
             params,
             columns,
         }))
     }
 
     pub(crate) fn name(&self) -> &str {
-        self.0.name
+        &self.0.name
     }
 
     /// Returns the expected types of the statement's parameters.

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

@@ -0,0 +1,57 @@
+use crate::to_statement::private::{Sealed, ToStatementType};
+use crate::Statement;
+
+mod private {
+    use crate::{Client, Error, Statement};
+
+    pub trait Sealed {}
+
+    pub enum ToStatementType<'a> {
+        Statement(&'a Statement),
+        Query(&'a str),
+    }
+
+    impl ToStatementType<'_> {
+        pub async fn into_statement(self, client: &Client) -> Result<Statement, Error> {
+            match self {
+                ToStatementType::Statement(s) => Ok(s.clone()),
+                ToStatementType::Query(s) => client.prepare(s).await,
+            }
+        }
+    }
+}
+
+/// A trait abstracting over prepared and unprepared statements.
+///
+/// Many methods are generic over this bound, so that they support both a raw query string as well as a statement which
+/// was prepared previously.
+///
+/// This trait is "sealed" and cannot be implemented by anything outside this crate.
+pub trait ToStatement: Sealed {
+    #[doc(hidden)]
+    fn __convert(&self) -> ToStatementType<'_>;
+}
+
+impl ToStatement for Statement {
+    fn __convert(&self) -> ToStatementType<'_> {
+        ToStatementType::Statement(self)
+    }
+}
+
+impl Sealed for Statement {}
+
+impl ToStatement for str {
+    fn __convert(&self) -> ToStatementType<'_> {
+        ToStatementType::Query(self)
+    }
+}
+
+impl Sealed for str {}
+
+impl ToStatement for String {
+    fn __convert(&self) -> ToStatementType<'_> {
+        ToStatementType::Query(self)
+    }
+}
+
+impl Sealed for String {}

libs/utils/Cargo.toml

@@ -28,7 +28,7 @@ inferno.workspace = true
 fail.workspace = true
 futures = { workspace = true }
 jsonwebtoken.workspace = true
-nix = {workspace = true, features = [ "ioctl" ] }
+nix.workspace = true
 once_cell.workspace = true
 pin-project-lite.workspace = true
 regex.workspace = true

libs/utils/src/lib.rs

@@ -93,9 +93,6 @@ pub mod try_rcu;
 
 pub mod guard_arc_swap;
 
-#[cfg(target_os = "linux")]
-pub mod linux_socket_ioctl;
-
 // Re-export used in macro. Avoids adding git-version as dep in target crates.
 #[doc(hidden)]
 pub use git_version;

libs/utils/src/linux_socket_ioctl.rs

@@ -1,35 +0,0 @@
-//! Linux-specific socket ioctls.
-//!
-//! <https://elixir.bootlin.com/linux/v6.1.128/source/include/uapi/linux/sockios.h#L25-L27>
-
-use std::{
-    io,
-    mem::MaybeUninit,
-    os::{fd::RawFd, raw::c_int},
-};
-
-use nix::libc::{FIONREAD, TIOCOUTQ};
-
-unsafe fn do_ioctl(socket_fd: RawFd, cmd: nix::libc::Ioctl) -> io::Result<c_int> {
-    let mut inq: MaybeUninit<c_int> = MaybeUninit::uninit();
-    let err = nix::libc::ioctl(socket_fd, cmd, inq.as_mut_ptr());
-    if err == 0 {
-        Ok(inq.assume_init())
-    } else {
-        Err(io::Error::last_os_error())
-    }
-}
-
-/// # Safety
-///
-/// Caller must ensure that `socket_fd` is a valid TCP socket file descriptor.
-pub unsafe fn inq(socket_fd: RawFd) -> io::Result<c_int> {
-    do_ioctl(socket_fd, FIONREAD)
-}
-
-/// # Safety
-///
-/// Caller must ensure that `socket_fd` is a valid TCP socket file descriptor.
-pub unsafe fn outq(socket_fd: RawFd) -> io::Result<c_int> {
-    do_ioctl(socket_fd, TIOCOUTQ)
-}

pageserver/src/basebackup.rs

@@ -13,7 +13,7 @@
 use anyhow::{anyhow, Context};
 use bytes::{BufMut, Bytes, BytesMut};
 use fail::fail_point;
-use pageserver_api::key::{rel_block_to_key, Key};
+use pageserver_api::key::Key;
 use postgres_ffi::pg_constants;
 use std::fmt::Write as FmtWrite;
 use std::time::{Instant, SystemTime};
@@ -42,8 +42,8 @@ use utils::lsn::Lsn;
 pub enum BasebackupError {
     #[error("basebackup pageserver error {0:#}")]
     Server(#[from] anyhow::Error),
-    #[error("basebackup client error {0:#} when {1}")]
-    Client(#[source] io::Error, &'static str),
+    #[error("basebackup client error {0:#}")]
+    Client(#[source] io::Error),
 }
 
 /// Create basebackup with non-rel data in it.
@@ -234,7 +234,7 @@ where
         self.ar
             .append(&header, self.buf.as_slice())
             .await
-            .map_err(|e| BasebackupError::Client(e, "flush"))?;
+            .map_err(BasebackupError::Client)?;
 
         self.total_blocks += nblocks;
         debug!("Added to basebackup slru {} relsize {}", segname, nblocks);
@@ -273,9 +273,9 @@ where
         for dir in subdirs.iter() {
             let header = new_tar_header_dir(dir)?;
             self.ar
-                .append(&header, io::empty())
+                .append(&header, &mut io::empty())
                 .await
-                .map_err(|e| BasebackupError::Client(e, "send_tarball"))?;
+                .context("could not add directory to basebackup tarball")?;
         }
 
         // Send config files.
@@ -286,13 +286,13 @@ where
                 self.ar
                     .append(&header, data)
                     .await
-                    .map_err(|e| BasebackupError::Client(e, "send_tarball,pg_hba.conf"))?;
+                    .context("could not add config file to basebackup tarball")?;
             } else {
                 let header = new_tar_header(filepath, 0)?;
                 self.ar
-                    .append(&header, io::empty())
+                    .append(&header, &mut io::empty())
                     .await
-                    .map_err(|e| BasebackupError::Client(e, "send_tarball,add_config_file"))?;
+                    .context("could not add config file to basebackup tarball")?;
             }
         }
         if !lazy_slru_download {
@@ -406,7 +406,7 @@ where
             self.ar
                 .append(&header, &*content)
                 .await
-                .map_err(|e| BasebackupError::Client(e, "send_tarball,add_aux_file"))?;
+                .context("could not add aux file to basebackup tarball")?;
         }
 
         if min_restart_lsn != Lsn::MAX {
@@ -419,7 +419,7 @@ where
             self.ar
                 .append(&header, &data[..])
                 .await
-                .map_err(|e| BasebackupError::Client(e, "send_tarball,restart.lsn"))?;
+                .context("could not add restart.lsn file to basebackup tarball")?;
         }
         for xid in self
             .timeline
@@ -451,9 +451,9 @@ where
             let crc32 = crc32c::crc32c(&content);
             content.extend_from_slice(&crc32.to_le_bytes());
             let header = new_tar_header("pg_logical/replorigin_checkpoint", content.len() as u64)?;
-            self.ar.append(&header, &*content).await.map_err(|e| {
-                BasebackupError::Client(e, "send_tarball,pg_logical/replorigin_checkpoint")
-            })?;
+            self.ar.append(&header, &*content).await.context(
+                "could not add pg_logical/replorigin_checkpoint file to basebackup tarball",
+            )?;
         }
 
         fail_point!("basebackup-before-control-file", |_| {
@@ -464,10 +464,7 @@ where
 
         // Generate pg_control and bootstrap WAL segment.
         self.add_pgcontrol_file().await?;
-        self.ar
-            .finish()
-            .await
-            .map_err(|e| BasebackupError::Client(e, "send_tarball,finish"))?;
+        self.ar.finish().await.map_err(BasebackupError::Client)?;
         debug!("all tarred up!");
         Ok(())
     }
@@ -485,9 +482,9 @@ where
             let file_name = dst.to_segfile_name(0);
             let header = new_tar_header(&file_name, 0)?;
             self.ar
-                .append(&header, io::empty())
+                .append(&header, &mut io::empty())
                 .await
-                .map_err(|e| BasebackupError::Client(e, "add_rel,empty"))?;
+                .map_err(BasebackupError::Client)?;
             return Ok(());
         }
 
@@ -501,9 +498,13 @@ where
             for blknum in startblk..endblk {
                 let img = self
                     .timeline
-                    // TODO: investigate using get_vectored for the entire startblk..endblk range.
-                    // But this code path is not on the critical path for most basebackups (?).
-                    .get(rel_block_to_key(src, blknum), self.lsn, self.ctx)
+                    .get_rel_page_at_lsn(
+                        src,
+                        blknum,
+                        Version::Lsn(self.lsn),
+                        self.ctx,
+                        self.io_concurrency.clone(),
+                    )
                     .await
                     .map_err(|e| BasebackupError::Server(e.into()))?;
                 segment_data.extend_from_slice(&img[..]);
@@ -514,7 +515,7 @@ where
             self.ar
                 .append(&header, segment_data.as_slice())
                 .await
-                .map_err(|e| BasebackupError::Client(e, "add_rel,segment"))?;
+                .map_err(BasebackupError::Client)?;
 
             seg += 1;
             startblk = endblk;
@@ -565,7 +566,7 @@ where
             self.ar
                 .append(&header, pg_version_str.as_bytes())
                 .await
-                .map_err(|e| BasebackupError::Client(e, "add_dbdir,PG_VERSION"))?;
+                .map_err(BasebackupError::Client)?;
 
             info!("timeline.pg_version {}", self.timeline.pg_version);
 
@@ -575,7 +576,7 @@ where
                 self.ar
                     .append(&header, &img[..])
                     .await
-                    .map_err(|e| BasebackupError::Client(e, "add_dbdir,global/pg_filenode.map"))?;
+                    .map_err(BasebackupError::Client)?;
             } else {
                 warn!("global/pg_filenode.map is missing");
             }
@@ -611,9 +612,9 @@ where
             let path = format!("base/{}", dbnode);
             let header = new_tar_header_dir(&path)?;
             self.ar
-                .append(&header, io::empty())
+                .append(&header, &mut io::empty())
                 .await
-                .map_err(|e| BasebackupError::Client(e, "add_dbdir,base"))?;
+                .map_err(BasebackupError::Client)?;
 
             if let Some(img) = relmap_img {
                 let dst_path = format!("base/{}/PG_VERSION", dbnode);
@@ -626,14 +627,14 @@ where
                 self.ar
                     .append(&header, pg_version_str.as_bytes())
                     .await
-                    .map_err(|e| BasebackupError::Client(e, "add_dbdir,base/PG_VERSION"))?;
+                    .map_err(BasebackupError::Client)?;
 
                 let relmap_path = format!("base/{}/pg_filenode.map", dbnode);
                 let header = new_tar_header(&relmap_path, img.len() as u64)?;
                 self.ar
                     .append(&header, &img[..])
                     .await
-                    .map_err(|e| BasebackupError::Client(e, "add_dbdir,base/pg_filenode.map"))?;
+                    .map_err(BasebackupError::Client)?;
             }
         };
         Ok(())
@@ -662,7 +663,7 @@ where
         self.ar
             .append(&header, &buf[..])
             .await
-            .map_err(|e| BasebackupError::Client(e, "add_twophase_file"))?;
+            .map_err(BasebackupError::Client)?;
 
         Ok(())
     }
@@ -692,7 +693,7 @@ where
                 zenith_signal.as_bytes(),
             )
             .await
-            .map_err(|e| BasebackupError::Client(e, "add_pgcontrol_file,zenith.signal"))?;
+            .map_err(BasebackupError::Client)?;
 
         let checkpoint_bytes = self
             .timeline
@@ -717,7 +718,7 @@ where
         self.ar
             .append(&header, &pg_control_bytes[..])
             .await
-            .map_err(|e| BasebackupError::Client(e, "add_pgcontrol_file,pg_control"))?;
+            .map_err(BasebackupError::Client)?;
 
         //send wal segment
         let segno = self.lsn.segment_number(WAL_SEGMENT_SIZE);
@@ -741,7 +742,7 @@ where
         self.ar
             .append(&header, &wal_seg[..])
             .await
-            .map_err(|e| BasebackupError::Client(e, "add_pgcontrol_file,wal_segment"))?;
+            .map_err(BasebackupError::Client)?;
         Ok(())
     }
 }

pageserver/src/http/openapi_spec.yml

@@ -1080,10 +1080,7 @@ components:
           type: integer
         state:
           type: string
-        min_readable_lsn:
-          type: string
-          format: hex
-        applied_gc_cutoff_lsn:
+        latest_gc_cutoff_lsn:
           type: string
           format: hex
 

pageserver/src/http/routes.rs

@@ -482,11 +482,6 @@ async fn build_timeline_info_common(
 
     let (pitr_history_size, within_ancestor_pitr) = timeline.get_pitr_history_stats();
 
-    let min_readable_lsn = std::cmp::max(
-        timeline.get_gc_cutoff_lsn(),
-        *timeline.get_applied_gc_cutoff_lsn(),
-    );
-
     let info = TimelineInfo {
         tenant_id: timeline.tenant_shard_id,
         timeline_id: timeline.timeline_id,
@@ -498,12 +493,7 @@ async fn build_timeline_info_common(
         initdb_lsn,
         last_record_lsn,
         prev_record_lsn: Some(timeline.get_prev_record_lsn()),
-        // Externally, expose the lowest LSN that can be used to create a branch as the "GC cutoff", although internally
-        // we distinguish between the "planned" GC cutoff (PITR point) and the "latest" GC cutoff (where we
-        // actually trimmed data to), which can pass each other when PITR is changed.
-        latest_gc_cutoff_lsn: min_readable_lsn,
-        min_readable_lsn,
-        applied_gc_cutoff_lsn: *timeline.get_applied_gc_cutoff_lsn(),
+        latest_gc_cutoff_lsn: *timeline.get_latest_gc_cutoff_lsn(),
         current_logical_size: current_logical_size.size_dont_care_about_accuracy(),
         current_logical_size_is_accurate: match current_logical_size.accuracy() {
             tenant::timeline::logical_size::Accuracy::Approximate => false,

pageserver/src/metrics.rs

@@ -1,6 +1,5 @@
 use std::collections::HashMap;
 use std::num::NonZeroUsize;
-use std::os::fd::RawFd;
 use std::pin::Pin;
 use std::sync::atomic::AtomicU64;
 use std::sync::{Arc, Mutex};
@@ -130,7 +129,7 @@ pub(crate) static LAYERS_PER_READ: Lazy<HistogramVec> = Lazy::new(|| {
         "Layers visited to serve a single read (read amplification). In a batch, all visited layers count towards every read.",
         &["tenant_id", "shard_id", "timeline_id"],
         // Low resolution to reduce cardinality.
-        vec![4.0, 8.0, 16.0, 32.0, 64.0, 128.0, 256.0],
+        vec![1.0, 5.0, 10.0, 25.0, 50.0, 100.0],
     )
     .expect("failed to define a metric")
 });
@@ -1440,66 +1439,27 @@ impl Drop for SmgrOpTimer {
 }
 
 impl SmgrOpFlushInProgress {
-    /// The caller must guarantee that `socket_fd`` outlives this function.
-    pub(crate) async fn measure<Fut, O>(
-        self,
-        started_at: Instant,
-        mut fut: Fut,
-        socket_fd: RawFd,
-    ) -> O
+    pub(crate) async fn measure<Fut, O>(self, mut started_at: Instant, mut fut: Fut) -> O
     where
         Fut: std::future::Future<Output = O>,
     {
         let mut fut = std::pin::pin!(fut);
 
-        let mut logged = false;
-        let mut last_counter_increment_at = started_at;
+        // Whenever observe_guard gets called, or dropped,
+        // it adds the time elapsed since its last call to metrics.
+        // Last call is tracked in `now`.
         let mut observe_guard = scopeguard::guard(
-            |is_timeout| {
+            || {
                 let now = Instant::now();
-
-                // Increment counter
-                {
-                    let elapsed_since_last_observe = now - last_counter_increment_at;
-                    self.global_micros
-                        .inc_by(u64::try_from(elapsed_since_last_observe.as_micros()).unwrap());
-                    self.per_timeline_micros
-                        .inc_by(u64::try_from(elapsed_since_last_observe.as_micros()).unwrap());
-                    last_counter_increment_at = now;
-                }
-
-                // Log something on every timeout, and on completion but only if we hit a timeout.
-                if is_timeout || logged {
-                    logged = true;
-                    let elapsed_total = now - started_at;
-                    let msg = if is_timeout {
-                        "slow flush ongoing"
-                    } else {
-                        "slow flush completed or cancelled"
-                    };
-
-                    let (inq, outq) = {
-                        // SAFETY: caller guarantees that `socket_fd` outlives this function.
-                        #[cfg(target_os = "linux")]
-                        unsafe {
-                            (
-                                utils::linux_socket_ioctl::inq(socket_fd).unwrap_or(-2),
-                                utils::linux_socket_ioctl::outq(socket_fd).unwrap_or(-2),
-                            )
-                        }
-                        #[cfg(not(target_os = "linux"))]
-                        {
-                            _ = socket_fd; // appease unused lint on macOS
-                            (-1, -1)
-                        }
-                    };
-
-                    let elapsed_total_secs = format!("{:.6}", elapsed_total.as_secs_f64());
-                    tracing::info!(elapsed_total_secs, inq, outq, msg);
-                }
+                let elapsed = now - started_at;
+                self.global_micros
+                    .inc_by(u64::try_from(elapsed.as_micros()).unwrap());
+                self.per_timeline_micros
+                    .inc_by(u64::try_from(elapsed.as_micros()).unwrap());
+                started_at = now;
             },
             |mut observe| {
-                observe(false);
+                observe();
             },
         );
 
@@ -1507,7 +1467,7 @@ impl SmgrOpFlushInProgress {
             match tokio::time::timeout(Duration::from_secs(10), &mut fut).await {
                 Ok(v) => return v,
                 Err(_timeout) => {
-                    (*observe_guard)(true);
+                    (*observe_guard)();
                 }
             }
         }

pageserver/src/page_service.rs

@@ -73,7 +73,6 @@ use pageserver_api::models::PageTraceEvent;
 use pageserver_api::reltag::SlruKind;
 use postgres_ffi::pg_constants::DEFAULTTABLESPACE_OID;
 use postgres_ffi::BLCKSZ;
-use std::os::fd::AsRawFd;
 
 /// How long we may wait for a [`crate::tenant::mgr::TenantSlot::InProgress`]` and/or a [`crate::tenant::Tenant`] which
 /// is not yet in state [`TenantState::Active`].
@@ -237,7 +236,7 @@ pub async fn libpq_listener_main(
 
 type ConnectionHandlerResult = anyhow::Result<()>;
 
-#[instrument(skip_all, fields(peer_addr, application_name))]
+#[instrument(skip_all, fields(peer_addr))]
 #[allow(clippy::too_many_arguments)]
 async fn page_service_conn_main(
     conf: &'static PageServerConf,
@@ -258,8 +257,6 @@ async fn page_service_conn_main(
         .set_nodelay(true)
         .context("could not set TCP_NODELAY")?;
 
-    let socket_fd = socket.as_raw_fd();
-
     let peer_addr = socket.peer_addr().context("get peer address")?;
     tracing::Span::current().record("peer_addr", field::display(peer_addr));
 
@@ -308,7 +305,7 @@ async fn page_service_conn_main(
         cancel.clone(),
         gate_guard,
     );
-    let pgbackend = PostgresBackend::new_from_io(socket_fd, socket, peer_addr, auth_type, None)?;
+    let pgbackend = PostgresBackend::new_from_io(socket, peer_addr, auth_type, None)?;
 
     match pgbackend.run(&mut conn_handler, &cancel).await {
         Ok(()) => {
@@ -917,7 +914,7 @@ impl PageServerHandler {
                     &shard,
                     req.hdr.request_lsn,
                     req.hdr.not_modified_since,
-                    &shard.get_applied_gc_cutoff_lsn(),
+                    &shard.get_latest_gc_cutoff_lsn(),
                     ctx,
                 )
                 // TODO: if we actually need to wait for lsn here, it delays the entire batch which doesn't need to wait
@@ -1289,15 +1286,12 @@ impl PageServerHandler {
             ))?;
 
             // what we want to do
-            let socket_fd = pgb_writer.socket_fd;
             let flush_fut = pgb_writer.flush();
             // metric for how long flushing takes
             let flush_fut = match flushing_timer {
-                Some(flushing_timer) => futures::future::Either::Left(flushing_timer.measure(
-                    Instant::now(),
-                    flush_fut,
-                    socket_fd,
-                )),
+                Some(flushing_timer) => {
+                    futures::future::Either::Left(flushing_timer.measure(Instant::now(), flush_fut))
+                }
                 None => futures::future::Either::Right(flush_fut),
             };
             // do it while respecting cancellation
@@ -1816,7 +1810,7 @@ impl PageServerHandler {
         req: &PagestreamExistsRequest,
         ctx: &RequestContext,
     ) -> Result<PagestreamBeMessage, PageStreamError> {
-        let latest_gc_cutoff_lsn = timeline.get_applied_gc_cutoff_lsn();
+        let latest_gc_cutoff_lsn = timeline.get_latest_gc_cutoff_lsn();
         let lsn = Self::wait_or_get_last_lsn(
             timeline,
             req.hdr.request_lsn,
@@ -1843,7 +1837,7 @@ impl PageServerHandler {
         req: &PagestreamNblocksRequest,
         ctx: &RequestContext,
     ) -> Result<PagestreamBeMessage, PageStreamError> {
-        let latest_gc_cutoff_lsn = timeline.get_applied_gc_cutoff_lsn();
+        let latest_gc_cutoff_lsn = timeline.get_latest_gc_cutoff_lsn();
         let lsn = Self::wait_or_get_last_lsn(
             timeline,
             req.hdr.request_lsn,
@@ -1870,7 +1864,7 @@ impl PageServerHandler {
         req: &PagestreamDbSizeRequest,
         ctx: &RequestContext,
     ) -> Result<PagestreamBeMessage, PageStreamError> {
-        let latest_gc_cutoff_lsn = timeline.get_applied_gc_cutoff_lsn();
+        let latest_gc_cutoff_lsn = timeline.get_latest_gc_cutoff_lsn();
         let lsn = Self::wait_or_get_last_lsn(
             timeline,
             req.hdr.request_lsn,
@@ -1960,7 +1954,7 @@ impl PageServerHandler {
         req: &PagestreamGetSlruSegmentRequest,
         ctx: &RequestContext,
     ) -> Result<PagestreamBeMessage, PageStreamError> {
-        let latest_gc_cutoff_lsn = timeline.get_applied_gc_cutoff_lsn();
+        let latest_gc_cutoff_lsn = timeline.get_latest_gc_cutoff_lsn();
         let lsn = Self::wait_or_get_last_lsn(
             timeline,
             req.hdr.request_lsn,
@@ -2056,8 +2050,7 @@ impl PageServerHandler {
     {
         fn map_basebackup_error(err: BasebackupError) -> QueryError {
             match err {
-                // TODO: passthrough the error site to the final error message?
-                BasebackupError::Client(e, _) => QueryError::Disconnected(ConnectionError::Io(e)),
+                BasebackupError::Client(e) => QueryError::Disconnected(ConnectionError::Io(e)),
                 BasebackupError::Server(e) => QueryError::Other(e),
             }
         }
@@ -2078,7 +2071,7 @@ impl PageServerHandler {
             //return Err(QueryError::NotFound("timeline is archived".into()))
         }
 
-        let latest_gc_cutoff_lsn = timeline.get_applied_gc_cutoff_lsn();
+        let latest_gc_cutoff_lsn = timeline.get_latest_gc_cutoff_lsn();
         if let Some(lsn) = lsn {
             // Backup was requested at a particular LSN. Wait for it to arrive.
             info!("waiting for {}", lsn);
@@ -2158,12 +2151,10 @@ impl PageServerHandler {
                 .await
                 .map_err(map_basebackup_error)?;
             }
-            writer.flush().await.map_err(|e| {
-                map_basebackup_error(BasebackupError::Client(
-                    e,
-                    "handle_basebackup_request,flush",
-                ))
-            })?;
+            writer
+                .flush()
+                .await
+                .map_err(|e| map_basebackup_error(BasebackupError::Client(e)))?;
         }
 
         pgb.write_message_noflush(&BeMessage::CopyDone)
@@ -2463,16 +2454,9 @@ where
     fn startup(
         &mut self,
         _pgb: &mut PostgresBackend<IO>,
-        sm: &FeStartupPacket,
+        _sm: &FeStartupPacket,
     ) -> Result<(), QueryError> {
         fail::fail_point!("ps::connection-start::startup-packet");
-
-        if let FeStartupPacket::StartupMessage { params, .. } = sm {
-            if let Some(app_name) = params.get("application_name") {
-                Span::current().record("application_name", field::display(app_name));
-            }
-        };
-
         Ok(())
     }
 

pageserver/src/pgdatadir_mapping.rs

@@ -23,14 +23,13 @@ use anyhow::{ensure, Context};
 use bytes::{Buf, Bytes, BytesMut};
 use enum_map::Enum;
 use itertools::Itertools;
+use pageserver_api::key::Key;
 use pageserver_api::key::{
     dbdir_key_range, rel_block_to_key, rel_dir_to_key, rel_key_range, rel_size_to_key,
-    rel_tag_sparse_key_range, relmap_file_key, repl_origin_key, repl_origin_key_range,
-    slru_block_to_key, slru_dir_to_key, slru_segment_key_range, slru_segment_size_to_key,
-    twophase_file_key, twophase_key_range, CompactKey, RelDirExists, AUX_FILES_KEY, CHECKPOINT_KEY,
-    CONTROLFILE_KEY, DBDIR_KEY, TWOPHASEDIR_KEY,
+    relmap_file_key, repl_origin_key, repl_origin_key_range, slru_block_to_key, slru_dir_to_key,
+    slru_segment_key_range, slru_segment_size_to_key, twophase_file_key, twophase_key_range,
+    CompactKey, AUX_FILES_KEY, CHECKPOINT_KEY, CONTROLFILE_KEY, DBDIR_KEY, TWOPHASEDIR_KEY,
 };
-use pageserver_api::key::{rel_tag_sparse_key, Key};
 use pageserver_api::keyspace::SparseKeySpace;
 use pageserver_api::record::NeonWalRecord;
 use pageserver_api::reltag::{BlockNumber, RelTag, SlruKind};
@@ -491,33 +490,12 @@ impl Timeline {
         if !dbdirs.contains_key(&(tag.spcnode, tag.dbnode)) {
             return Ok(false);
         }
-
-        // Read path: first read the new reldir keyspace. Early return if the relation exists.
-        // Otherwise, read the old reldir keyspace.
-        // TODO: if IndexPart::rel_size_migration is `Migrated`, we only need to read from v2.
-
-        if self.get_rel_size_v2_enabled() {
-            // fetch directory listing (new)
-            let key = rel_tag_sparse_key(tag.spcnode, tag.dbnode, tag.relnode, tag.forknum);
-            let buf = RelDirExists::decode_option(version.sparse_get(self, key, ctx).await?)
-                .map_err(|_| PageReconstructError::Other(anyhow::anyhow!("invalid reldir key")))?;
-            let exists_v2 = buf == RelDirExists::Exists;
-            // Fast path: if the relation exists in the new format, return true.
-            // TODO: we should have a verification mode that checks both keyspaces
-            // to ensure the relation only exists in one of them.
-            if exists_v2 {
-                return Ok(true);
-            }
-        }
-
-        // fetch directory listing (old)
-
+        // fetch directory listing
         let key = rel_dir_to_key(tag.spcnode, tag.dbnode);
         let buf = version.get(self, key, ctx).await?;
 
         let dir = RelDirectory::des(&buf)?;
-        let exists_v1 = dir.rels.contains(&(tag.relnode, tag.forknum));
-        Ok(exists_v1)
+        Ok(dir.rels.contains(&(tag.relnode, tag.forknum)))
     }
 
     /// Get a list of all existing relations in given tablespace and database.
@@ -535,12 +513,12 @@ impl Timeline {
         version: Version<'_>,
         ctx: &RequestContext,
     ) -> Result<HashSet<RelTag>, PageReconstructError> {
-        // fetch directory listing (old)
+        // fetch directory listing
         let key = rel_dir_to_key(spcnode, dbnode);
         let buf = version.get(self, key, ctx).await?;
 
         let dir = RelDirectory::des(&buf)?;
-        let rels_v1: HashSet<RelTag> =
+        let rels: HashSet<RelTag> =
             HashSet::from_iter(dir.rels.iter().map(|(relnode, forknum)| RelTag {
                 spcnode,
                 dbnode,
@@ -548,46 +526,6 @@ impl Timeline {
                 forknum: *forknum,
             }));
 
-        if !self.get_rel_size_v2_enabled() {
-            return Ok(rels_v1);
-        }
-
-        // scan directory listing (new), merge with the old results
-        let key_range = rel_tag_sparse_key_range(spcnode, dbnode);
-        let io_concurrency = IoConcurrency::spawn_from_conf(
-            self.conf,
-            self.gate
-                .enter()
-                .map_err(|_| PageReconstructError::Cancelled)?,
-        );
-        let results = self
-            .scan(
-                KeySpace::single(key_range),
-                version.get_lsn(),
-                ctx,
-                io_concurrency,
-            )
-            .await?;
-        let mut rels = rels_v1;
-        for (key, val) in results {
-            let val = RelDirExists::decode(&val?)
-                .map_err(|_| PageReconstructError::Other(anyhow::anyhow!("invalid reldir key")))?;
-            assert_eq!(key.field6, 1);
-            assert_eq!(key.field2, spcnode);
-            assert_eq!(key.field3, dbnode);
-            let tag = RelTag {
-                spcnode,
-                dbnode,
-                relnode: key.field4,
-                forknum: key.field5,
-            };
-            if val == RelDirExists::Removed {
-                debug_assert!(!rels.contains(&tag), "removed reltag in v2");
-                continue;
-            }
-            let did_not_contain = rels.insert(tag);
-            debug_assert!(did_not_contain, "duplicate reltag in v2");
-        }
         Ok(rels)
     }
 
@@ -673,7 +611,7 @@ impl Timeline {
     ) -> Result<LsnForTimestamp, PageReconstructError> {
         pausable_failpoint!("find-lsn-for-timestamp-pausable");
 
-        let gc_cutoff_lsn_guard = self.get_applied_gc_cutoff_lsn();
+        let gc_cutoff_lsn_guard = self.get_latest_gc_cutoff_lsn();
         let gc_cutoff_planned = {
             let gc_info = self.gc_info.read().unwrap();
             gc_info.min_cutoff()
@@ -1206,11 +1144,7 @@ impl Timeline {
 
         let dense_keyspace = result.to_keyspace();
         let sparse_keyspace = SparseKeySpace(KeySpace {
-            ranges: vec![
-                Key::metadata_aux_key_range(),
-                repl_origin_key_range(),
-                Key::rel_dir_sparse_key_range(),
-            ],
+            ranges: vec![Key::metadata_aux_key_range(), repl_origin_key_range()],
         });
 
         if cfg!(debug_assertions) {
@@ -1340,22 +1274,12 @@ pub struct DatadirModification<'a> {
 
     /// For special "directory" keys that store key-value maps, track the size of the map
     /// if it was updated in this modification.
-    pending_directory_entries: Vec<(DirectoryKind, MetricsUpdate)>,
+    pending_directory_entries: Vec<(DirectoryKind, usize)>,
 
     /// An **approximation** of how many metadata bytes will be written to the EphemeralFile.
     pending_metadata_bytes: usize,
 }
 
-#[derive(Debug, Clone, Copy, PartialEq, Eq)]
-pub enum MetricsUpdate {
-    /// Set the metrics to this value
-    Set(u64),
-    /// Increment the metrics by this value
-    Add(u64),
-    /// Decrement the metrics by this value
-    Sub(u64),
-}
-
 impl DatadirModification<'_> {
     // When a DatadirModification is committed, we do a monolithic serialization of all its contents.  WAL records can
     // contain multiple pages, so the pageserver's record-based batch size isn't sufficient to bound this allocation: we
@@ -1435,8 +1359,7 @@ impl DatadirModification<'_> {
         let buf = DbDirectory::ser(&DbDirectory {
             dbdirs: HashMap::new(),
         })?;
-        self.pending_directory_entries
-            .push((DirectoryKind::Db, MetricsUpdate::Set(0)));
+        self.pending_directory_entries.push((DirectoryKind::Db, 0));
         self.put(DBDIR_KEY, Value::Image(buf.into()));
 
         let buf = if self.tline.pg_version >= 17 {
@@ -1449,7 +1372,7 @@ impl DatadirModification<'_> {
             })
         }?;
         self.pending_directory_entries
-            .push((DirectoryKind::TwoPhase, MetricsUpdate::Set(0)));
+            .push((DirectoryKind::TwoPhase, 0));
         self.put(TWOPHASEDIR_KEY, Value::Image(buf.into()));
 
         let buf: Bytes = SlruSegmentDirectory::ser(&SlruSegmentDirectory::default())?.into();
@@ -1459,23 +1382,17 @@ impl DatadirModification<'_> {
         // harmless but they'd just be dropped on later compaction.
         if self.tline.tenant_shard_id.is_shard_zero() {
             self.put(slru_dir_to_key(SlruKind::Clog), empty_dir.clone());
-            self.pending_directory_entries.push((
-                DirectoryKind::SlruSegment(SlruKind::Clog),
-                MetricsUpdate::Set(0),
-            ));
+            self.pending_directory_entries
+                .push((DirectoryKind::SlruSegment(SlruKind::Clog), 0));
             self.put(
                 slru_dir_to_key(SlruKind::MultiXactMembers),
                 empty_dir.clone(),
             );
-            self.pending_directory_entries.push((
-                DirectoryKind::SlruSegment(SlruKind::Clog),
-                MetricsUpdate::Set(0),
-            ));
+            self.pending_directory_entries
+                .push((DirectoryKind::SlruSegment(SlruKind::Clog), 0));
             self.put(slru_dir_to_key(SlruKind::MultiXactOffsets), empty_dir);
-            self.pending_directory_entries.push((
-                DirectoryKind::SlruSegment(SlruKind::MultiXactOffsets),
-                MetricsUpdate::Set(0),
-            ));
+            self.pending_directory_entries
+                .push((DirectoryKind::SlruSegment(SlruKind::MultiXactOffsets), 0));
         }
 
         Ok(())
@@ -1741,16 +1658,10 @@ impl DatadirModification<'_> {
         }
         if r.is_none() {
             // Create RelDirectory
-            // TODO: if we have fully migrated to v2, no need to create this directory
             let buf = RelDirectory::ser(&RelDirectory {
                 rels: HashSet::new(),
             })?;
-            self.pending_directory_entries
-                .push((DirectoryKind::Rel, MetricsUpdate::Set(0)));
-            if self.tline.get_rel_size_v2_enabled() {
-                self.pending_directory_entries
-                    .push((DirectoryKind::RelV2, MetricsUpdate::Set(0)));
-            }
+            self.pending_directory_entries.push((DirectoryKind::Rel, 0));
             self.put(
                 rel_dir_to_key(spcnode, dbnode),
                 Value::Image(Bytes::from(buf)),
@@ -1774,10 +1685,8 @@ impl DatadirModification<'_> {
             if !dir.xids.insert(xid) {
                 anyhow::bail!("twophase file for xid {} already exists", xid);
             }
-            self.pending_directory_entries.push((
-                DirectoryKind::TwoPhase,
-                MetricsUpdate::Set(dir.xids.len() as u64),
-            ));
+            self.pending_directory_entries
+                .push((DirectoryKind::TwoPhase, dir.xids.len()));
             Bytes::from(TwoPhaseDirectoryV17::ser(&dir)?)
         } else {
             let xid = xid as u32;
@@ -1785,10 +1694,8 @@ impl DatadirModification<'_> {
             if !dir.xids.insert(xid) {
                 anyhow::bail!("twophase file for xid {} already exists", xid);
             }
-            self.pending_directory_entries.push((
-                DirectoryKind::TwoPhase,
-                MetricsUpdate::Set(dir.xids.len() as u64),
-            ));
+            self.pending_directory_entries
+                .push((DirectoryKind::TwoPhase, dir.xids.len()));
             Bytes::from(TwoPhaseDirectory::ser(&dir)?)
         };
         self.put(TWOPHASEDIR_KEY, Value::Image(newdirbuf));
@@ -1837,10 +1744,8 @@ impl DatadirModification<'_> {
         let mut dir = DbDirectory::des(&buf)?;
         if dir.dbdirs.remove(&(spcnode, dbnode)).is_some() {
             let buf = DbDirectory::ser(&dir)?;
-            self.pending_directory_entries.push((
-                DirectoryKind::Db,
-                MetricsUpdate::Set(dir.dbdirs.len() as u64),
-            ));
+            self.pending_directory_entries
+                .push((DirectoryKind::Db, dir.dbdirs.len()));
             self.put(DBDIR_KEY, Value::Image(buf.into()));
         } else {
             warn!(
@@ -1873,85 +1778,39 @@ impl DatadirModification<'_> {
         // tablespace.  Create the reldir entry for it if so.
         let mut dbdir = DbDirectory::des(&self.get(DBDIR_KEY, ctx).await.context("read db")?)
             .context("deserialize db")?;
-
-        let dbdir_exists =
+        let rel_dir_key = rel_dir_to_key(rel.spcnode, rel.dbnode);
+        let mut rel_dir =
             if let hash_map::Entry::Vacant(e) = dbdir.dbdirs.entry((rel.spcnode, rel.dbnode)) {
                 // Didn't exist. Update dbdir
                 e.insert(false);
                 let buf = DbDirectory::ser(&dbdir).context("serialize db")?;
-                self.pending_directory_entries.push((
-                    DirectoryKind::Db,
-                    MetricsUpdate::Set(dbdir.dbdirs.len() as u64),
-                ));
+                self.pending_directory_entries
+                    .push((DirectoryKind::Db, dbdir.dbdirs.len()));
                 self.put(DBDIR_KEY, Value::Image(buf.into()));
-                false
-            } else {
-                true
-            };
 
-        let rel_dir_key = rel_dir_to_key(rel.spcnode, rel.dbnode);
-        let mut rel_dir = if !dbdir_exists {
-            // Create the RelDirectory
-            RelDirectory::default()
-        } else {
-            // reldir already exists, fetch it
-            RelDirectory::des(&self.get(rel_dir_key, ctx).await.context("read db")?)
-                .context("deserialize db")?
-        };
+                // and create the RelDirectory
+                RelDirectory::default()
+            } else {
+                // reldir already exists, fetch it
+                RelDirectory::des(&self.get(rel_dir_key, ctx).await.context("read db")?)
+                    .context("deserialize db")?
+            };
 
         // Add the new relation to the rel directory entry, and write it back
         if !rel_dir.rels.insert((rel.relnode, rel.forknum)) {
             return Err(RelationError::AlreadyExists);
         }
 
-        if self.tline.get_rel_size_v2_enabled() {
-            let sparse_rel_dir_key =
-                rel_tag_sparse_key(rel.spcnode, rel.dbnode, rel.relnode, rel.forknum);
-            // check if the rel_dir_key exists in v2
-            let val = self
-                .sparse_get(sparse_rel_dir_key, ctx)
-                .await
-                .map_err(|e| RelationError::Other(e.into()))?;
-            let val = RelDirExists::decode_option(val)
-                .map_err(|_| RelationError::Other(anyhow::anyhow!("invalid reldir key")))?;
-            if val == RelDirExists::Exists {
-                return Err(RelationError::AlreadyExists);
-            }
-            self.put(
-                sparse_rel_dir_key,
-                Value::Image(RelDirExists::Exists.encode()),
-            );
-            if !dbdir_exists {
-                self.pending_directory_entries
-                    .push((DirectoryKind::Rel, MetricsUpdate::Set(0)));
-                self.pending_directory_entries
-                    .push((DirectoryKind::RelV2, MetricsUpdate::Set(0)));
-                // We don't write `rel_dir_key -> rel_dir.rels` back to the storage in the v2 path unless it's the initial creation.
-                // TODO: if we have fully migrated to v2, no need to create this directory. Otherwise, there
-                // will be key not found errors if we don't create an empty one for rel_size_v2.
-                self.put(
-                    rel_dir_key,
-                    Value::Image(Bytes::from(
-                        RelDirectory::ser(&RelDirectory::default()).context("serialize")?,
-                    )),
-                );
-            }
-            self.pending_directory_entries
-                .push((DirectoryKind::RelV2, MetricsUpdate::Add(1)));
-        } else {
-            if !dbdir_exists {
-                self.pending_directory_entries
-                    .push((DirectoryKind::Rel, MetricsUpdate::Set(0)))
-            }
-            self.pending_directory_entries
-                .push((DirectoryKind::Rel, MetricsUpdate::Add(1)));
-            self.put(
-                rel_dir_key,
-                Value::Image(Bytes::from(
-                    RelDirectory::ser(&rel_dir).context("serialize")?,
-                )),
-            );
-        }
+        self.pending_directory_entries
+            .push((DirectoryKind::Rel, rel_dir.rels.len()));
+
+        self.put(
+            rel_dir_key,
+            Value::Image(Bytes::from(
+                RelDirectory::ser(&rel_dir).context("serialize")?,
+            )),
+        );
+
         // Put size
         let size_key = rel_size_to_key(rel);
         let buf = nblocks.to_le_bytes();
@@ -2037,34 +1896,9 @@ impl DatadirModification<'_> {
 
             let mut dirty = false;
             for rel_tag in rel_tags {
-                let found = if dir.rels.remove(&(rel_tag.relnode, rel_tag.forknum)) {
-                    self.pending_directory_entries
-                        .push((DirectoryKind::Rel, MetricsUpdate::Sub(1)));
+                if dir.rels.remove(&(rel_tag.relnode, rel_tag.forknum)) {
                     dirty = true;
-                    true
-                } else if self.tline.get_rel_size_v2_enabled() {
-                    // The rel is not found in the old reldir key, so we need to check the new sparse keyspace.
-                    // Note that a relation can only exist in one of the two keyspaces (guaranteed by the ingestion
-                    // logic).
-                    let key =
-                        rel_tag_sparse_key(spc_node, db_node, rel_tag.relnode, rel_tag.forknum);
-                    let val = RelDirExists::decode_option(self.sparse_get(key, ctx).await?)
-                        .map_err(|_| RelationError::Other(anyhow::anyhow!("invalid reldir key")))?;
-                    if val == RelDirExists::Exists {
-                        self.pending_directory_entries
-                            .push((DirectoryKind::RelV2, MetricsUpdate::Sub(1)));
-                        // put tombstone
-                        self.put(key, Value::Image(RelDirExists::Removed.encode()));
-                        // no need to set dirty to true
-                        true
-                    } else {
-                        false
-                    }
-                } else {
-                    false
-                };
 
-                if found {
                     // update logical size
                     let size_key = rel_size_to_key(rel_tag);
                     let old_size = self.get(size_key, ctx).await?.get_u32_le();
@@ -2080,6 +1914,8 @@ impl DatadirModification<'_> {
 
             if dirty {
                 self.put(dir_key, Value::Image(Bytes::from(RelDirectory::ser(&dir)?)));
+                self.pending_directory_entries
+                    .push((DirectoryKind::Rel, dir.rels.len()));
             }
         }
 
@@ -2103,10 +1939,8 @@ impl DatadirModification<'_> {
         if !dir.segments.insert(segno) {
             anyhow::bail!("slru segment {kind:?}/{segno} already exists");
         }
-        self.pending_directory_entries.push((
-            DirectoryKind::SlruSegment(kind),
-            MetricsUpdate::Set(dir.segments.len() as u64),
-        ));
+        self.pending_directory_entries
+            .push((DirectoryKind::SlruSegment(kind), dir.segments.len()));
         self.put(
             dir_key,
             Value::Image(Bytes::from(SlruSegmentDirectory::ser(&dir)?)),
@@ -2153,10 +1987,8 @@ impl DatadirModification<'_> {
         if !dir.segments.remove(&segno) {
             warn!("slru segment {:?}/{} does not exist", kind, segno);
         }
-        self.pending_directory_entries.push((
-            DirectoryKind::SlruSegment(kind),
-            MetricsUpdate::Set(dir.segments.len() as u64),
-        ));
+        self.pending_directory_entries
+            .push((DirectoryKind::SlruSegment(kind), dir.segments.len()));
         self.put(
             dir_key,
             Value::Image(Bytes::from(SlruSegmentDirectory::ser(&dir)?)),
@@ -2188,10 +2020,8 @@ impl DatadirModification<'_> {
             if !dir.xids.remove(&xid) {
                 warn!("twophase file for xid {} does not exist", xid);
             }
-            self.pending_directory_entries.push((
-                DirectoryKind::TwoPhase,
-                MetricsUpdate::Set(dir.xids.len() as u64),
-            ));
+            self.pending_directory_entries
+                .push((DirectoryKind::TwoPhase, dir.xids.len()));
             Bytes::from(TwoPhaseDirectoryV17::ser(&dir)?)
         } else {
             let xid: u32 = u32::try_from(xid)?;
@@ -2200,10 +2030,8 @@ impl DatadirModification<'_> {
             if !dir.xids.remove(&xid) {
                 warn!("twophase file for xid {} does not exist", xid);
             }
-            self.pending_directory_entries.push((
-                DirectoryKind::TwoPhase,
-                MetricsUpdate::Set(dir.xids.len() as u64),
-            ));
+            self.pending_directory_entries
+                .push((DirectoryKind::TwoPhase, dir.xids.len()));
             Bytes::from(TwoPhaseDirectory::ser(&dir)?)
         };
         self.put(TWOPHASEDIR_KEY, Value::Image(newdirbuf));
@@ -2319,7 +2147,7 @@ impl DatadirModification<'_> {
         }
 
         for (kind, count) in std::mem::take(&mut self.pending_directory_entries) {
-            writer.update_directory_entries_count(kind, count);
+            writer.update_directory_entries_count(kind, count as u64);
         }
 
         Ok(())
@@ -2405,7 +2233,7 @@ impl DatadirModification<'_> {
         }
 
         for (kind, count) in std::mem::take(&mut self.pending_directory_entries) {
-            writer.update_directory_entries_count(kind, count);
+            writer.update_directory_entries_count(kind, count as u64);
         }
 
         self.pending_metadata_bytes = 0;
@@ -2469,22 +2297,6 @@ impl DatadirModification<'_> {
         self.tline.get(key, lsn, ctx).await
     }
 
-    /// Get a key from the sparse keyspace. Automatically converts the missing key error
-    /// and the empty value into None.
-    async fn sparse_get(
-        &self,
-        key: Key,
-        ctx: &RequestContext,
-    ) -> Result<Option<Bytes>, PageReconstructError> {
-        let val = self.get(key, ctx).await;
-        match val {
-            Ok(val) if val.is_empty() => Ok(None),
-            Ok(val) => Ok(Some(val)),
-            Err(PageReconstructError::MissingKey(_)) => Ok(None),
-            Err(e) => Err(e),
-        }
-    }
-
     fn put(&mut self, key: Key, val: Value) {
         if Self::is_data_key(&key) {
             self.put_data(key.to_compact(), val)
@@ -2567,23 +2379,6 @@ impl Version<'_> {
         }
     }
 
-    /// Get a key from the sparse keyspace. Automatically converts the missing key error
-    /// and the empty value into None.
-    async fn sparse_get(
-        &self,
-        timeline: &Timeline,
-        key: Key,
-        ctx: &RequestContext,
-    ) -> Result<Option<Bytes>, PageReconstructError> {
-        let val = self.get(timeline, key, ctx).await;
-        match val {
-            Ok(val) if val.is_empty() => Ok(None),
-            Ok(val) => Ok(Some(val)),
-            Err(PageReconstructError::MissingKey(_)) => Ok(None),
-            Err(e) => Err(e),
-        }
-    }
-
     fn get_lsn(&self) -> Lsn {
         match self {
             Version::Lsn(lsn) => *lsn,
@@ -2643,7 +2438,6 @@ pub(crate) enum DirectoryKind {
     Rel,
     AuxFiles,
     SlruSegment(SlruKind),
-    RelV2,
 }
 
 impl DirectoryKind {

pageserver/src/tenant.rs

@@ -40,8 +40,6 @@ use remote_timeline_client::manifest::{
 use remote_timeline_client::UploadQueueNotReadyError;
 use remote_timeline_client::FAILED_REMOTE_OP_RETRIES;
 use remote_timeline_client::FAILED_UPLOAD_WARN_THRESHOLD;
-use secondary::heatmap::HeatMapTenant;
-use secondary::heatmap::HeatMapTimeline;
 use std::collections::BTreeMap;
 use std::fmt;
 use std::future::Future;
@@ -57,7 +55,6 @@ use timeline::offload::OffloadError;
 use timeline::CompactFlags;
 use timeline::CompactOptions;
 use timeline::CompactionError;
-use timeline::PreviousHeatmap;
 use timeline::ShutdownMode;
 use tokio::io::BufReader;
 use tokio::sync::watch;
@@ -265,7 +262,6 @@ struct TimelinePreload {
     timeline_id: TimelineId,
     client: RemoteTimelineClient,
     index_part: Result<MaybeDeletedIndexPart, DownloadError>,
-    previous_heatmap: Option<PreviousHeatmap>,
 }
 
 pub(crate) struct TenantPreload {
@@ -1132,7 +1128,6 @@ impl Tenant {
         resources: TimelineResources,
         mut index_part: IndexPart,
         metadata: TimelineMetadata,
-        previous_heatmap: Option<PreviousHeatmap>,
         ancestor: Option<Arc<Timeline>>,
         cause: LoadTimelineCause,
         ctx: &RequestContext,
@@ -1163,7 +1158,6 @@ impl Tenant {
         let timeline = self.create_timeline_struct(
             timeline_id,
             &metadata,
-            previous_heatmap,
             ancestor.clone(),
             resources,
             CreateTimelineCause::Load,
@@ -1563,18 +1557,8 @@ impl Tenant {
             }
         }
 
-        // TODO(vlad): Could go to S3 if the secondary is freezing cold and hasn't even
-        // pulled the first heatmap. Not entirely necessary since the storage controller
-        // will kick the secondary in any case and cause a download.
-        let maybe_heatmap_at = self.read_on_disk_heatmap().await;
-
         let timelines = self
-            .load_timelines_metadata(
-                remote_timeline_ids,
-                remote_storage,
-                maybe_heatmap_at,
-                cancel,
-            )
+            .load_timelines_metadata(remote_timeline_ids, remote_storage, cancel)
             .await?;
 
         Ok(TenantPreload {
@@ -1587,26 +1571,6 @@ impl Tenant {
         })
     }
 
-    async fn read_on_disk_heatmap(&self) -> Option<(HeatMapTenant, std::time::Instant)> {
-        let on_disk_heatmap_path = self.conf.tenant_heatmap_path(&self.tenant_shard_id);
-        match tokio::fs::read_to_string(on_disk_heatmap_path).await {
-            Ok(heatmap) => match serde_json::from_str::<HeatMapTenant>(&heatmap) {
-                Ok(heatmap) => Some((heatmap, std::time::Instant::now())),
-                Err(err) => {
-                    error!("Failed to deserialize old heatmap: {err}");
-                    None
-                }
-            },
-            Err(err) => match err.kind() {
-                std::io::ErrorKind::NotFound => None,
-                _ => {
-                    error!("Unexpected IO error reading old heatmap: {err}");
-                    None
-                }
-            },
-        }
-    }
-
     ///
     /// Background task that downloads all data for a tenant and brings it to Active state.
     ///
@@ -1694,10 +1658,7 @@ impl Tenant {
             match index_part {
                 MaybeDeletedIndexPart::IndexPart(index_part) => {
                     timeline_ancestors.insert(timeline_id, index_part.metadata.clone());
-                    remote_index_and_client.insert(
-                        timeline_id,
-                        (index_part, preload.client, preload.previous_heatmap),
-                    );
+                    remote_index_and_client.insert(timeline_id, (index_part, preload.client));
                 }
                 MaybeDeletedIndexPart::Deleted(index_part) => {
                     info!(
@@ -1716,7 +1677,7 @@ impl Tenant {
         // layer file.
         let sorted_timelines = tree_sort_timelines(timeline_ancestors, |m| m.ancestor_timeline())?;
         for (timeline_id, remote_metadata) in sorted_timelines {
-            let (index_part, remote_client, previous_heatmap) = remote_index_and_client
+            let (index_part, remote_client) = remote_index_and_client
                 .remove(&timeline_id)
                 .expect("just put it in above");
 
@@ -1736,7 +1697,6 @@ impl Tenant {
                     timeline_id,
                     index_part,
                     remote_metadata,
-                    previous_heatmap,
                     self.get_timeline_resources_for(remote_client),
                     LoadTimelineCause::Attach,
                     ctx,
@@ -1886,13 +1846,11 @@ impl Tenant {
     }
 
     #[instrument(skip_all, fields(timeline_id=%timeline_id))]
-    #[allow(clippy::too_many_arguments)]
     async fn load_remote_timeline(
         self: &Arc<Self>,
         timeline_id: TimelineId,
         index_part: IndexPart,
         remote_metadata: TimelineMetadata,
-        previous_heatmap: Option<PreviousHeatmap>,
         resources: TimelineResources,
         cause: LoadTimelineCause,
         ctx: &RequestContext,
@@ -1922,7 +1880,6 @@ impl Tenant {
             resources,
             index_part,
             remote_metadata,
-            previous_heatmap,
             ancestor,
             cause,
             ctx,
@@ -1934,29 +1891,14 @@ impl Tenant {
         self: &Arc<Tenant>,
         timeline_ids: HashSet<TimelineId>,
         remote_storage: &GenericRemoteStorage,
-        heatmap: Option<(HeatMapTenant, std::time::Instant)>,
         cancel: CancellationToken,
     ) -> anyhow::Result<HashMap<TimelineId, TimelinePreload>> {
-        let mut timeline_heatmaps = heatmap.map(|h| (h.0.into_timelines_index(), h.1));
-
         let mut part_downloads = JoinSet::new();
         for timeline_id in timeline_ids {
             let cancel_clone = cancel.clone();
-
-            let previous_timeline_heatmap = timeline_heatmaps.as_mut().and_then(|hs| {
-                hs.0.remove(&timeline_id).map(|h| PreviousHeatmap::Active {
-                    heatmap: h,
-                    read_at: hs.1,
-                })
-            });
             part_downloads.spawn(
-                self.load_timeline_metadata(
-                    timeline_id,
-                    remote_storage.clone(),
-                    previous_timeline_heatmap,
-                    cancel_clone,
-                )
-                .instrument(info_span!("download_index_part", %timeline_id)),
+                self.load_timeline_metadata(timeline_id, remote_storage.clone(), cancel_clone)
+                    .instrument(info_span!("download_index_part", %timeline_id)),
             );
         }
 
@@ -2004,7 +1946,6 @@ impl Tenant {
         self: &Arc<Tenant>,
         timeline_id: TimelineId,
         remote_storage: GenericRemoteStorage,
-        previous_heatmap: Option<PreviousHeatmap>,
         cancel: CancellationToken,
     ) -> impl Future<Output = TimelinePreload> {
         let client = self.build_timeline_client(timeline_id, remote_storage);
@@ -2020,7 +1961,6 @@ impl Tenant {
                 client,
                 timeline_id,
                 index_part,
-                previous_heatmap,
             }
         }
     }
@@ -2132,12 +2072,7 @@ impl Tenant {
             })?;
 
         let timeline_preload = self
-            .load_timeline_metadata(
-                timeline_id,
-                self.remote_storage.clone(),
-                None,
-                cancel.clone(),
-            )
+            .load_timeline_metadata(timeline_id, self.remote_storage.clone(), cancel.clone())
             .await;
 
         let index_part = match timeline_preload.index_part {
@@ -2171,7 +2106,6 @@ impl Tenant {
             timeline_id,
             index_part,
             remote_metadata,
-            None,
             timeline_resources,
             LoadTimelineCause::Unoffload,
             &ctx,
@@ -2887,7 +2821,7 @@ impl Tenant {
         };
         let metadata = index_part.metadata.clone();
         self
-            .load_remote_timeline(timeline_id, index_part, metadata, None, resources, LoadTimelineCause::ImportPgdata{
+            .load_remote_timeline(timeline_id, index_part, metadata, resources, LoadTimelineCause::ImportPgdata{
                 create_guard: timeline_create_guard, activate, }, &ctx)
             .await?
             .ready_to_activate()
@@ -3924,13 +3858,6 @@ impl Tenant {
             .unwrap_or(self.conf.default_tenant_conf.compaction_threshold)
     }
 
-    pub fn get_rel_size_v2_enabled(&self) -> bool {
-        let tenant_conf = self.tenant_conf.load().tenant_conf.clone();
-        tenant_conf
-            .rel_size_v2_enabled
-            .unwrap_or(self.conf.default_tenant_conf.rel_size_v2_enabled)
-    }
-
     pub fn get_compaction_upper_limit(&self) -> usize {
         let tenant_conf = self.tenant_conf.load().tenant_conf.clone();
         tenant_conf
@@ -4103,7 +4030,6 @@ impl Tenant {
         &self,
         new_timeline_id: TimelineId,
         new_metadata: &TimelineMetadata,
-        previous_heatmap: Option<PreviousHeatmap>,
         ancestor: Option<Arc<Timeline>>,
         resources: TimelineResources,
         cause: CreateTimelineCause,
@@ -4127,7 +4053,6 @@ impl Tenant {
             self.conf,
             Arc::clone(&self.tenant_conf),
             new_metadata,
-            previous_heatmap,
             ancestor,
             new_timeline_id,
             self.tenant_shard_id,
@@ -4770,24 +4695,24 @@ impl Tenant {
         // We check it against both the planned GC cutoff stored in 'gc_info',
         // and the 'latest_gc_cutoff' of the last GC that was performed.  The
         // planned GC cutoff in 'gc_info' is normally larger than
-        // 'applied_gc_cutoff_lsn', but beware of corner cases like if you just
+        // 'latest_gc_cutoff_lsn', but beware of corner cases like if you just
         // changed the GC settings for the tenant to make the PITR window
         // larger, but some of the data was already removed by an earlier GC
         // iteration.
 
         // check against last actual 'latest_gc_cutoff' first
-        let applied_gc_cutoff_lsn = src_timeline.get_applied_gc_cutoff_lsn();
+        let latest_gc_cutoff_lsn = src_timeline.get_latest_gc_cutoff_lsn();
         {
             let gc_info = src_timeline.gc_info.read().unwrap();
             let planned_cutoff = gc_info.min_cutoff();
             if gc_info.lsn_covered_by_lease(start_lsn) {
-                tracing::info!("skipping comparison of {start_lsn} with gc cutoff {} and planned gc cutoff {planned_cutoff} due to lsn lease", *applied_gc_cutoff_lsn);
+                tracing::info!("skipping comparison of {start_lsn} with gc cutoff {} and planned gc cutoff {planned_cutoff} due to lsn lease", *latest_gc_cutoff_lsn);
             } else {
                 src_timeline
-                    .check_lsn_is_in_scope(start_lsn, &applied_gc_cutoff_lsn)
+                    .check_lsn_is_in_scope(start_lsn, &latest_gc_cutoff_lsn)
                     .context(format!(
                         "invalid branch start lsn: less than latest GC cutoff {}",
-                        *applied_gc_cutoff_lsn,
+                        *latest_gc_cutoff_lsn,
                     ))
                     .map_err(CreateTimelineError::AncestorLsn)?;
 
@@ -4826,7 +4751,7 @@ impl Tenant {
             dst_prev,
             Some(src_id),
             start_lsn,
-            *src_timeline.applied_gc_cutoff_lsn.read(), // FIXME: should we hold onto this guard longer?
+            *src_timeline.latest_gc_cutoff_lsn.read(), // FIXME: should we hold onto this guard longer?
             src_timeline.initdb_lsn,
             src_timeline.pg_version,
         );
@@ -5199,7 +5124,6 @@ impl Tenant {
             .create_timeline_struct(
                 new_timeline_id,
                 new_metadata,
-                None,
                 ancestor,
                 resources,
                 CreateTimelineCause::Load,
@@ -5647,7 +5571,7 @@ pub(crate) mod harness {
                 lsn_lease_length_for_ts: Some(tenant_conf.lsn_lease_length_for_ts),
                 timeline_offloading: Some(tenant_conf.timeline_offloading),
                 wal_receiver_protocol_override: tenant_conf.wal_receiver_protocol_override,
-                rel_size_v2_enabled: Some(tenant_conf.rel_size_v2_enabled),
+                rel_size_v2_enabled: tenant_conf.rel_size_v2_enabled,
                 gc_compaction_enabled: Some(tenant_conf.gc_compaction_enabled),
                 gc_compaction_initial_threshold_kb: Some(
                     tenant_conf.gc_compaction_initial_threshold_kb,
@@ -6206,8 +6130,8 @@ mod tests {
         make_some_layers(tline.as_ref(), Lsn(0x20), &ctx).await?;
 
         repo.gc_iteration(Some(TIMELINE_ID), 0x10, Duration::ZERO)?;
-        let applied_gc_cutoff_lsn = tline.get_applied_gc_cutoff_lsn();
-        assert!(*applied_gc_cutoff_lsn > Lsn(0x25));
+        let latest_gc_cutoff_lsn = tline.get_latest_gc_cutoff_lsn();
+        assert!(*latest_gc_cutoff_lsn > Lsn(0x25));
         match tline.get(*TEST_KEY, Lsn(0x25)) {
             Ok(_) => panic!("request for page should have failed"),
             Err(err) => assert!(err.to_string().contains("not found at")),
@@ -8503,7 +8427,7 @@ mod tests {
             .await?;
         {
             tline
-                .applied_gc_cutoff_lsn
+                .latest_gc_cutoff_lsn
                 .lock_for_write()
                 .store_and_unlock(Lsn(0x30))
                 .wait()
@@ -8611,7 +8535,7 @@ mod tests {
         // increase GC horizon and compact again
         {
             tline
-                .applied_gc_cutoff_lsn
+                .latest_gc_cutoff_lsn
                 .lock_for_write()
                 .store_and_unlock(Lsn(0x40))
                 .wait()
@@ -8779,8 +8703,8 @@ mod tests {
 
         // Force set disk consistent lsn so we can get the cutoff at `end_lsn`.
         info!(
-            "applied_gc_cutoff_lsn: {}",
-            *timeline.get_applied_gc_cutoff_lsn()
+            "latest_gc_cutoff_lsn: {}",
+            *timeline.get_latest_gc_cutoff_lsn()
         );
         timeline.force_set_disk_consistent_lsn(end_lsn);
 
@@ -8806,7 +8730,7 @@ mod tests {
 
         // Make lease on a already GC-ed LSN.
         // 0/80 does not have a valid lease + is below latest_gc_cutoff
-        assert!(Lsn(0x80) < *timeline.get_applied_gc_cutoff_lsn());
+        assert!(Lsn(0x80) < *timeline.get_latest_gc_cutoff_lsn());
         timeline
             .init_lsn_lease(Lsn(0x80), timeline.get_lsn_lease_length(), &ctx)
             .expect_err("lease request on GC-ed LSN should fail");
@@ -8997,7 +8921,7 @@ mod tests {
         };
         {
             tline
-                .applied_gc_cutoff_lsn
+                .latest_gc_cutoff_lsn
                 .lock_for_write()
                 .store_and_unlock(Lsn(0x30))
                 .wait()
@@ -9084,7 +9008,7 @@ mod tests {
         // increase GC horizon and compact again
         {
             tline
-                .applied_gc_cutoff_lsn
+                .latest_gc_cutoff_lsn
                 .lock_for_write()
                 .store_and_unlock(Lsn(0x40))
                 .wait()
@@ -9537,7 +9461,7 @@ mod tests {
             .await?;
         {
             tline
-                .applied_gc_cutoff_lsn
+                .latest_gc_cutoff_lsn
                 .lock_for_write()
                 .store_and_unlock(Lsn(0x30))
                 .wait()
@@ -9684,7 +9608,7 @@ mod tests {
         // increase GC horizon and compact again
         {
             tline
-                .applied_gc_cutoff_lsn
+                .latest_gc_cutoff_lsn
                 .lock_for_write()
                 .store_and_unlock(Lsn(0x38))
                 .wait()
@@ -9785,7 +9709,7 @@ mod tests {
             .await?;
         {
             tline
-                .applied_gc_cutoff_lsn
+                .latest_gc_cutoff_lsn
                 .lock_for_write()
                 .store_and_unlock(Lsn(0x30))
                 .wait()
@@ -10036,7 +9960,7 @@ mod tests {
 
         {
             parent_tline
-                .applied_gc_cutoff_lsn
+                .latest_gc_cutoff_lsn
                 .lock_for_write()
                 .store_and_unlock(Lsn(0x10))
                 .wait()
@@ -10056,7 +9980,7 @@ mod tests {
 
         {
             branch_tline
-                .applied_gc_cutoff_lsn
+                .latest_gc_cutoff_lsn
                 .lock_for_write()
                 .store_and_unlock(Lsn(0x50))
                 .wait()
@@ -10412,7 +10336,7 @@ mod tests {
 
         {
             tline
-                .applied_gc_cutoff_lsn
+                .latest_gc_cutoff_lsn
                 .lock_for_write()
                 .store_and_unlock(Lsn(0x30))
                 .wait()
@@ -10797,7 +10721,7 @@ mod tests {
             .await?;
         {
             tline
-                .applied_gc_cutoff_lsn
+                .latest_gc_cutoff_lsn
                 .lock_for_write()
                 .store_and_unlock(Lsn(0x30))
                 .wait()
@@ -11048,7 +10972,7 @@ mod tests {
             .await?;
         {
             tline
-                .applied_gc_cutoff_lsn
+                .latest_gc_cutoff_lsn
                 .lock_for_write()
                 .store_and_unlock(Lsn(0x30))
                 .wait()

pageserver/src/tenant/config.rs

@@ -485,9 +485,7 @@ impl TenantConfOpt {
             wal_receiver_protocol_override: self
                 .wal_receiver_protocol_override
                 .or(global_conf.wal_receiver_protocol_override),
-            rel_size_v2_enabled: self
-                .rel_size_v2_enabled
-                .unwrap_or(global_conf.rel_size_v2_enabled),
+            rel_size_v2_enabled: self.rel_size_v2_enabled.or(global_conf.rel_size_v2_enabled),
             gc_compaction_enabled: self
                 .gc_compaction_enabled
                 .unwrap_or(global_conf.gc_compaction_enabled),

pageserver/src/tenant/metadata.rs

@@ -130,10 +130,7 @@ struct TimelineMetadataBodyV2 {
     prev_record_lsn: Option<Lsn>,
     ancestor_timeline: Option<TimelineId>,
     ancestor_lsn: Lsn,
-
-    // The LSN at which GC was last executed.  Synonym of [`Timeline::applied_gc_cutoff_lsn`].
     latest_gc_cutoff_lsn: Lsn,
-
     initdb_lsn: Lsn,
     pg_version: u32,
 }

pageserver/src/tenant/secondary/heatmap.rs

@@ -1,4 +1,4 @@
-use std::{collections::HashMap, time::SystemTime};
+use std::time::SystemTime;
 
 use crate::tenant::{remote_timeline_client::index::LayerFileMetadata, storage_layer::LayerName};
 
@@ -8,7 +8,7 @@ use serde_with::{serde_as, DisplayFromStr, TimestampSeconds};
 use utils::{generation::Generation, id::TimelineId};
 
 #[derive(Serialize, Deserialize)]
-pub(crate) struct HeatMapTenant {
+pub(super) struct HeatMapTenant {
     /// Generation of the attached location that uploaded the heatmap: this is not required
     /// for correctness, but acts as a hint to secondary locations in order to detect thrashing
     /// in the unlikely event that two attached locations are both uploading conflicting heatmaps.
@@ -25,17 +25,8 @@ pub(crate) struct HeatMapTenant {
     pub(super) upload_period_ms: Option<u128>,
 }
 
-impl HeatMapTenant {
-    pub(crate) fn into_timelines_index(self) -> HashMap<TimelineId, HeatMapTimeline> {
-        self.timelines
-            .into_iter()
-            .map(|htl| (htl.timeline_id, htl))
-            .collect()
-    }
-}
-
 #[serde_as]
-#[derive(Serialize, Deserialize, Clone)]
+#[derive(Serialize, Deserialize)]
 pub(crate) struct HeatMapTimeline {
     #[serde_as(as = "DisplayFromStr")]
     pub(crate) timeline_id: TimelineId,
@@ -44,13 +35,13 @@ pub(crate) struct HeatMapTimeline {
 }
 
 #[serde_as]
-#[derive(Serialize, Deserialize, Clone)]
+#[derive(Serialize, Deserialize)]
 pub(crate) struct HeatMapLayer {
     pub(crate) name: LayerName,
     pub(crate) metadata: LayerFileMetadata,
 
     #[serde_as(as = "TimestampSeconds<i64>")]
-    pub(crate) access_time: SystemTime,
+    pub(super) access_time: SystemTime,
     // TODO: an actual 'heat' score that would let secondary locations prioritize downloading
     // the hottest layers, rather than trying to simply mirror whatever layers are on-disk on the primary.
 }

pageserver/src/tenant/size.rs

@@ -394,7 +394,7 @@ pub(super) async fn gather_inputs(
             ancestor_lsn,
             last_record: last_record_lsn,
             // this is not used above, because it might not have updated recently enough
-            latest_gc_cutoff: *timeline.get_applied_gc_cutoff_lsn(),
+            latest_gc_cutoff: *timeline.get_latest_gc_cutoff_lsn(),
             next_pitr_cutoff,
             retention_param_cutoff,
             lease_points,

pageserver/src/tenant/storage_layer/layer.rs

@@ -136,22 +136,6 @@ pub(crate) fn local_layer_path(
     }
 }
 
-pub(crate) enum LastEviction {
-    Never,
-    At(std::time::Instant),
-    Evicting,
-}
-
-impl LastEviction {
-    pub(crate) fn happened_after(&self, timepoint: std::time::Instant) -> bool {
-        match self {
-            LastEviction::Never => false,
-            LastEviction::At(evicted_at) => evicted_at > &timepoint,
-            LastEviction::Evicting => true,
-        }
-    }
-}
-
 impl Layer {
     /// Creates a layer value for a file we know to not be resident.
     pub(crate) fn for_evicted(
@@ -421,17 +405,6 @@ impl Layer {
         self.0.metadata()
     }
 
-    pub(crate) fn last_evicted_at(&self) -> LastEviction {
-        match self.0.last_evicted_at.try_lock() {
-            Ok(lock) => match *lock {
-                None => LastEviction::Never,
-                Some(at) => LastEviction::At(at),
-            },
-            Err(std::sync::TryLockError::WouldBlock) => LastEviction::Evicting,
-            Err(std::sync::TryLockError::Poisoned(p)) => panic!("Lock poisoned: {p}"),
-        }
-    }
-
     pub(crate) fn get_timeline_id(&self) -> Option<TimelineId> {
         self.0
             .timeline
@@ -683,9 +656,7 @@ struct LayerInner {
 
     /// When the Layer was last evicted but has not been downloaded since.
     ///
-    /// This is used for skipping evicted layers from the previous heatmap (see
-    /// `[Timeline::generate_heatmap]`) and for updating metrics
-    /// (see [`LayerImplMetrics::redownload_after`]).
+    /// This is used solely for updating metrics. See [`LayerImplMetrics::redownload_after`].
     last_evicted_at: std::sync::Mutex<Option<std::time::Instant>>,
 
     #[cfg(test)]

pageserver/src/tenant/timeline.rs

@@ -117,7 +117,7 @@ use pageserver_api::config::tenant_conf_defaults::DEFAULT_PITR_INTERVAL;
 use crate::config::PageServerConf;
 use crate::keyspace::{KeyPartitioning, KeySpace};
 use crate::metrics::{TimelineMetrics, DELTAS_PER_READ_GLOBAL, LAYERS_PER_READ_GLOBAL};
-use crate::pgdatadir_mapping::{CalculateLogicalSizeError, MetricsUpdate};
+use crate::pgdatadir_mapping::CalculateLogicalSizeError;
 use crate::tenant::config::TenantConfOpt;
 use pageserver_api::reltag::RelTag;
 use pageserver_api::shard::ShardIndex;
@@ -150,15 +150,16 @@ use super::{
     config::TenantConf, storage_layer::LayerVisibilityHint, upload_queue::NotInitialized,
     MaybeOffloaded,
 };
-use super::{
-    debug_assert_current_span_has_tenant_and_timeline_id, AttachedTenantConf, HeatMapTimeline,
-};
+use super::{debug_assert_current_span_has_tenant_and_timeline_id, AttachedTenantConf};
 use super::{remote_timeline_client::index::IndexPart, storage_layer::LayerFringe};
 use super::{
     remote_timeline_client::RemoteTimelineClient, remote_timeline_client::WaitCompletionError,
     storage_layer::ReadableLayer,
 };
-use super::{secondary::heatmap::HeatMapLayer, GcError};
+use super::{
+    secondary::heatmap::{HeatMapLayer, HeatMapTimeline},
+    GcError,
+};
 
 #[cfg(test)]
 use pageserver_api::value::Value;
@@ -327,7 +328,6 @@ pub struct Timeline {
     // in `crate::page_service` writes these metrics.
     pub(crate) query_metrics: crate::metrics::SmgrQueryTimePerTimeline,
 
-    directory_metrics_inited: [AtomicBool; DirectoryKind::KINDS_NUM],
     directory_metrics: [AtomicU64; DirectoryKind::KINDS_NUM],
 
     /// Ensures layers aren't frozen by checkpointer between
@@ -352,11 +352,8 @@ pub struct Timeline {
     /// to be notified when layer flushing has finished, subscribe to the layer_flush_done channel
     layer_flush_done_tx: tokio::sync::watch::Sender<(u64, Result<(), FlushLayerError>)>,
 
-    // The LSN at which we have executed GC: whereas [`Self::gc_info`] records the LSN at which
-    // we _intend_ to GC (i.e. the PITR cutoff), this LSN records where we actually last did it.
-    // Because PITR interval is mutable, it's possible for this LSN to be earlier or later than
-    // the planned GC cutoff.
-    pub applied_gc_cutoff_lsn: Rcu<Lsn>,
+    // Needed to ensure that we can't create a branch at a point that was already garbage collected
+    pub latest_gc_cutoff_lsn: Rcu<Lsn>,
 
     pub(crate) gc_compaction_layer_update_lock: tokio::sync::RwLock<()>,
 
@@ -465,16 +462,6 @@ pub struct Timeline {
 
     /// If Some, collects GetPage metadata for an ongoing PageTrace.
     pub(crate) page_trace: ArcSwapOption<Sender<PageTraceEvent>>,
-
-    previous_heatmap: ArcSwapOption<PreviousHeatmap>,
-}
-
-pub(crate) enum PreviousHeatmap {
-    Active {
-        heatmap: HeatMapTimeline,
-        read_at: std::time::Instant,
-    },
-    Obsolete,
 }
 
 pub type TimelineDeleteProgress = Arc<tokio::sync::Mutex<DeleteTimelineFlow>>;
@@ -1090,15 +1077,9 @@ impl Timeline {
         (history, gc_info.within_ancestor_pitr)
     }
 
-    /// Read timeline's GC cutoff: this is the LSN at which GC has started to happen
-    pub(crate) fn get_applied_gc_cutoff_lsn(&self) -> RcuReadGuard<Lsn> {
-        self.applied_gc_cutoff_lsn.read()
-    }
-
-    /// Read timeline's planned GC cutoff: this is the logical end of history that users
-    /// are allowed to read (based on configured PITR), even if physically we have more history.
-    pub(crate) fn get_gc_cutoff_lsn(&self) -> Lsn {
-        self.gc_info.read().unwrap().cutoffs.time
+    /// Lock and get timeline's GC cutoff
+    pub(crate) fn get_latest_gc_cutoff_lsn(&self) -> RcuReadGuard<Lsn> {
+        self.latest_gc_cutoff_lsn.read()
     }
 
     /// Look up given page version.
@@ -1606,7 +1587,7 @@ impl Timeline {
                     };
 
                     if init || validate {
-                        let latest_gc_cutoff_lsn = self.get_applied_gc_cutoff_lsn();
+                        let latest_gc_cutoff_lsn = self.get_latest_gc_cutoff_lsn();
                         if lsn < *latest_gc_cutoff_lsn {
                             bail!("tried to request an lsn lease for an lsn below the latest gc cutoff. requested at {} gc cutoff {}", lsn, *latest_gc_cutoff_lsn);
                         }
@@ -2356,14 +2337,6 @@ impl Timeline {
             .unwrap_or(self.conf.default_tenant_conf.compaction_threshold)
     }
 
-    pub(crate) fn get_rel_size_v2_enabled(&self) -> bool {
-        let tenant_conf = self.tenant_conf.load();
-        tenant_conf
-            .tenant_conf
-            .rel_size_v2_enabled
-            .unwrap_or(self.conf.default_tenant_conf.rel_size_v2_enabled)
-    }
-
     fn get_compaction_upper_limit(&self) -> usize {
         let tenant_conf = self.tenant_conf.load();
         tenant_conf
@@ -2586,7 +2559,6 @@ impl Timeline {
         conf: &'static PageServerConf,
         tenant_conf: Arc<ArcSwap<AttachedTenantConf>>,
         metadata: &TimelineMetadata,
-        previous_heatmap: Option<PreviousHeatmap>,
         ancestor: Option<Arc<Timeline>>,
         timeline_id: TimelineId,
         tenant_shard_id: TenantShardId,
@@ -2673,7 +2645,6 @@ impl Timeline {
                 ),
 
                 directory_metrics: array::from_fn(|_| AtomicU64::new(0)),
-                directory_metrics_inited: array::from_fn(|_| AtomicBool::new(false)),
 
                 flush_loop_state: Mutex::new(FlushLoopState::NotStarted),
 
@@ -2688,7 +2659,7 @@ impl Timeline {
                     LastImageLayerCreationStatus::default(),
                 )),
 
-                applied_gc_cutoff_lsn: Rcu::new(metadata.latest_gc_cutoff_lsn()),
+                latest_gc_cutoff_lsn: Rcu::new(metadata.latest_gc_cutoff_lsn()),
                 initdb_lsn: metadata.initdb_lsn(),
 
                 current_logical_size: if disk_consistent_lsn.is_valid() {
@@ -2750,8 +2721,6 @@ impl Timeline {
                 create_idempotency,
 
                 page_trace: Default::default(),
-
-                previous_heatmap: ArcSwapOption::from_pointee(previous_heatmap),
             };
 
             result.repartition_threshold =
@@ -3440,42 +3409,8 @@ impl Timeline {
         }
     }
 
-    pub(crate) fn update_directory_entries_count(&self, kind: DirectoryKind, count: MetricsUpdate) {
-        // TODO: this directory metrics is not correct -- we could have multiple reldirs in the system
-        // for each of the database, but we only store one value, and therefore each pgdirmodification
-        // would overwrite the previous value if they modify different databases.
-
-        match count {
-            MetricsUpdate::Set(count) => {
-                self.directory_metrics[kind.offset()].store(count, AtomicOrdering::Relaxed);
-                self.directory_metrics_inited[kind.offset()].store(true, AtomicOrdering::Relaxed);
-            }
-            MetricsUpdate::Add(count) => {
-                // TODO: these operations are not atomic; but we only have one writer to the metrics, so
-                // it's fine.
-                if self.directory_metrics_inited[kind.offset()].load(AtomicOrdering::Relaxed) {
-                    // The metrics has been initialized with `MetricsUpdate::Set` before, so we can add/sub
-                    // the value reliably.
-                    self.directory_metrics[kind.offset()].fetch_add(count, AtomicOrdering::Relaxed);
-                }
-                // Otherwise, ignore this update
-            }
-            MetricsUpdate::Sub(count) => {
-                // TODO: these operations are not atomic; but we only have one writer to the metrics, so
-                // it's fine.
-                if self.directory_metrics_inited[kind.offset()].load(AtomicOrdering::Relaxed) {
-                    // The metrics has been initialized with `MetricsUpdate::Set` before.
-                    // The operation could overflow so we need to normalize the value.
-                    let prev_val =
-                        self.directory_metrics[kind.offset()].load(AtomicOrdering::Relaxed);
-                    let res = prev_val.saturating_sub(count);
-                    self.directory_metrics[kind.offset()].store(res, AtomicOrdering::Relaxed);
-                }
-                // Otherwise, ignore this update
-            }
-        };
-
-        // TODO: remove this, there's no place in the code that updates this aux metrics.
+    pub(crate) fn update_directory_entries_count(&self, kind: DirectoryKind, count: u64) {
+        self.directory_metrics[kind.offset()].store(count, AtomicOrdering::Relaxed);
         let aux_metric =
             self.directory_metrics[DirectoryKind::AuxFiles.offset()].load(AtomicOrdering::Relaxed);
 
@@ -3524,52 +3459,12 @@ impl Timeline {
 
         let guard = self.layers.read().await;
 
-        // Firstly, if there's any heatmap left over from when this location
-        // was a secondary, take that into account. Keep layers that are:
-        // * present in the layer map
-        // * visible
-        // * non-resident
-        // * not evicted since we read the heatmap
-        //
-        // Without this, a new cold, attached location would clobber the previous
-        // heatamp.
-        let previous_heatmap = self.previous_heatmap.load();
-        let visible_non_resident = match previous_heatmap.as_deref() {
-            Some(PreviousHeatmap::Active { heatmap, read_at }) => {
-                Some(heatmap.layers.iter().filter_map(|hl| {
-                    let desc: PersistentLayerDesc = hl.name.clone().into();
-                    let layer = guard.try_get_from_key(&desc.key())?;
-
-                    if layer.visibility() == LayerVisibilityHint::Covered {
-                        return None;
-                    }
-
-                    if layer.is_likely_resident() {
-                        return None;
-                    }
-
-                    if layer.last_evicted_at().happened_after(*read_at) {
-                        return None;
-                    }
-
-                    Some((desc, hl.metadata.clone(), hl.access_time))
-                }))
-            }
-            Some(PreviousHeatmap::Obsolete) => None,
-            None => None,
-        };
-
-        // Secondly, all currently visible, resident layers are included.
         let resident = guard.likely_resident_layers().filter_map(|layer| {
             match layer.visibility() {
                 LayerVisibilityHint::Visible => {
                     // Layer is visible to one or more read LSNs: elegible for inclusion in layer map
                     let last_activity_ts = layer.latest_activity();
-                    Some((
-                        layer.layer_desc().clone(),
-                        layer.metadata(),
-                        last_activity_ts,
-                    ))
+                    Some((layer.layer_desc(), layer.metadata(), last_activity_ts))
                 }
                 LayerVisibilityHint::Covered => {
                     // Layer is resident but unlikely to be read: not elegible for inclusion in heatmap.
@@ -3578,18 +3473,7 @@ impl Timeline {
             }
         });
 
-        let mut layers = match visible_non_resident {
-            Some(non_resident) => {
-                let mut non_resident = non_resident.peekable();
-                if non_resident.peek().is_none() {
-                    self.previous_heatmap
-                        .store(Some(PreviousHeatmap::Obsolete.into()));
-                }
-
-                non_resident.chain(resident).collect::<Vec<_>>()
-            }
-            None => resident.collect::<Vec<_>>(),
-        };
+        let mut layers = resident.collect::<Vec<_>>();
 
         // Sort layers in order of which to download first.  For a large set of layers to download, we
         // want to prioritize those layers which are most likely to still be in the resident many minutes
@@ -3693,9 +3577,7 @@ impl Timeline {
             // space. If that's not the case, we had at least one key encounter a gap in the image layer
             // and stop the search as a result of that.
             let mut removed = keyspace.remove_overlapping_with(&image_covered_keyspace);
-            // Do not fire missing key error and end early for sparse keys. Note that we hava already removed
-            // non-inherited keyspaces before, so we can safely do a full `SPARSE_RANGE` remove instead of
-            // figuring out what is the inherited key range and do a fine-grained pruning.
+            // Do not fire missing key error for sparse keys.
             removed.remove_overlapping_with(&KeySpace {
                 ranges: vec![SPARSE_RANGE],
             });
@@ -3780,7 +3662,7 @@ impl Timeline {
         // the timeline, then it will remove layers that are required for fulfilling
         // the current get request (read-path cannot "look back" and notice the new
         // image layer).
-        let _gc_cutoff_holder = timeline.get_applied_gc_cutoff_lsn();
+        let _gc_cutoff_holder = timeline.get_latest_gc_cutoff_lsn();
 
         // See `compaction::compact_with_gc` for why we need this.
         let _guard = timeline.gc_compaction_layer_update_lock.read().await;
@@ -4467,7 +4349,7 @@ impl Timeline {
         let update = crate::tenant::metadata::MetadataUpdate::new(
             disk_consistent_lsn,
             ondisk_prev_record_lsn,
-            *self.applied_gc_cutoff_lsn.read(),
+            *self.latest_gc_cutoff_lsn.read(),
         );
 
         fail_point!("checkpoint-before-saving-metadata", |x| bail!(
@@ -4670,74 +4552,71 @@ impl Timeline {
         let mut wrote_keys = false;
 
         let mut key_request_accum = KeySpaceAccum::new();
-        for range in &partition.ranges {
-            let mut key = range.start;
-            while key < range.end {
-                // Decide whether to retain this key: usually we do, but sharded tenants may
-                // need to drop keys that don't belong to them.  If we retain the key, add it
-                // to `key_request_accum` for later issuing a vectored get
-                if self.shard_identity.is_key_disposable(&key) {
-                    debug!(
-                        "Dropping key {} during compaction (it belongs on shard {:?})",
-                        key,
-                        self.shard_identity.get_shard_number(&key)
-                    );
-                } else {
-                    key_request_accum.add_key(key);
+
+        for (key, range_end) in partition.iter() {
+            // Decide whether to retain this key: usually we do, but sharded tenants may
+            // need to drop keys that don't belong to them.  If we retain the key, add it
+            // to `key_request_accum` for later issuing a vectored get
+            if self.shard_identity.is_key_disposable(&key) {
+                debug!(
+                    "Dropping key {} during compaction (it belongs on shard {:?})",
+                    key,
+                    self.shard_identity.get_shard_number(&key)
+                );
+            } else {
+                key_request_accum.add_key(key);
+            }
+
+            let last_key_in_range = key.next() == range_end;
+
+            // Maybe flush `key_rest_accum`
+            if key_request_accum.raw_size() >= Timeline::MAX_GET_VECTORED_KEYS
+                || (last_key_in_range && key_request_accum.raw_size() > 0)
+            {
+                let results = self
+                    .get_vectored(
+                        key_request_accum.consume_keyspace(),
+                        lsn,
+                        io_concurrency.clone(),
+                        ctx,
+                    )
+                    .await?;
+
+                if self.cancel.is_cancelled() {
+                    return Err(CreateImageLayersError::Cancelled);
                 }
 
-                let last_key_in_range = key.next() == range.end;
-                key = key.next();
-
-                // Maybe flush `key_rest_accum`
-                if key_request_accum.raw_size() >= Timeline::MAX_GET_VECTORED_KEYS
-                    || (last_key_in_range && key_request_accum.raw_size() > 0)
-                {
-                    let results = self
-                        .get_vectored(
-                            key_request_accum.consume_keyspace(),
-                            lsn,
-                            io_concurrency.clone(),
-                            ctx,
-                        )
-                        .await?;
-
-                    if self.cancel.is_cancelled() {
-                        return Err(CreateImageLayersError::Cancelled);
-                    }
-
-                    for (img_key, img) in results {
-                        let img = match img {
-                            Ok(img) => img,
-                            Err(err) => {
-                                // If we fail to reconstruct a VM or FSM page, we can zero the
-                                // page without losing any actual user data. That seems better
-                                // than failing repeatedly and getting stuck.
-                                //
-                                // We had a bug at one point, where we truncated the FSM and VM
-                                // in the pageserver, but the Postgres didn't know about that
-                                // and continued to generate incremental WAL records for pages
-                                // that didn't exist in the pageserver. Trying to replay those
-                                // WAL records failed to find the previous image of the page.
-                                // This special case allows us to recover from that situation.
-                                // See https://github.com/neondatabase/neon/issues/2601.
-                                //
-                                // Unfortunately we cannot do this for the main fork, or for
-                                // any metadata keys, keys, as that would lead to actual data
-                                // loss.
-                                if img_key.is_rel_fsm_block_key() || img_key.is_rel_vm_block_key() {
-                                    warn!("could not reconstruct FSM or VM key {img_key}, filling with zeros: {err:?}");
-                                    ZERO_PAGE.clone()
-                                } else {
-                                    return Err(CreateImageLayersError::from(err));
-                                }
+                for (img_key, img) in results {
+                    let img = match img {
+                        Ok(img) => img,
+                        Err(err) => {
+                            // If we fail to reconstruct a VM or FSM page, we can zero the
+                            // page without losing any actual user data. That seems better
+                            // than failing repeatedly and getting stuck.
+                            //
+                            // We had a bug at one point, where we truncated the FSM and VM
+                            // in the pageserver, but the Postgres didn't know about that
+                            // and continued to generate incremental WAL records for pages
+                            // that didn't exist in the pageserver. Trying to replay those
+                            // WAL records failed to find the previous image of the page.
+                            // This special case allows us to recover from that situation.
+                            // See https://github.com/neondatabase/neon/issues/2601.
+                            //
+                            // Unfortunately we cannot do this for the main fork, or for
+                            // any metadata keys, keys, as that would lead to actual data
+                            // loss.
+                            if img_key.is_rel_fsm_block_key() || img_key.is_rel_vm_block_key() {
+                                warn!("could not reconstruct FSM or VM key {img_key}, filling with zeros: {err:?}");
+                                ZERO_PAGE.clone()
+                            } else {
+                                return Err(CreateImageLayersError::from(err));
                             }
-                        };
+                        }
+                    };
 
-                        // Write all the keys we just read into our new image layer.
-                        image_layer_writer.put_image(img_key, img, ctx).await?;
-                        wrote_keys = true;
-                    }
+                    // Write all the keys we just read into our new image layer.
+                    image_layer_writer.put_image(img_key, img, ctx).await?;
+                    wrote_keys = true;
                 }
             }
         }
@@ -5695,7 +5574,7 @@ impl Timeline {
                 // PITR interval is set & we didn't look up a timestamp successfully.  Conservatively assume PITR
                 // cannot advance beyond what was already GC'd, and respect space-based retention
                 GcCutoffs {
-                    time: *self.get_applied_gc_cutoff_lsn(),
+                    time: *self.get_latest_gc_cutoff_lsn(),
                     space: space_cutoff,
                 }
             }
@@ -5816,7 +5695,7 @@ impl Timeline {
         let mut result: GcResult = GcResult::default();
 
         // Nothing to GC. Return early.
-        let latest_gc_cutoff = *self.get_applied_gc_cutoff_lsn();
+        let latest_gc_cutoff = *self.get_latest_gc_cutoff_lsn();
         if latest_gc_cutoff >= new_gc_cutoff {
             info!(
                 "Nothing to GC: new_gc_cutoff_lsn {new_gc_cutoff}, latest_gc_cutoff_lsn {latest_gc_cutoff}",
@@ -5830,7 +5709,7 @@ impl Timeline {
         //
         // The GC cutoff should only ever move forwards.
         let waitlist = {
-            let write_guard = self.applied_gc_cutoff_lsn.lock_for_write();
+            let write_guard = self.latest_gc_cutoff_lsn.lock_for_write();
             if *write_guard > new_gc_cutoff {
                 return Err(GcError::BadLsn {
                     why: format!(
@@ -6770,32 +6649,18 @@ fn is_send() {
 
 #[cfg(test)]
 mod tests {
-    use std::sync::Arc;
-
     use pageserver_api::key::Key;
     use pageserver_api::value::Value;
-    use tracing::Instrument;
     use utils::{id::TimelineId, lsn::Lsn};
 
     use crate::tenant::{
         harness::{test_img, TenantHarness},
         layer_map::LayerMap,
-        storage_layer::{Layer, LayerName, LayerVisibilityHint},
+        storage_layer::{Layer, LayerName},
         timeline::{DeltaLayerTestDesc, EvictionError},
-        PreviousHeatmap, Timeline,
+        Timeline,
     };
 
-    use super::HeatMapTimeline;
-
-    fn assert_heatmaps_have_same_layers(lhs: &HeatMapTimeline, rhs: &HeatMapTimeline) {
-        assert_eq!(lhs.layers.len(), rhs.layers.len());
-        let lhs_rhs = lhs.layers.iter().zip(rhs.layers.iter());
-        for (l, r) in lhs_rhs {
-            assert_eq!(l.name, r.name);
-            assert_eq!(l.metadata, r.metadata);
-        }
-    }
-
     #[tokio::test]
     async fn test_heatmap_generation() {
         let harness = TenantHarness::create("heatmap_generation").await.unwrap();
@@ -6869,7 +6734,7 @@ mod tests {
         assert_eq!(heatmap.layers.last().unwrap().name, l0_delta.layer_name());
 
         let mut last_lsn = Lsn::MAX;
-        for layer in &heatmap.layers {
+        for layer in heatmap.layers {
             // Covered layer should be omitted
             assert!(layer.name != covered_delta.layer_name());
 
@@ -6884,144 +6749,6 @@ mod tests {
                 last_lsn = layer_lsn;
             }
         }
-
-        // Evict all the layers and stash the old heatmap in the timeline.
-        // This simulates a migration to a cold secondary location.
-
-        let guard = timeline.layers.read().await;
-        let mut all_layers = Vec::new();
-        let forever = std::time::Duration::from_secs(120);
-        for layer in guard.likely_resident_layers() {
-            all_layers.push(layer.clone());
-            layer.evict_and_wait(forever).await.unwrap();
-        }
-        drop(guard);
-
-        timeline
-            .previous_heatmap
-            .store(Some(Arc::new(PreviousHeatmap::Active {
-                heatmap: heatmap.clone(),
-                read_at: std::time::Instant::now(),
-            })));
-
-        // Generate a new heatmap and assert that it contains the same layers as the old one.
-        let post_migration_heatmap = timeline.generate_heatmap().await.unwrap();
-        assert_heatmaps_have_same_layers(&heatmap, &post_migration_heatmap);
-
-        // Download each layer one by one. Generate the heatmap at each step and check
-        // that it's stable.
-        for layer in all_layers {
-            if layer.visibility() == LayerVisibilityHint::Covered {
-                continue;
-            }
-
-            eprintln!("Downloading {layer} and re-generating heatmap");
-
-            let _resident = layer
-                .download_and_keep_resident()
-                .instrument(tracing::info_span!(
-                    parent: None,
-                    "download_layer",
-                    tenant_id = %timeline.tenant_shard_id.tenant_id,
-                    shard_id = %timeline.tenant_shard_id.shard_slug(),
-                    timeline_id = %timeline.timeline_id
-                ))
-                .await
-                .unwrap();
-
-            let post_download_heatmap = timeline.generate_heatmap().await.unwrap();
-            assert_heatmaps_have_same_layers(&heatmap, &post_download_heatmap);
-        }
-
-        // Everything from the post-migration heatmap is now resident.
-        // Check that we drop it from memory.
-        assert!(matches!(
-            timeline.previous_heatmap.load().as_deref(),
-            Some(PreviousHeatmap::Obsolete)
-        ));
-    }
-
-    #[tokio::test]
-    async fn test_previous_heatmap_obsoletion() {
-        let harness = TenantHarness::create("heatmap_previous_heatmap_obsoletion")
-            .await
-            .unwrap();
-
-        let l0_delta = DeltaLayerTestDesc::new(
-            Lsn(0x20)..Lsn(0x30),
-            Key::from_hex("000000000000000000000000000000000000").unwrap()
-                ..Key::from_hex("FFFFFFFFFFFFFFFFFFFFFFFFFFFFFFFFFFFF").unwrap(),
-            vec![(
-                Key::from_hex("720000000033333333444444445500000000").unwrap(),
-                Lsn(0x25),
-                Value::Image(test_img("foo")),
-            )],
-        );
-
-        let image_layer = (
-            Lsn(0x40),
-            vec![(
-                Key::from_hex("620000000033333333444444445500000000").unwrap(),
-                test_img("bar"),
-            )],
-        );
-
-        let delta_layers = vec![l0_delta];
-        let image_layers = vec![image_layer];
-
-        let (tenant, ctx) = harness.load().await;
-        let timeline = tenant
-            .create_test_timeline_with_layers(
-                TimelineId::generate(),
-                Lsn(0x10),
-                14,
-                &ctx,
-                delta_layers,
-                image_layers,
-                Lsn(0x100),
-            )
-            .await
-            .unwrap();
-
-        // Layer visibility is an input to heatmap generation, so refresh it first
-        timeline.update_layer_visibility().await.unwrap();
-
-        let heatmap = timeline
-            .generate_heatmap()
-            .await
-            .expect("Infallible while timeline is not shut down");
-
-        // Both layers should be in the heatmap
-        assert!(!heatmap.layers.is_empty());
-
-        // Now simulate a migration.
-        timeline
-            .previous_heatmap
-            .store(Some(Arc::new(PreviousHeatmap::Active {
-                heatmap: heatmap.clone(),
-                read_at: std::time::Instant::now(),
-            })));
-
-        // Evict all the layers in the previous heatmap
-        let guard = timeline.layers.read().await;
-        let forever = std::time::Duration::from_secs(120);
-        for layer in guard.likely_resident_layers() {
-            layer.evict_and_wait(forever).await.unwrap();
-        }
-        drop(guard);
-
-        // Generate a new heatmap and check that the previous heatmap
-        // has been marked obsolete.
-        let post_eviction_heatmap = timeline
-            .generate_heatmap()
-            .await
-            .expect("Infallible while timeline is not shut down");
-
-        assert!(post_eviction_heatmap.layers.is_empty());
-        assert!(matches!(
-            timeline.previous_heatmap.load().as_deref(),
-            Some(PreviousHeatmap::Obsolete)
-        ));
     }
 
     #[tokio::test]

pageserver/src/tenant/timeline/compaction.rs

@@ -852,7 +852,7 @@ impl Timeline {
         //
         // Holding this read guard also blocks [`Self::gc_timeline`] from entering while we
         // are rewriting layers.
-        let latest_gc_cutoff = self.get_applied_gc_cutoff_lsn();
+        let latest_gc_cutoff = self.get_latest_gc_cutoff_lsn();
 
         tracing::info!(
             "latest_gc_cutoff: {}, pitr cutoff {}",
@@ -2202,7 +2202,7 @@ impl Timeline {
 
         // TODO: ensure the child branches will not use anything below the watermark, or consider
         // them when computing the watermark.
-        gc_cutoff_lsn.min(*self.get_applied_gc_cutoff_lsn())
+        gc_cutoff_lsn.min(*self.get_latest_gc_cutoff_lsn())
     }
 
     /// Split a gc-compaction job into multiple compaction jobs. The split is based on the key range and the estimated size of the compaction job.

pageserver/src/tenant/timeline/delete.rs

@@ -294,7 +294,6 @@ impl DeleteTimelineFlow {
                 timeline_id,
                 local_metadata,
                 None, // Ancestor is not needed for deletion.
-                None, // Previous heatmap is not needed for deletion
                 tenant.get_timeline_resources_for(remote_client),
                 // Important. We dont pass ancestor above because it can be missing.
                 // Thus we need to skip the validation here.

pageserver/src/tenant/timeline/offload.rs

@@ -7,9 +7,7 @@ use super::Timeline;
 use crate::span::debug_assert_current_span_has_tenant_and_timeline_id;
 use crate::tenant::remote_timeline_client::ShutdownIfArchivedError;
 use crate::tenant::timeline::delete::{make_timeline_delete_guard, TimelineDeleteGuardKind};
-use crate::tenant::{
-    DeleteTimelineError, OffloadedTimeline, Tenant, TenantManifestError, TimelineOrOffloaded,
-};
+use crate::tenant::{OffloadedTimeline, Tenant, TenantManifestError, TimelineOrOffloaded};
 
 #[derive(thiserror::Error, Debug)]
 pub(crate) enum OffloadError {
@@ -39,25 +37,12 @@ pub(crate) async fn offload_timeline(
     debug_assert_current_span_has_tenant_and_timeline_id();
     tracing::info!("offloading archived timeline");
 
-    let delete_guard_res = make_timeline_delete_guard(
+    let (timeline, guard) = make_timeline_delete_guard(
         tenant,
         timeline.timeline_id,
         TimelineDeleteGuardKind::Offload,
-    );
-    if let Err(DeleteTimelineError::HasChildren(children)) = delete_guard_res {
-        let is_archived = timeline.is_archived();
-        if is_archived == Some(true) {
-            tracing::error!("timeline is archived but has non-archived children: {children:?}");
-            return Err(OffloadError::NotArchived);
-        }
-        tracing::info!(
-            ?is_archived,
-            "timeline is not archived and has unarchived children"
-        );
-        return Err(OffloadError::NotArchived);
-    };
-    let (timeline, guard) =
-        delete_guard_res.map_err(|e| OffloadError::Other(anyhow::anyhow!(e)))?;
+    )
+    .map_err(|e| OffloadError::Other(anyhow::anyhow!(e)))?;
 
     let TimelineOrOffloaded::Timeline(timeline) = timeline else {
         tracing::error!("timeline already offloaded, but given timeline object");

pageserver/src/virtual_file.rs

@@ -496,8 +496,7 @@ pub(crate) fn is_fatal_io_error(e: &std::io::Error) -> bool {
 /// bad storage or bad configuration, and we can't fix that from inside
 /// a running process.
 pub(crate) fn on_fatal_io_error(e: &std::io::Error, context: &str) -> ! {
-    let backtrace = std::backtrace::Backtrace::force_capture();
-    tracing::error!("Fatal I/O error: {e}: {context})\n{backtrace}");
+    tracing::error!("Fatal I/O error: {e}: {context})");
     std::process::abort();
 }
 
@@ -948,18 +947,13 @@ impl VirtualFileInner {
     where
         Buf: tokio_epoll_uring::IoBufMut + Send,
     {
-        let file_guard = match self
-            .lock_file()
-            .await
-            .maybe_fatal_err("lock_file inside VirtualFileInner::read_at")
-        {
+        let file_guard = match self.lock_file().await {
             Ok(file_guard) => file_guard,
             Err(e) => return (buf, Err(e)),
         };
 
         observe_duration!(StorageIoOperation::Read, {
             let ((_file_guard, buf), res) = io_engine::get().read_at(file_guard, offset, buf).await;
-            let res = res.maybe_fatal_err("io_engine read_at inside VirtualFileInner::read_at");
             if let Ok(size) = res {
                 STORAGE_IO_SIZE
                     .with_label_values(&[

pgxn/neon/extension_server.c

@@ -14,12 +14,10 @@
 
 #include "utils/guc.h"
 
-#include "extension_server.h"
+#include "extension_server.h" 
 #include "neon_utils.h"
 
 static int	extension_server_port = 0;
-static int	extension_server_request_timeout = 60;
-static int	extension_server_connect_timeout = 60;
 
 static download_extension_file_hook_type prev_download_extension_file_hook = NULL;
 
@@ -36,18 +34,19 @@ static download_extension_file_hook_type prev_download_extension_file_hook = NUL
 static bool
 neon_download_extension_file_http(const char *filename, bool is_library)
 {
+	static CURL	   *handle = NULL;
+
 	CURLcode	res;
-	bool		ret = false;
-	CURL	   *handle = NULL;
 	char	   *compute_ctl_url;
+	bool		ret = false;
 
-	handle = alloc_curl_handle();
+	if (handle == NULL)
+	{
+		handle = alloc_curl_handle();
 
-	curl_easy_setopt(handle, CURLOPT_CUSTOMREQUEST, "POST");
-	if (extension_server_request_timeout > 0)
-		curl_easy_setopt(handle, CURLOPT_TIMEOUT, (long)extension_server_request_timeout /* seconds */ );
-	if (extension_server_connect_timeout > 0)
-		curl_easy_setopt(handle, CURLOPT_CONNECTTIMEOUT, (long)extension_server_connect_timeout /* seconds */ );
+		curl_easy_setopt(handle, CURLOPT_CUSTOMREQUEST, "POST");
+		curl_easy_setopt(handle, CURLOPT_TIMEOUT, 3L /* seconds */ );
+	}
 
 	compute_ctl_url = psprintf("http://localhost:%d/extension_server/%s%s",
 							   extension_server_port, filename, is_library ? "?is_library=true" : "");
@@ -58,8 +57,6 @@ neon_download_extension_file_http(const char *filename, bool is_library)
 
 	/* Perform the request, res will get the return code */
 	res = curl_easy_perform(handle);
-	curl_easy_cleanup(handle);
-
 	/* Check for errors */
 	if (res == CURLE_OK)
 	{
@@ -91,24 +88,6 @@ pg_init_extension_server()
 							0,	/* no flags required */
 							NULL, NULL, NULL);
 
-	DefineCustomIntVariable("neon.extension_server_request_timeout",
-							"timeout for fetching extensions in seconds",
-							NULL,
-							&extension_server_request_timeout,
-							60, 0, INT_MAX,
-							PGC_SUSET,
-							GUC_UNIT_S,
-							NULL, NULL, NULL);
-
-	DefineCustomIntVariable("neon.extension_server_connect_timeout",
-							"timeout for connecting to the extension server in seconds",
-							NULL,
-							&extension_server_connect_timeout,
-							60, 0, INT_MAX,
-							PGC_SUSET,
-							GUC_UNIT_S,
-							NULL, NULL, NULL);
-
 	/* set download_extension_file_hook */
 	prev_download_extension_file_hook = download_extension_file_hook;
 	download_extension_file_hook = neon_download_extension_file_http;

pgxn/neon/hll.c

@@ -122,8 +122,8 @@ addSHLL(HyperLogLogState *cState, uint32 hash)
 	index = hash >> HLL_C_BITS;
 
 	/* Compute the rank of the remaining 32 - "k" (registerWidth) bits */
-	count = rho(hash << HLL_BIT_WIDTH, HLL_C_BITS) - 1;
-	Assert(count <= HLL_C_BITS);
+	count = rho(hash << HLL_BIT_WIDTH, HLL_C_BITS);
+
 	cState->regs[index][count] = now;
 }
 
@@ -136,7 +136,7 @@ getMaximum(const TimestampTz* reg, TimestampTz since)
 	{
 		if (reg[i] >= since)
 		{
-			max = i + 1;
+			max = i;
 		}
 	}
 

pgxn/neon/libpagestore.c

@@ -378,9 +378,8 @@ pageserver_connect(shardno_t shard_no, int elevel)
 	{
 	case PS_Disconnected:
 	{
-		const char *keywords[4];
-		const char *values[4];
-		char pid_str[16];
+		const char *keywords[3];
+		const char *values[3];
 		int			n_pgsql_params;
 		TimestampTz	now;
 		int64		us_since_last_attempt;
@@ -425,30 +424,14 @@ pageserver_connect(shardno_t shard_no, int elevel)
 		 * can override the password from the env variable. Seems useful, although
 		 * we don't currently use that capability anywhere.
 		 */
-		n_pgsql_params = 0;
-
-		/*
-		 * Pageserver logs include this in the connection's tracing span.
-		 * This allows for reasier log correlation between compute and pageserver.
-		 */
-		keywords[n_pgsql_params] = "application_name";
-		{
-			int ret = snprintf(pid_str, sizeof(pid_str), "%d", MyProcPid);
-			if (ret < 0 || ret >= (int)(sizeof(pid_str)))
-				elog(FATAL, "stack-allocated buffer too small to hold pid");
-		}
-		/* lifetime: PQconnectStartParams strdups internally */
-		values[n_pgsql_params] = (const char*) pid_str;
-		n_pgsql_params++;
-
-		keywords[n_pgsql_params] = "dbname";
-		values[n_pgsql_params] = connstr;
-		n_pgsql_params++;
+		keywords[0] = "dbname";
+		values[0] = connstr;
+		n_pgsql_params = 1;
 
 		if (neon_auth_token)
 		{
-			keywords[n_pgsql_params] = "password";
-			values[n_pgsql_params] = neon_auth_token;
+			keywords[1] = "password";
+			values[1] = neon_auth_token;
 			n_pgsql_params++;
 		}
 

pgxn/neon/pagestore_smgr.c

@@ -3765,7 +3765,7 @@ neon_dbsize(Oid dbNode)
  *	neon_truncate() -- Truncate relation to specified number of blocks.
  */
 static void
-neon_truncate(SMgrRelation reln, ForkNumber forknum, BlockNumber old_blocks, BlockNumber nblocks)
+neon_truncate(SMgrRelation reln, ForkNumber forknum, BlockNumber nblocks)
 {
 	XLogRecPtr	lsn;
 
@@ -3780,7 +3780,7 @@ neon_truncate(SMgrRelation reln, ForkNumber forknum, BlockNumber old_blocks, Blo
 
 		case RELPERSISTENCE_TEMP:
 		case RELPERSISTENCE_UNLOGGED:
-			mdtruncate(reln, forknum, old_blocks, nblocks);
+			mdtruncate(reln, forknum, nblocks);
 			return;
 
 		default:
@@ -3818,7 +3818,7 @@ neon_truncate(SMgrRelation reln, ForkNumber forknum, BlockNumber old_blocks, Blo
 
 #ifdef DEBUG_COMPARE_LOCAL
 	if (IS_LOCAL_REL(reln))
-		mdtruncate(reln, forknum, old_blocks, nblocks);
+		mdtruncate(reln, forknum, nblocks);
 #endif
 }
 

pgxn/neon_walredo/inmem_smgr.c

@@ -96,7 +96,7 @@ static void inmem_writeback(SMgrRelation reln, ForkNumber forknum,
 							BlockNumber blocknum, BlockNumber nblocks);
 static BlockNumber inmem_nblocks(SMgrRelation reln, ForkNumber forknum);
 static void inmem_truncate(SMgrRelation reln, ForkNumber forknum,
-						   BlockNumber old_blocks, BlockNumber nblocks);
+						   BlockNumber nblocks);
 static void inmem_immedsync(SMgrRelation reln, ForkNumber forknum);
 #if PG_MAJORVERSION_NUM >= 17
 static void inmem_registersync(SMgrRelation reln, ForkNumber forknum);
@@ -345,7 +345,7 @@ inmem_nblocks(SMgrRelation reln, ForkNumber forknum)
  *	inmem_truncate() -- Truncate relation to specified number of blocks.
  */
 static void
-inmem_truncate(SMgrRelation reln, ForkNumber forknum, BlockNumber old_blocks, BlockNumber nblocks)
+inmem_truncate(SMgrRelation reln, ForkNumber forknum, BlockNumber nblocks)
 {
 }
 

poetry.lock

@@ -412,7 +412,6 @@ files = [
 
 [package.dependencies]
 botocore-stubs = "*"
-mypy-boto3-kms = {version = ">=1.26.0,<1.27.0", optional = true, markers = "extra == \"kms\""}
 mypy-boto3-s3 = {version = ">=1.26.0,<1.27.0", optional = true, markers = "extra == \"s3\""}
 types-s3transfer = "*"
 typing-extensions = ">=4.1.0"
@@ -2023,18 +2022,6 @@ install-types = ["pip"]
 mypyc = ["setuptools (>=50)"]
 reports = ["lxml"]
 
-[[package]]
-name = "mypy-boto3-kms"
-version = "1.26.147"
-description = "Type annotations for boto3.KMS 1.26.147 service generated with mypy-boto3-builder 7.14.5"
-optional = false
-python-versions = ">=3.7"
-groups = ["main"]
-files = [
-    {file = "mypy-boto3-kms-1.26.147.tar.gz", hash = "sha256:816a4d1bb0585e1b9620a3f96c1d69a06f53b7b5621858579dd77c60dbb5fa5c"},
-    {file = "mypy_boto3_kms-1.26.147-py3-none-any.whl", hash = "sha256:493f0db674a25c88769f5cb8ab8ac00d3dda5dfc903d5cda34c990ee64689f79"},
-]
-
 [[package]]
 name = "mypy-boto3-s3"
 version = "1.26.0.post1"
@@ -2771,18 +2758,18 @@ pytest = ">=5,<8"
 
 [[package]]
 name = "pytest-timeout"
-version = "2.3.1"
+version = "2.1.0"
 description = "pytest plugin to abort hanging tests"
 optional = false
-python-versions = ">=3.7"
+python-versions = ">=3.6"
 groups = ["main"]
 files = [
-    {file = "pytest-timeout-2.3.1.tar.gz", hash = "sha256:12397729125c6ecbdaca01035b9e5239d4db97352320af155b3f5de1ba5165d9"},
-    {file = "pytest_timeout-2.3.1-py3-none-any.whl", hash = "sha256:68188cb703edfc6a18fad98dc25a3c61e9f24d644b0b70f33af545219fc7813e"},
+    {file = "pytest-timeout-2.1.0.tar.gz", hash = "sha256:c07ca07404c612f8abbe22294b23c368e2e5104b521c1790195561f37e1ac3d9"},
+    {file = "pytest_timeout-2.1.0-py3-none-any.whl", hash = "sha256:f6f50101443ce70ad325ceb4473c4255e9d74e3c7cd0ef827309dfa4c0d975c6"},
 ]
 
 [package.dependencies]
-pytest = ">=7.0.0"
+pytest = ">=5.0.0"
 
 [[package]]
 name = "pytest-xdist"
@@ -3820,4 +3807,4 @@ cffi = ["cffi (>=1.11)"]
 [metadata]
 lock-version = "2.1"
 python-versions = "^3.11"
-content-hash = "00ddc42c32e235b6171845fc066dcab078282ed832cd464d5e8a0afa959dd04a"
+content-hash = "4dc3165fe22c0e0f7a030ea0f8a680ae2ff74561d8658c393abbe9112caaf5d7"

proxy/Cargo.toml

@@ -19,6 +19,7 @@ aws-config.workspace = true
 aws-sdk-iam.workspace = true
 aws-sigv4.workspace = true
 base64.workspace = true
+boxcar = "0.2.8"
 bstr.workspace = true
 bytes = { workspace = true, features = ["serde"] }
 camino.workspace = true
@@ -62,6 +63,7 @@ postgres_backend.workspace = true
 postgres-client = { package = "tokio-postgres2", path = "../libs/proxy/tokio-postgres2" }
 postgres-protocol = { package = "postgres-protocol2", path = "../libs/proxy/postgres-protocol2" }
 pq_proto.workspace = true
+prometheus.workspace = true
 rand.workspace = true
 regex.workspace = true
 remote_storage = { version = "0.1", path = "../libs/remote_storage/" }
@@ -79,6 +81,7 @@ sha2 = { workspace = true, features = ["asm", "oid"] }
 smol_str.workspace = true
 smallvec.workspace = true
 socket2.workspace = true
+strum.workspace = true
 strum_macros.workspace = true
 subtle.workspace = true
 thiserror.workspace = true
@@ -92,6 +95,7 @@ tracing-subscriber.workspace = true
 tracing-utils.workspace = true
 tracing.workspace = true
 tracing-log.workspace = true
+tracing-serde.workspace = true
 tracing-opentelemetry.workspace = true
 try-lock.workspace = true
 typed-json.workspace = true

proxy/src/auth/backend/console_redirect.rs

@@ -140,8 +140,9 @@ async fn authenticate(
     let (psql_session_id, waiter) = loop {
         let psql_session_id = new_psql_session_id();
 
-        if let Ok(waiter) = control_plane::mgmt::get_waiter(&psql_session_id) {
-            break (psql_session_id, waiter);
+        match control_plane::mgmt::get_waiter(&psql_session_id) {
+            Ok(waiter) => break (psql_session_id, waiter),
+            Err(_e) => continue,
         }
     };
 

proxy/src/auth/backend/jwt.rs

@@ -220,11 +220,11 @@ async fn fetch_jwks(
 }
 
 impl JwkCacheEntryLock {
-    async fn acquire_permit(self: &Arc<Self>) -> JwkRenewalPermit<'_> {
+    async fn acquire_permit<'a>(self: &'a Arc<Self>) -> JwkRenewalPermit<'a> {
         JwkRenewalPermit::acquire_permit(self).await
     }
 
-    fn try_acquire_permit(self: &Arc<Self>) -> Option<JwkRenewalPermit<'_>> {
+    fn try_acquire_permit<'a>(self: &'a Arc<Self>) -> Option<JwkRenewalPermit<'a>> {
         JwkRenewalPermit::try_acquire_permit(self)
     }
 
@@ -393,7 +393,7 @@ impl JwkCacheEntryLock {
                 verify_rsa_signature(header_payload.as_bytes(), &sig, key, &header.algorithm)?;
             }
             key => return Err(JwtError::UnsupportedKeyType(key.into())),
-        }
+        };
 
         tracing::debug!(?payload, "JWT signature valid with claims");
 
@@ -510,7 +510,7 @@ fn verify_rsa_signature(
             key.verify(data, &sig)?;
         }
         _ => return Err(JwtError::InvalidRsaSigningAlgorithm),
-    }
+    };
 
     Ok(())
 }

proxy/src/binary/local_proxy.rs

@@ -4,20 +4,6 @@ use std::str::FromStr;
 use std::sync::Arc;
 use std::time::Duration;
 
-use anyhow::{bail, ensure, Context};
-use camino::{Utf8Path, Utf8PathBuf};
-use clap::Parser;
-use compute_api::spec::LocalProxySpec;
-use futures::future::Either;
-use thiserror::Error;
-use tokio::net::TcpListener;
-use tokio::sync::Notify;
-use tokio::task::JoinSet;
-use tokio_util::sync::CancellationToken;
-use tracing::{debug, error, info, warn};
-use utils::sentry_init::init_sentry;
-use utils::{pid_file, project_build_tag, project_git_version};
-
 use crate::auth::backend::jwt::JwkCache;
 use crate::auth::backend::local::{LocalBackend, JWKS_ROLE_MAP};
 use crate::auth::{self};
@@ -39,10 +25,24 @@ use crate::serverless::{self, GlobalConnPoolOptions};
 use crate::tls::client_config::compute_client_config_with_root_certs;
 use crate::types::RoleName;
 use crate::url::ApiUrl;
+use anyhow::{bail, ensure, Context};
+use camino::{Utf8Path, Utf8PathBuf};
+use compute_api::spec::LocalProxySpec;
+use futures::future::Either;
 
 project_git_version!(GIT_VERSION);
 project_build_tag!(BUILD_TAG);
 
+use clap::Parser;
+use thiserror::Error;
+use tokio::net::TcpListener;
+use tokio::sync::Notify;
+use tokio::task::JoinSet;
+use tokio_util::sync::CancellationToken;
+use tracing::{debug, error, info, warn};
+use utils::sentry_init::init_sentry;
+use utils::{pid_file, project_build_tag, project_git_version};
+
 /// Neon proxy/router
 #[derive(Parser)]
 #[command(version = GIT_VERSION, about)]

proxy/src/binary/pg_sni_router.rs

@@ -5,6 +5,12 @@
 /// the outside. Similar to an ingress controller for HTTPS.
 use std::{net::SocketAddr, sync::Arc};
 
+use crate::context::RequestContext;
+use crate::metrics::{Metrics, ThreadPoolMetrics};
+use crate::protocol2::ConnectionInfo;
+use crate::proxy::{copy_bidirectional_client_compute, run_until_cancelled, ErrorSource};
+use crate::stream::{PqStream, Stream};
+use crate::tls::TlsServerEndPoint;
 use anyhow::{anyhow, bail, ensure, Context};
 use clap::Arg;
 use futures::future::Either;
@@ -19,13 +25,6 @@ use tracing::{error, info, Instrument};
 use utils::project_git_version;
 use utils::sentry_init::init_sentry;
 
-use crate::context::RequestContext;
-use crate::metrics::{Metrics, ThreadPoolMetrics};
-use crate::protocol2::ConnectionInfo;
-use crate::proxy::{copy_bidirectional_client_compute, run_until_cancelled, ErrorSource};
-use crate::stream::{PqStream, Stream};
-use crate::tls::TlsServerEndPoint;
-
 project_git_version!(GIT_VERSION);
 
 fn cli() -> clap::Command {

proxy/src/binary/proxy.rs

@@ -3,16 +3,6 @@ use std::pin::pin;
 use std::sync::Arc;
 use std::time::Duration;
 
-use anyhow::bail;
-use futures::future::Either;
-use remote_storage::RemoteStorageConfig;
-use tokio::net::TcpListener;
-use tokio::task::JoinSet;
-use tokio_util::sync::CancellationToken;
-use tracing::{info, warn, Instrument};
-use utils::sentry_init::init_sentry;
-use utils::{project_build_tag, project_git_version};
-
 use crate::auth::backend::jwt::JwkCache;
 use crate::auth::backend::{AuthRateLimiter, ConsoleRedirectBackend, MaybeOwned};
 use crate::cancellation::{handle_cancel_messages, CancellationHandler};
@@ -34,6 +24,15 @@ use crate::serverless::cancel_set::CancelSet;
 use crate::serverless::GlobalConnPoolOptions;
 use crate::tls::client_config::compute_client_config_with_root_certs;
 use crate::{auth, control_plane, http, serverless, usage_metrics};
+use anyhow::bail;
+use futures::future::Either;
+use remote_storage::RemoteStorageConfig;
+use tokio::net::TcpListener;
+use tokio::task::JoinSet;
+use tokio_util::sync::CancellationToken;
+use tracing::{info, warn, Instrument};
+use utils::sentry_init::init_sentry;
+use utils::{project_build_tag, project_git_version};
 
 project_git_version!(GIT_VERSION);
 project_build_tag!(BUILD_TAG);
@@ -304,7 +303,7 @@ pub async fn run() -> anyhow::Result<()> {
     match auth_backend {
         Either::Left(auth_backend) => info!("Authentication backend: {auth_backend}"),
         Either::Right(auth_backend) => info!("Authentication backend: {auth_backend:?}"),
-    }
+    };
     info!("Using region: {}", args.aws_region);
 
     // TODO: untangle the config args
@@ -804,9 +803,8 @@ fn build_auth_backend(
 mod tests {
     use std::time::Duration;
 
-    use clap::Parser;
-
     use crate::rate_limiter::RateBucketInfo;
+    use clap::Parser;
 
     #[test]
     fn parse_endpoint_rps_limit() {

proxy/src/cache/endpoints.rs

@@ -242,7 +242,7 @@ impl EndpointsCache {
                             });
                             tracing::error!("error parsing value {value:?}: {err:?}");
                         }
-                    }
+                    };
                 }
                 if total.is_power_of_two() {
                     tracing::debug!("endpoints read {}", total);

proxy/src/cancellation.rs

@@ -501,7 +501,7 @@ impl Session {
             _guard: Metrics::get()
                 .proxy
                 .cancel_channel_size
-                .guard(RedisMsgKind::HDel),
+                .guard(RedisMsgKind::HSet),
         };
 
         let _ = tx.send_timeout(op, REDIS_SEND_TIMEOUT).await.map_err(|e| {

proxy/src/compute.rs

@@ -137,8 +137,8 @@ impl ConnCfg {
             match k {
                 // Only set `user` if it's not present in the config.
                 // Console redirect auth flow takes username from the console's response.
-                "user" if self.user_is_set() => {}
-                "database" if self.db_is_set() => {}
+                "user" if self.user_is_set() => continue,
+                "database" if self.db_is_set() => continue,
                 "options" => {
                     if let Some(options) = filtered_options(v) {
                         self.set_param(k, &options);

proxy/src/console_redirect_proxy.rs

@@ -82,7 +82,7 @@ pub async fn task_main(
                     error!("per-client task finished with an error: failed to set socket option: {e:#}");
                     return;
                 }
-            }
+            };
 
             let ctx = RequestContext::new(
                 session_id,

proxy/src/control_plane/mod.rs

@@ -19,7 +19,8 @@ use crate::cache::{Cached, TimedLru};
 use crate::config::ComputeConfig;
 use crate::context::RequestContext;
 use crate::control_plane::messages::{ControlPlaneErrorMessage, MetricsAuxInfo};
-use crate::intern::{AccountIdInt, ProjectIdInt};
+use crate::intern::AccountIdInt;
+use crate::intern::ProjectIdInt;
 use crate::types::{EndpointCacheKey, EndpointId};
 use crate::{compute, scram};
 

proxy/src/logging.rs

@@ -7,8 +7,9 @@ use chrono::{DateTime, Utc};
 use opentelemetry::trace::TraceContextExt;
 use scopeguard::defer;
 use serde::ser::{SerializeMap, Serializer};
+use tracing::span;
 use tracing::subscriber::Interest;
-use tracing::{callsite, span, Event, Metadata, Span, Subscriber};
+use tracing::{callsite, Event, Metadata, Span, Subscriber};
 use tracing_opentelemetry::OpenTelemetrySpanExt;
 use tracing_subscriber::filter::{EnvFilter, LevelFilter};
 use tracing_subscriber::fmt::format::{Format, Full};

proxy/src/protocol2.rs

@@ -119,7 +119,7 @@ pub(crate) async fn read_proxy_protocol<T: AsyncRead + Unpin>(
         // if no more bytes available then exit
         if bytes_read == 0 {
             return Ok((ChainRW { inner: read, buf }, ConnectHeader::Missing));
-        }
+        };
 
         // check if we have enough bytes to continue
         if let Some(header) = buf.try_get::<ProxyProtocolV2Header>() {
@@ -169,7 +169,7 @@ fn process_proxy_payload(
                 header.version_and_command
             ),
         )),
-    }
+    };
 
     let size_err =
         "invalid proxy protocol length. payload not large enough to fit requested IP addresses";

proxy/src/proxy/connect_compute.rs

@@ -198,7 +198,7 @@ where
 
                 warn!(error = ?e, num_retries, retriable = true, COULD_NOT_CONNECT);
             }
-        }
+        };
 
         let wait_duration = retry_after(num_retries, compute.retry);
         num_retries += 1;

proxy/src/proxy/mod.rs

@@ -118,7 +118,7 @@ pub async fn task_main(
                     error!("per-client task finished with an error: failed to set socket option: {e:#}");
                     return;
                 }
-            }
+            };
 
             let ctx = RequestContext::new(
                 session_id,

proxy/src/redis/notifications.rs

@@ -169,7 +169,7 @@ impl<C: ProjectInfoCache + Send + Sync + 'static> MessageHandler<C> {
                         });
                         tracing::error!("broken message: {e}");
                     }
-                }
+                };
                 return Ok(());
             }
             Ok(msg) => msg,
@@ -180,7 +180,7 @@ impl<C: ProjectInfoCache + Send + Sync + 'static> MessageHandler<C> {
                 match serde_json::from_str::<NotificationHeader>(&payload) {
                     Ok(header) => tracing::error!(topic = header.topic, "broken message: {e}"),
                     Err(_) => tracing::error!("broken message: {e}"),
-                }
+                };
                 return Ok(());
             }
         };

proxy/src/serverless/backend.rs

@@ -372,7 +372,7 @@ impl PoolingBackend {
             debug!("setting up backend session state");
 
             // initiates the auth session
-            if let Err(e) = client.batch_execute("select auth.init();").await {
+            if let Err(e) = client.execute("select auth.init()", &[]).await {
                 discard.discard();
                 return Err(e.into());
             }
@@ -651,7 +651,7 @@ async fn connect_http2(
                     e,
                 )));
             }
-        }
+        };
     };
 
     let (client, connection) = hyper::client::conn::http2::Builder::new(TokioExecutor::new())

proxy/src/serverless/local_conn_pool.rs

@@ -23,6 +23,7 @@ use indexmap::IndexMap;
 use jose_jwk::jose_b64::base64ct::{Base64UrlUnpadded, Encoding};
 use parking_lot::RwLock;
 use postgres_client::tls::NoTlsStream;
+use postgres_client::types::ToSql;
 use postgres_client::AsyncMessage;
 use serde_json::value::RawValue;
 use tokio::net::TcpStream;
@@ -280,9 +281,13 @@ impl ClientInnerCommon<postgres_client::Client> {
             let token = resign_jwt(&local_data.key, payload, local_data.jti)?;
 
             // initiates the auth session
-            // this is safe from query injections as the jwt format free of any escape characters.
-            let query = format!("discard all; select auth.jwt_session_init('{token}')");
-            self.inner.batch_execute(&query).await?;
+            self.inner.batch_execute("discard all").await?;
+            self.inner
+                .execute(
+                    "select auth.jwt_session_init($1)",
+                    &[&&*token as &(dyn ToSql + Sync)],
+                )
+                .await?;
 
             let pid = self.inner.get_process_id();
             info!(pid, jti = local_data.jti, "user session state init");

pyproject.toml

@@ -17,12 +17,12 @@ Jinja2 = "^3.1.5"
 types-requests = "^2.31.0.0"
 types-psycopg2 = "^2.9.21.20241019"
 boto3 = "^1.34.11"
-boto3-stubs = {extras = ["s3", "kms"], version = "^1.26.16"}
+boto3-stubs = {extras = ["s3"], version = "^1.26.16"}
 moto = {extras = ["server"], version = "^5.0.6"}
 backoff = "^2.2.1"
 pytest-lazy-fixture = "^0.6.3"
 prometheus-client = "^0.14.1"
-pytest-timeout = "^2.3.1"
+pytest-timeout = "^2.1.0"
 Werkzeug = "^3.0.6"
 pytest-order = "^1.1.0"
 allure-pytest = "^2.13.2"

pytest.ini

@@ -11,7 +11,7 @@ markers =
 testpaths =
     test_runner
 minversion = 6.0
-log_format = %(asctime)s.%(msecs)03d %(levelname)s [%(filename)s:%(lineno)d] %(message)s
+log_format = %(asctime)s.%(msecs)-3d %(levelname)s [%(filename)s:%(lineno)d] %(message)s
 log_date_format = %Y-%m-%d %H:%M:%S
 log_cli = true
 timeout = 300

safekeeper/client/src/mgmt_api.rs

@@ -5,7 +5,7 @@
 
 use http_utils::error::HttpErrorBody;
 use reqwest::{IntoUrl, Method, StatusCode};
-use safekeeper_api::models::{SafekeeperUtilization, TimelineCreateRequest, TimelineStatus};
+use safekeeper_api::models::{TimelineCreateRequest, TimelineStatus};
 use std::error::Error as _;
 use utils::{
     id::{NodeId, TenantId, TimelineId},
@@ -32,9 +32,6 @@ pub enum Error {
     /// Status is not ok; parsed error in body as `HttpErrorBody`.
     #[error("safekeeper API: {1}")]
     ApiError(StatusCode, String),
-
-    #[error("Cancelled")]
-    Cancelled,
 }
 
 pub type Result<T> = std::result::Result<T, Error>;
@@ -127,10 +124,9 @@ impl Client {
         self.get(&uri).await
     }
 
-    pub async fn utilization(&self) -> Result<SafekeeperUtilization> {
+    pub async fn utilization(&self) -> Result<reqwest::Response> {
         let uri = format!("{}/v1/utilization/", self.mgmt_api_endpoint);
-        let resp = self.get(&uri).await?;
-        resp.json().await.map_err(Error::ReceiveBody)
+        self.get(&uri).await
     }
 
     async fn post<B: serde::Serialize, U: IntoUrl>(

safekeeper/src/http/routes.rs

@@ -626,7 +626,7 @@ pub fn make_router(
                 failpoints_handler(r, cancel).await
             })
         })
-        .get("/v1/utilization", |r| request_span(r, utilization_handler))
+        .get("/v1/uzilization", |r| request_span(r, utilization_handler))
         .delete("/v1/tenant/:tenant_id", |r| {
             request_span(r, tenant_delete_handler)
         })

safekeeper/src/wal_backup.rs

@@ -310,12 +310,9 @@ impl WalBackupTask {
                     retry_attempt = 0;
                 }
                 Err(e) => {
-                    // We might have managed to upload some segment even though
-                    // some later in the range failed, so log backup_lsn
-                    // separately.
                     error!(
-                        "failed while offloading range {}-{}, backup_lsn {}: {:?}",
-                        backup_lsn, commit_lsn, backup_lsn, e
+                        "failed while offloading range {}-{}: {:?}",
+                        backup_lsn, commit_lsn, e
                     );
 
                     retry_attempt = retry_attempt.saturating_add(1);
@@ -341,13 +338,6 @@ async fn backup_lsn_range(
     let start_lsn = *backup_lsn;
     let segments = get_segments(start_lsn, end_lsn, wal_seg_size);
 
-    info!(
-        "offloading segnos {:?} of range [{}-{})",
-        segments.iter().map(|&s| s.seg_no).collect::<Vec<_>>(),
-        start_lsn,
-        end_lsn,
-    );
-
     // Pool of concurrent upload tasks. We use `FuturesOrdered` to
     // preserve order of uploads, and update `backup_lsn` only after
     // all previous uploads are finished.
@@ -384,10 +374,10 @@ async fn backup_lsn_range(
     }
 
     info!(
-        "offloaded segnos {:?} of range [{}-{})",
+        "offloaded segnos {:?} up to {}, previous backup_lsn {}",
         segments.iter().map(|&s| s.seg_no).collect::<Vec<_>>(),
-        start_lsn,
         end_lsn,
+        start_lsn,
     );
     Ok(())
 }

safekeeper/src/wal_service.rs

@@ -13,8 +13,6 @@ use tokio_util::sync::CancellationToken;
 use tracing::*;
 use utils::{auth::Scope, measured_stream::MeasuredStream};
 
-use std::os::fd::AsRawFd;
-
 use crate::metrics::TrafficMetrics;
 use crate::SafeKeeperConf;
 use crate::{handler::SafekeeperPostgresHandler, GlobalTimelines};
@@ -64,7 +62,6 @@ async fn handle_socket(
     global_timelines: Arc<GlobalTimelines>,
 ) -> Result<(), QueryError> {
     socket.set_nodelay(true)?;
-    let socket_fd = socket.as_raw_fd();
     let peer_addr = socket.peer_addr()?;
 
     // Set timeout on reading from the socket. It prevents hanged up connection
@@ -110,7 +107,7 @@ async fn handle_socket(
         auth_pair,
         global_timelines,
     );
-    let pgbackend = PostgresBackend::new_from_io(socket_fd, socket, peer_addr, auth_type, None)?;
+    let pgbackend = PostgresBackend::new_from_io(socket, peer_addr, auth_type, None)?;
     // libpq protocol between safekeeper and walproposer / pageserver
     // We don't use shutdown.
     pgbackend

storage_controller/Cargo.toml

@@ -32,9 +32,6 @@ postgres_connection.workspace = true
 rand.workspace = true
 reqwest = { workspace = true, features = ["stream"] }
 routerify.workspace = true
-safekeeper_api.workspace = true
-safekeeper_client.workspace = true
-regex.workspace = true
 rustls-native-certs.workspace = true
 serde.workspace = true
 serde_json.workspace = true

storage_controller/src/heartbeater.rs

@@ -1,10 +1,6 @@
 use futures::{stream::FuturesUnordered, StreamExt};
-use safekeeper_api::models::SafekeeperUtilization;
-use safekeeper_client::mgmt_api;
 use std::{
     collections::HashMap,
-    fmt::Debug,
-    future::Future,
     sync::Arc,
     time::{Duration, Instant},
 };
@@ -13,15 +9,15 @@ use tokio_util::sync::CancellationToken;
 use pageserver_api::{controller_api::NodeAvailability, models::PageserverUtilization};
 
 use thiserror::Error;
-use utils::{id::NodeId, logging::SecretString};
+use utils::id::NodeId;
 
-use crate::{node::Node, safekeeper::Safekeeper};
+use crate::node::Node;
 
-struct HeartbeaterTask<Server, State> {
-    receiver: tokio::sync::mpsc::UnboundedReceiver<HeartbeatRequest<Server, State>>,
+struct HeartbeaterTask {
+    receiver: tokio::sync::mpsc::UnboundedReceiver<HeartbeatRequest>,
     cancel: CancellationToken,
 
-    state: HashMap<NodeId, State>,
+    state: HashMap<NodeId, PageserverState>,
 
     max_offline_interval: Duration,
     max_warming_up_interval: Duration,
@@ -40,17 +36,8 @@ pub(crate) enum PageserverState {
     Offline,
 }
 
-#[derive(Debug, Clone)]
-pub(crate) enum SafekeeperState {
-    Available {
-        last_seen_at: Instant,
-        utilization: SafekeeperUtilization,
-    },
-    Offline,
-}
-
 #[derive(Debug)]
-pub(crate) struct AvailablityDeltas<State>(pub Vec<(NodeId, State)>);
+pub(crate) struct AvailablityDeltas(pub Vec<(NodeId, PageserverState)>);
 
 #[derive(Debug, Error)]
 pub(crate) enum HeartbeaterError {
@@ -58,28 +45,23 @@ pub(crate) enum HeartbeaterError {
     Cancel,
 }
 
-struct HeartbeatRequest<Server, State> {
-    servers: Arc<HashMap<NodeId, Server>>,
-    reply: tokio::sync::oneshot::Sender<Result<AvailablityDeltas<State>, HeartbeaterError>>,
+struct HeartbeatRequest {
+    pageservers: Arc<HashMap<NodeId, Node>>,
+    reply: tokio::sync::oneshot::Sender<Result<AvailablityDeltas, HeartbeaterError>>,
 }
 
-pub(crate) struct Heartbeater<Server, State> {
-    sender: tokio::sync::mpsc::UnboundedSender<HeartbeatRequest<Server, State>>,
+pub(crate) struct Heartbeater {
+    sender: tokio::sync::mpsc::UnboundedSender<HeartbeatRequest>,
 }
 
-#[allow(private_bounds)]
-impl<Server: Send + Sync + 'static, State: Debug + Send + 'static> Heartbeater<Server, State>
-where
-    HeartbeaterTask<Server, State>: HeartBeat<Server, State>,
-{
+impl Heartbeater {
     pub(crate) fn new(
         jwt_token: Option<String>,
         max_offline_interval: Duration,
         max_warming_up_interval: Duration,
         cancel: CancellationToken,
     ) -> Self {
-        let (sender, receiver) =
-            tokio::sync::mpsc::unbounded_channel::<HeartbeatRequest<Server, State>>();
+        let (sender, receiver) = tokio::sync::mpsc::unbounded_channel::<HeartbeatRequest>();
         let mut heartbeater = HeartbeaterTask::new(
             receiver,
             jwt_token,
@@ -94,12 +76,12 @@ where
 
     pub(crate) async fn heartbeat(
         &self,
-        servers: Arc<HashMap<NodeId, Server>>,
-    ) -> Result<AvailablityDeltas<State>, HeartbeaterError> {
+        pageservers: Arc<HashMap<NodeId, Node>>,
+    ) -> Result<AvailablityDeltas, HeartbeaterError> {
         let (sender, receiver) = tokio::sync::oneshot::channel();
         self.sender
             .send(HeartbeatRequest {
-                servers,
+                pageservers,
                 reply: sender,
             })
             .map_err(|_| HeartbeaterError::Cancel)?;
@@ -111,12 +93,9 @@ where
     }
 }
 
-impl<Server, State: Debug> HeartbeaterTask<Server, State>
-where
-    HeartbeaterTask<Server, State>: HeartBeat<Server, State>,
-{
+impl HeartbeaterTask {
     fn new(
-        receiver: tokio::sync::mpsc::UnboundedReceiver<HeartbeatRequest<Server, State>>,
+        receiver: tokio::sync::mpsc::UnboundedReceiver<HeartbeatRequest>,
         jwt_token: Option<String>,
         max_offline_interval: Duration,
         max_warming_up_interval: Duration,
@@ -131,13 +110,14 @@ where
             jwt_token,
         }
     }
+
     async fn run(&mut self) {
         loop {
             tokio::select! {
                 request = self.receiver.recv() => {
                     match request {
                         Some(req) => {
-                            let res = self.heartbeat(req.servers).await;
+                            let res = self.heartbeat(req.pageservers).await;
                             req.reply.send(res).unwrap();
                         },
                         None => { return; }
@@ -147,20 +127,11 @@ where
             }
         }
     }
-}
 
-pub(crate) trait HeartBeat<Server, State> {
-    fn heartbeat(
-        &mut self,
-        pageservers: Arc<HashMap<NodeId, Server>>,
-    ) -> impl Future<Output = Result<AvailablityDeltas<State>, HeartbeaterError>> + Send;
-}
-
-impl HeartBeat<Node, PageserverState> for HeartbeaterTask<Node, PageserverState> {
     async fn heartbeat(
         &mut self,
         pageservers: Arc<HashMap<NodeId, Node>>,
-    ) -> Result<AvailablityDeltas<PageserverState>, HeartbeaterError> {
+    ) -> Result<AvailablityDeltas, HeartbeaterError> {
         let mut new_state = HashMap::new();
 
         let mut heartbeat_futs = FuturesUnordered::new();
@@ -301,121 +272,3 @@ impl HeartBeat<Node, PageserverState> for HeartbeaterTask<Node, PageserverState>
         Ok(AvailablityDeltas(deltas))
     }
 }
-
-impl HeartBeat<Safekeeper, SafekeeperState> for HeartbeaterTask<Safekeeper, SafekeeperState> {
-    async fn heartbeat(
-        &mut self,
-        safekeepers: Arc<HashMap<NodeId, Safekeeper>>,
-    ) -> Result<AvailablityDeltas<SafekeeperState>, HeartbeaterError> {
-        let mut new_state = HashMap::new();
-
-        let mut heartbeat_futs = FuturesUnordered::new();
-        for (node_id, sk) in &*safekeepers {
-            heartbeat_futs.push({
-                let jwt_token = self
-                    .jwt_token
-                    .as_ref()
-                    .map(|t| SecretString::from(t.to_owned()));
-                let cancel = self.cancel.clone();
-
-                async move {
-                    let response = sk
-                        .with_client_retries(
-                            |client| async move { client.get_utilization().await },
-                            &jwt_token,
-                            3,
-                            3,
-                            Duration::from_secs(1),
-                            &cancel,
-                        )
-                        .await;
-
-                    let status = match response {
-                        Ok(utilization) => SafekeeperState::Available {
-                            last_seen_at: Instant::now(),
-                            utilization,
-                        },
-                        Err(mgmt_api::Error::Cancelled) => {
-                            // This indicates cancellation of the request.
-                            // We ignore the node in this case.
-                            return None;
-                        }
-                        Err(_) => SafekeeperState::Offline,
-                    };
-
-                    Some((*node_id, status))
-                }
-            });
-
-            loop {
-                let maybe_status = tokio::select! {
-                    next = heartbeat_futs.next() => {
-                        match next {
-                            Some(result) => result,
-                            None => { break; }
-                        }
-                    },
-                    _ = self.cancel.cancelled() => { return Err(HeartbeaterError::Cancel); }
-                };
-
-                if let Some((node_id, status)) = maybe_status {
-                    new_state.insert(node_id, status);
-                }
-            }
-        }
-
-        let mut offline = 0;
-        for state in new_state.values() {
-            match state {
-                SafekeeperState::Offline { .. } => offline += 1,
-                SafekeeperState::Available { .. } => {}
-            }
-        }
-
-        tracing::info!(
-            "Heartbeat round complete for {} safekeepers, {} offline",
-            new_state.len(),
-            offline
-        );
-
-        let mut deltas = Vec::new();
-        let now = Instant::now();
-        for (node_id, sk_state) in new_state.iter_mut() {
-            use std::collections::hash_map::Entry::*;
-            let entry = self.state.entry(*node_id);
-
-            let mut needs_update = false;
-            match entry {
-                Occupied(ref occ) => match (occ.get(), &sk_state) {
-                    (SafekeeperState::Offline, SafekeeperState::Offline) => {}
-                    (SafekeeperState::Available { last_seen_at, .. }, SafekeeperState::Offline) => {
-                        if now - *last_seen_at >= self.max_offline_interval {
-                            deltas.push((*node_id, sk_state.clone()));
-                            needs_update = true;
-                        }
-                    }
-                    _ => {
-                        deltas.push((*node_id, sk_state.clone()));
-                        needs_update = true;
-                    }
-                },
-                Vacant(_) => {
-                    // This is a new node. Don't generate a delta for it.
-                    deltas.push((*node_id, sk_state.clone()));
-                }
-            }
-
-            match entry {
-                Occupied(mut occ) if needs_update => {
-                    (*occ.get_mut()) = sk_state.clone();
-                }
-                Vacant(vac) => {
-                    vac.insert(sk_state.clone());
-                }
-                _ => {}
-            }
-        }
-
-        Ok(AvailablityDeltas(deltas))
-    }
-}

storage_controller/src/http.rs

@@ -516,17 +516,6 @@ async fn handle_tenant_timeline_block_unblock_gc(
     json_response(StatusCode::OK, ())
 }
 
-// For metric labels where we would like to include the approximate path, but exclude high-cardinality fields like query parameters
-// and tenant/timeline IDs.  Since we are proxying to arbitrary paths, we don't have routing templates to
-// compare to, so we can just filter out our well known ID format with regexes.
-fn path_without_ids(path: &str) -> String {
-    static ID_REGEX: std::sync::OnceLock<regex::Regex> = std::sync::OnceLock::new();
-    ID_REGEX
-        .get_or_init(|| regex::Regex::new(r"([0-9a-fA-F]{32}(-[0-9]{4})?|\?.*)").unwrap())
-        .replace_all(path, "")
-        .to_string()
-}
-
 async fn handle_tenant_timeline_passthrough(
     service: Arc<Service>,
     req: Request<Body>,
@@ -562,7 +551,10 @@ async fn handle_tenant_timeline_passthrough(
         .metrics_group
         .storage_controller_passthrough_request_latency;
 
-    let path_label = path_without_ids(&path)
+    // This is a bit awkward. We remove the param from the request
+    // and join the words by '_' to get a label for the request.
+    let just_path = path.replace(&tenant_shard_str, "");
+    let path_label = just_path
         .split('/')
         .filter(|token| !token.is_empty())
         .collect::<Vec<_>>()
@@ -2097,16 +2089,3 @@ pub fn make_router(
             )
         })
 }
-
-#[cfg(test)]
-mod test {
-
-    use super::path_without_ids;
-
-    #[test]
-    fn test_path_without_ids() {
-        assert_eq!(path_without_ids("/v1/tenant/1a2b3344556677881122334455667788/timeline/AA223344556677881122334455667788"), "/v1/tenant//timeline/");
-        assert_eq!(path_without_ids("/v1/tenant/1a2b3344556677881122334455667788-0108/timeline/AA223344556677881122334455667788"), "/v1/tenant//timeline/");
-        assert_eq!(path_without_ids("/v1/tenant/1a2b3344556677881122334455667788-0108/timeline/AA223344556677881122334455667788?parameter=foo"), "/v1/tenant//timeline/");
-    }
-}

storage_controller/src/lib.rs

@@ -17,8 +17,6 @@ mod pageserver_client;
 mod peer_client;
 pub mod persistence;
 mod reconciler;
-mod safekeeper;
-mod safekeeper_client;
 mod scheduler;
 mod schema;
 pub mod service;

storage_controller/src/main.rs

@@ -12,8 +12,7 @@ use storage_controller::persistence::Persistence;
 use storage_controller::service::chaos_injector::ChaosInjector;
 use storage_controller::service::{
     Config, Service, HEARTBEAT_INTERVAL_DEFAULT, LONG_RECONCILE_THRESHOLD_DEFAULT,
-    MAX_OFFLINE_INTERVAL_DEFAULT, MAX_WARMING_UP_INTERVAL_DEFAULT,
-    PRIORITY_RECONCILER_CONCURRENCY_DEFAULT, RECONCILER_CONCURRENCY_DEFAULT,
+    MAX_OFFLINE_INTERVAL_DEFAULT, MAX_WARMING_UP_INTERVAL_DEFAULT, RECONCILER_CONCURRENCY_DEFAULT,
 };
 use tokio::signal::unix::SignalKind;
 use tokio_util::sync::CancellationToken;
@@ -76,14 +75,10 @@ struct Cli {
     #[arg(long)]
     split_threshold: Option<u64>,
 
-    /// Maximum number of normal-priority reconcilers that may run in parallel
+    /// Maximum number of reconcilers that may run in parallel
     #[arg(long)]
     reconciler_concurrency: Option<usize>,
 
-    /// Maximum number of high-priority reconcilers that may run in parallel
-    #[arg(long)]
-    priority_reconciler_concurrency: Option<usize>,
-
     /// How long to wait for the initial database connection to be available.
     #[arg(long, default_value = "5s")]
     db_connect_timeout: humantime::Duration,
@@ -294,9 +289,6 @@ async fn async_main() -> anyhow::Result<()> {
         reconciler_concurrency: args
             .reconciler_concurrency
             .unwrap_or(RECONCILER_CONCURRENCY_DEFAULT),
-        priority_reconciler_concurrency: args
-            .priority_reconciler_concurrency
-            .unwrap_or(PRIORITY_RECONCILER_CONCURRENCY_DEFAULT),
         split_threshold: args.split_threshold,
         neon_local_repo_dir: args.neon_local_repo_dir,
         max_secondary_lag_bytes: args.max_secondary_lag_bytes,

storage_controller/src/metrics.rs

@@ -80,11 +80,6 @@ pub(crate) struct StorageControllerMetricGroup {
     pub(crate) storage_controller_pageserver_request_error:
         measured::CounterVec<PageserverRequestLabelGroupSet>,
 
-    /// Count of HTTP requests to the safekeeper that resulted in an error,
-    /// broken down by the safekeeper node id, request name and method
-    pub(crate) storage_controller_safekeeper_request_error:
-        measured::CounterVec<PageserverRequestLabelGroupSet>,
-
     /// Latency of HTTP requests to the pageserver, broken down by pageserver
     /// node id, request name and method. This include both successful and unsuccessful
     /// requests.
@@ -92,13 +87,6 @@ pub(crate) struct StorageControllerMetricGroup {
     pub(crate) storage_controller_pageserver_request_latency:
         measured::HistogramVec<PageserverRequestLabelGroupSet, 5>,
 
-    /// Latency of HTTP requests to the safekeeper, broken down by safekeeper
-    /// node id, request name and method. This include both successful and unsuccessful
-    /// requests.
-    #[metric(metadata = histogram::Thresholds::exponential_buckets(0.1, 2.0))]
-    pub(crate) storage_controller_safekeeper_request_latency:
-        measured::HistogramVec<PageserverRequestLabelGroupSet, 5>,
-
     /// Count of pass-through HTTP requests to the pageserver that resulted in an error,
     /// broken down by the pageserver node id, request name and method
     pub(crate) storage_controller_passthrough_request_error:

storage_controller/src/persistence.rs

@@ -1185,6 +1185,23 @@ impl Persistence {
         Ok(safekeepers)
     }
 
+    pub(crate) async fn safekeeper_get(
+        &self,
+        id: i64,
+    ) -> Result<SafekeeperPersistence, DatabaseError> {
+        use crate::schema::safekeepers::dsl::{id as id_column, safekeepers};
+        self.with_conn(move |conn| {
+            Box::pin(async move {
+                Ok(safekeepers
+                    .filter(id_column.eq(&id))
+                    .select(SafekeeperPersistence::as_select())
+                    .get_result(conn)
+                    .await?)
+            })
+        })
+        .await
+    }
+
     pub(crate) async fn safekeeper_upsert(
         &self,
         record: SafekeeperUpsert,
@@ -1537,21 +1554,6 @@ pub(crate) struct SafekeeperPersistence {
 }
 
 impl SafekeeperPersistence {
-    pub(crate) fn from_upsert(
-        upsert: SafekeeperUpsert,
-        scheduling_policy: SkSchedulingPolicy,
-    ) -> Self {
-        crate::persistence::SafekeeperPersistence {
-            id: upsert.id,
-            region_id: upsert.region_id,
-            version: upsert.version,
-            host: upsert.host,
-            port: upsert.port,
-            http_port: upsert.http_port,
-            availability_zone_id: upsert.availability_zone_id,
-            scheduling_policy: String::from(scheduling_policy),
-        }
-    }
     pub(crate) fn as_describe_response(&self) -> Result<SafekeeperDescribeResponse, DatabaseError> {
         let scheduling_policy =
             SkSchedulingPolicy::from_str(&self.scheduling_policy).map_err(|e| {

storage_controller/src/reconciler.rs

@@ -1,7 +1,7 @@
 use crate::pageserver_client::PageserverClient;
 use crate::persistence::Persistence;
 use crate::{compute_hook, service};
-use pageserver_api::controller_api::{AvailabilityZone, MigrationConfig, PlacementPolicy};
+use pageserver_api::controller_api::{AvailabilityZone, PlacementPolicy};
 use pageserver_api::models::{
     LocationConfig, LocationConfigMode, LocationConfigSecondary, TenantConfig, TenantWaitLsnRequest,
 };
@@ -91,10 +91,9 @@ pub(crate) struct ReconcilerConfigBuilder {
 }
 
 impl ReconcilerConfigBuilder {
-    /// Priority is special: you must pick one thoughtfully, do not just use 'normal' as the default
-    pub(crate) fn new(priority: ReconcilerPriority) -> Self {
+    pub(crate) fn new() -> Self {
         Self {
-            config: ReconcilerConfig::new(priority),
+            config: ReconcilerConfig::default(),
         }
     }
 
@@ -130,18 +129,8 @@ impl ReconcilerConfigBuilder {
     }
 }
 
-// Higher priorities are used for user-facing tasks, so that a long backlog of housekeeping work (e.g. reconciling on startup, rescheduling
-// things on node changes) does not starve user-facing tasks.
-#[derive(Debug, Copy, Clone)]
-pub(crate) enum ReconcilerPriority {
-    Normal,
-    High,
-}
-
-#[derive(Debug, Copy, Clone)]
+#[derive(Default, Debug, Copy, Clone)]
 pub(crate) struct ReconcilerConfig {
-    pub(crate) priority: ReconcilerPriority,
-
     // During live migration give up on warming-up the secondary
     // after this timeout.
     secondary_warmup_timeout: Option<Duration>,
@@ -156,18 +145,6 @@ pub(crate) struct ReconcilerConfig {
 }
 
 impl ReconcilerConfig {
-    /// Configs are always constructed with an explicit priority, to force callers to think about whether
-    /// the operation they're scheduling is high-priority or not. Normal priority is not a safe default, because
-    /// scheduling something user-facing at normal priority can result in it getting starved out by background work.
-    pub(crate) fn new(priority: ReconcilerPriority) -> Self {
-        Self {
-            priority,
-            secondary_warmup_timeout: None,
-            secondary_download_request_timeout: None,
-            tenant_creation_hint: false,
-        }
-    }
-
     pub(crate) fn get_secondary_warmup_timeout(&self) -> Duration {
         const SECONDARY_WARMUP_TIMEOUT_DEFAULT: Duration = Duration::from_secs(300);
         self.secondary_warmup_timeout
@@ -185,24 +162,6 @@ impl ReconcilerConfig {
     }
 }
 
-impl From<&MigrationConfig> for ReconcilerConfig {
-    fn from(value: &MigrationConfig) -> Self {
-        // Run reconciler at high priority because MigrationConfig comes from human requests that should
-        // be presumed urgent.
-        let mut builder = ReconcilerConfigBuilder::new(ReconcilerPriority::High);
-
-        if let Some(timeout) = value.secondary_warmup_timeout {
-            builder = builder.secondary_warmup_timeout(timeout)
-        }
-
-        if let Some(timeout) = value.secondary_download_request_timeout {
-            builder = builder.secondary_download_request_timeout(timeout)
-        }
-
-        builder.build()
-    }
-}
-
 /// RAII resource units granted to a Reconciler, which it should keep alive until it finishes doing I/O
 pub(crate) struct ReconcileUnits {
     _sem_units: tokio::sync::OwnedSemaphorePermit,

storage_controller/src/safekeeper.rs

@@ -1,139 +0,0 @@
-use std::{str::FromStr, time::Duration};
-
-use pageserver_api::controller_api::{SafekeeperDescribeResponse, SkSchedulingPolicy};
-use reqwest::StatusCode;
-use safekeeper_client::mgmt_api;
-use tokio_util::sync::CancellationToken;
-use utils::{backoff, id::NodeId, logging::SecretString};
-
-use crate::{
-    heartbeater::SafekeeperState,
-    persistence::{DatabaseError, SafekeeperPersistence},
-    safekeeper_client::SafekeeperClient,
-};
-
-#[derive(Clone)]
-pub struct Safekeeper {
-    pub(crate) skp: SafekeeperPersistence,
-    cancel: CancellationToken,
-    listen_http_addr: String,
-    listen_http_port: u16,
-    id: NodeId,
-    availability: SafekeeperState,
-}
-
-impl Safekeeper {
-    pub(crate) fn from_persistence(skp: SafekeeperPersistence, cancel: CancellationToken) -> Self {
-        Self {
-            cancel,
-            listen_http_addr: skp.host.clone(),
-            listen_http_port: skp.http_port as u16,
-            id: NodeId(skp.id as u64),
-            skp,
-            availability: SafekeeperState::Offline,
-        }
-    }
-    pub(crate) fn base_url(&self) -> String {
-        format!("http://{}:{}", self.listen_http_addr, self.listen_http_port)
-    }
-
-    pub(crate) fn get_id(&self) -> NodeId {
-        self.id
-    }
-    pub(crate) fn describe_response(&self) -> Result<SafekeeperDescribeResponse, DatabaseError> {
-        self.skp.as_describe_response()
-    }
-    pub(crate) fn set_availability(&mut self, availability: SafekeeperState) {
-        self.availability = availability;
-    }
-    /// Perform an operation (which is given a [`SafekeeperClient`]) with retries
-    pub(crate) async fn with_client_retries<T, O, F>(
-        &self,
-        mut op: O,
-        jwt: &Option<SecretString>,
-        warn_threshold: u32,
-        max_retries: u32,
-        timeout: Duration,
-        cancel: &CancellationToken,
-    ) -> mgmt_api::Result<T>
-    where
-        O: FnMut(SafekeeperClient) -> F,
-        F: std::future::Future<Output = mgmt_api::Result<T>>,
-    {
-        fn is_fatal(e: &mgmt_api::Error) -> bool {
-            use mgmt_api::Error::*;
-            match e {
-                ReceiveBody(_) | ReceiveErrorBody(_) => false,
-                ApiError(StatusCode::SERVICE_UNAVAILABLE, _)
-                | ApiError(StatusCode::GATEWAY_TIMEOUT, _)
-                | ApiError(StatusCode::REQUEST_TIMEOUT, _) => false,
-                ApiError(_, _) => true,
-                Cancelled => true,
-            }
-        }
-
-        backoff::retry(
-            || {
-                let http_client = reqwest::ClientBuilder::new()
-                    .timeout(timeout)
-                    .build()
-                    .expect("Failed to construct HTTP client");
-
-                let client = SafekeeperClient::from_client(
-                    self.get_id(),
-                    http_client,
-                    self.base_url(),
-                    jwt.clone(),
-                );
-
-                let node_cancel_fut = self.cancel.cancelled();
-
-                let op_fut = op(client);
-
-                async {
-                    tokio::select! {
-                        r = op_fut=> {r},
-                        _ = node_cancel_fut => {
-                        Err(mgmt_api::Error::Cancelled)
-                    }}
-                }
-            },
-            is_fatal,
-            warn_threshold,
-            max_retries,
-            &format!(
-                "Call to node {} ({}:{}) management API",
-                self.id, self.listen_http_addr, self.listen_http_port
-            ),
-            cancel,
-        )
-        .await
-        .unwrap_or(Err(mgmt_api::Error::Cancelled))
-    }
-
-    pub(crate) fn update_from_record(&mut self, record: crate::persistence::SafekeeperUpsert) {
-        let crate::persistence::SafekeeperUpsert {
-            active: _,
-            availability_zone_id: _,
-            host,
-            http_port,
-            id,
-            port: _,
-            region_id: _,
-            version: _,
-        } = record.clone();
-        if id != self.id.0 as i64 {
-            // The way the function is called ensures this. If we regress on that, it's a bug.
-            panic!(
-                "id can't be changed via update_from_record function: {id} != {}",
-                self.id.0
-            );
-        }
-        self.skp = crate::persistence::SafekeeperPersistence::from_upsert(
-            record,
-            SkSchedulingPolicy::from_str(&self.skp.scheduling_policy).unwrap(),
-        );
-        self.listen_http_port = http_port as u16;
-        self.listen_http_addr = host;
-    }
-}

storage_controller/src/safekeeper_client.rs

@@ -1,105 +0,0 @@
-use crate::metrics::PageserverRequestLabelGroup;
-use safekeeper_api::models::{SafekeeperUtilization, TimelineCreateRequest, TimelineStatus};
-use safekeeper_client::mgmt_api::{Client, Result};
-use utils::{
-    id::{NodeId, TenantId, TimelineId},
-    logging::SecretString,
-};
-
-/// Thin wrapper around [`safekeeper_client::mgmt_api::Client`]. It allows the storage
-/// controller to collect metrics in a non-intrusive manner.
-///
-/// Analogous to [`crate::pageserver_client::PageserverClient`].
-#[derive(Debug, Clone)]
-pub(crate) struct SafekeeperClient {
-    inner: Client,
-    node_id_label: String,
-}
-
-macro_rules! measured_request {
-    ($name:literal, $method:expr, $node_id: expr, $invoke:expr) => {{
-        let labels = PageserverRequestLabelGroup {
-            pageserver_id: $node_id,
-            path: $name,
-            method: $method,
-        };
-
-        let latency = &crate::metrics::METRICS_REGISTRY
-            .metrics_group
-            .storage_controller_safekeeper_request_latency;
-        let _timer_guard = latency.start_timer(labels.clone());
-
-        let res = $invoke;
-
-        if res.is_err() {
-            let error_counters = &crate::metrics::METRICS_REGISTRY
-                .metrics_group
-                .storage_controller_pageserver_request_error;
-            error_counters.inc(labels)
-        }
-
-        res
-    }};
-}
-
-impl SafekeeperClient {
-    #[allow(dead_code)]
-    pub(crate) fn new(
-        node_id: NodeId,
-        mgmt_api_endpoint: String,
-        jwt: Option<SecretString>,
-    ) -> Self {
-        Self {
-            inner: Client::from_client(reqwest::Client::new(), mgmt_api_endpoint, jwt),
-            node_id_label: node_id.0.to_string(),
-        }
-    }
-
-    pub(crate) fn from_client(
-        node_id: NodeId,
-        raw_client: reqwest::Client,
-        mgmt_api_endpoint: String,
-        jwt: Option<SecretString>,
-    ) -> Self {
-        Self {
-            inner: Client::from_client(raw_client, mgmt_api_endpoint, jwt),
-            node_id_label: node_id.0.to_string(),
-        }
-    }
-
-    #[allow(dead_code)]
-    pub(crate) async fn create_timeline(
-        &self,
-        req: &TimelineCreateRequest,
-    ) -> Result<TimelineStatus> {
-        measured_request!(
-            "create_timeline",
-            crate::metrics::Method::Post,
-            &self.node_id_label,
-            self.inner.create_timeline(req).await
-        )
-    }
-
-    #[allow(dead_code)]
-    pub(crate) async fn delete_timeline(
-        &self,
-        tenant_id: TenantId,
-        timeline_id: TimelineId,
-    ) -> Result<TimelineStatus> {
-        measured_request!(
-            "delete_timeline",
-            crate::metrics::Method::Delete,
-            &self.node_id_label,
-            self.inner.delete_timeline(tenant_id, timeline_id).await
-        )
-    }
-
-    pub(crate) async fn get_utilization(&self) -> Result<SafekeeperUtilization> {
-        measured_request!(
-            "utilization",
-            crate::metrics::Method::Get,
-            &self.node_id_label,
-            self.inner.utilization().await
-        )
-    }
-}