incorporate some github comments in the doc

This is not ready, I'm still collecting an organizing my own
thoughts. I will update when it's ready for review.

That said, feel free to leave comments already if you wish.

.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

.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/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

docs/rfcs/042-compute-pageserver-communicator.md

@@ -0,0 +1,203 @@
+# Compute <-> Pageserver Communicator Rewrite
+
+## Summary
+
+## Motivation
+
+- prefetching logic is complicated
+- handling async libpq connections in C code are difficult and error-prone
+- only few people are comfortable working on the code
+- new AIO (maybe) coming up in Postgres v18
+- cannot process prefetch replies until another I/O function is called. Makes it impossible to accurately measure when a reply was received
+- every backend opens a separate connection to pageservers -> lots of connections, first query in backend is slow
+
+- desire for better protocol, not libpq-based
+
+- By writing the "communicator" as a separate rust module, it can be reused in
+  tests, outside PostgreSQL.
+
+## Non Goals (if relevant)
+
+- We will keep LFC unmodified for now. It might be a good idea to rewrite it
+  too, but it's out of scope here.
+
+- We should consider a similar rewrite for the walproposer - safekeeper
+  communication, but it's out of scope for this RFC
+
+## Impacted components (e.g. pageserver, safekeeper, console, etc)
+
+- Most changes are to the neon extension in compute.
+
+- Pageserver, to implement the new protocol.
+
+## Proposed implementation
+
+- we will use the new implementation with all PostgreSQL versions.
+- we will have a feature flag to switch between old and new communicator. Once we're
+  comfortable with the new communicator, remove old code and protocol.
+
+- What about relation size cache? Out of scope? Or move it to the communicator process,
+  and have smgrnblocks() requests always through communicator process?
+
+### Communicator process
+
+There is one communicator process in the Postgres server. It's a background
+worker process. It handles all communication with the pageservers.
+
+The communicator process is written in a mix of C and Rust. Mostly in Rust, but
+some C code is needed to interface with the Postgres facilities. For example:
+- logging
+- error handling (in a very limited form, we don't want to ereport() on most errors)
+- expose a shared memory area for IPC
+- signal other processes
+
+We will write unsafe rust or C glue code for those facilities, which allow us to
+write the rest of the communicator in safe rust.
+
+The Rust parts of the communicator process can use multiple threads and
+tokio. The glue code is written taken that into account, making it safe.
+
+pqrx is a rust crate for writing Postgres extensions in Rust. We will _not_ use
+that. It's a fine crate, good for most extensions, but I don't think we need
+most of the facilities that it provides. Our wrappers are more low-level than
+what most extensions need. We don't expose SQL functions or types from this
+extension for example.
+
+### Communicator <-> backend interface
+
+The backends and the communicator process communicate via shared memory. Each
+backend has a fixed number of "request slots", forming a ring. When a backend
+wants to perform an I/O, it writes the request details like blk # and LSN, to
+the next available slot. The request also includes a pointer or buffer ID where
+the resulting page should be written. The backend then wakes up the
+communicator, with a signal/futex/latch or something, telling the communicator
+that it has work to do.
+
+The communicator picks up the request from the backend's ring, and performs
+it. It writes result page to the address requested by the backend (most likely a
+shared buffer that the backend is holding a lock on), marks the request as
+completed, and wakes up the backend.
+
+In this design, because each backend has its own small ring, a backend doesn't
+need to do any locking to manipulate the request slots. Similarly, after a
+request has been submitted, the communicator has temporary ownership of the
+request slot, and doesn't need to do locking on it.
+
+This design is somewhat similar to how the upcoming AIO patches in PostgreSQL
+will work. That should make it easy to adapt to new PostgreSQL versions.
+
+In the above example, I assumed a GetPage request, but everything applies to
+other other request types like "smgrnblocks" too.
+
+Q: How we are going to handle backend termination while it is waiting for response? Should we allow it or wait request completion? If PS is down, it can take quite long time during which user will not be able to interrupt the query.
+
+### Prefetching
+
+A backend can also issue a "blind" prefetch request. When a communicator
+processes a blind prefetch request, it starts the GetPage request and writes the
+result to a local buffer within the communicator process. But it could also
+decide to do nothing, or to schedule the request with a lower priority. It
+doesn't give any result back to the requesting backend, hence it's "blind".
+Later, when the backend - or a different backend - requests the page that was
+prefetched and the prefetch was performed and completed, the communicator
+process can satisfy the request quickly from the private buffer.
+
+In this design, the "prefetch" slots are shared by all backends. If one backend
+issues a prefetch request but never consumes it, but another backend reads the
+same page, the prefetch can be used to satisfy the request. (In our current
+implementaiton, it would be wasted, and the same GetPage is performed
+twice. https://github.com/neondatabase/neon/pull/10442 helps with that, but
+doesn't fully fix the problem)
+
+### PostgreSQL versions < 17
+
+- In 16 and below, prefetching calls are made without holding the buffer pinned.
+Backends will perform "blind" prefetch requests for smgrprefetch().
+
+### PostgreSQL version 17
+
+In version 17, when prefetching is requested, the pages are already pinned in
+the buffer manager. We possibly could write the page directly to the shared
+buffer, but there's a risk that the backend stops the scan and releases the pins
+without ever performing the real I/Os. Because of that, backends will perform
+blind prefetch requests like in v16; we can't easily take advantage of the
+pinned buffer.
+
+### PostgreSQL version 18, if the AIO patches are committed
+
+With the AIO patches, prefetching is no longer performed with posix_fadvise
+calls. The backends will start the prefetch I/Os "for real", into the locked
+shared buffer. On completion of an AIO, the process that processes the
+completion will have to call a small callback routine that releases the buffer
+lock and wakes up any processes waiting on the I/O. It'll require some care to
+execute that safely from the communicator process.
+
+### Compute <-> Pageserver Protocol
+
+As part of the project, we will change the protocol. Desires for new protocol:
+
+- Use protobuf or something else more standard. Maybe gRPC. So that we can use
+  standard tools like Wireshark to easily analyze the traffic.
+
+- Batching. Have capability to request more than one page in one request.
+
+In principle, changing the protocol is an independent change from the new
+communicator process. But it makes sense to do at the same time:
+
+- Switching to Rust in the communicator process makes it possible to use
+  existing libraries
+
+- Using a library might help with managing the pool of pageserver connnection,
+  so we want need to implement that ourselves
+
+### Reliability, failure modes and corner cases (if relevant)
+
+### Interaction/Sequence diagram (if relevant)
+
+### Scalability (if relevant)
+
+- Could the single communicator process become a bottleneck? In the new v18 AIO
+  system, the process needs to execute all the I/O completion callbacks. They're
+  very short, but I still wonder if a single process can handle it.
+
+- Goal is to sustain ~2.5GB/s bandwidth (typical EC2 NIC).
+   - John: I'd presume a single process to be capable of that if it's just passing buffers around. If we needed more than one in future it would probably be quite a small lift to make that happen (but I bet we never do). (I don't fully know what's involved in the execute all the I/O completion callbacks part though)
+
+
+### Security implications (if relevant)
+
+- We currently use libpq authentication with a JWT token. We can continue to use
+  the token for authentication in the new protocol.
+
+### Unresolved questions (if relevant)
+
+## Alternative implementation (if relevant)
+
+I think UDP might also be a good fit for the protocol. No overhead of
+establishing or holding a connection. No head-of-line blocking; prefetch
+requests can be processed with lower priority. We would control our own
+destiny. But it has its own set of challenges: congestion control,
+authentication & encryption.
+
+## Pros/cons of proposed approaches (if relevant)
+
+## Definition of Done (if relevant)
+
+New communicator has replaced the old code, deployed in production, old protocol
+support is removed.
+
+Implentation phases:
+
+- Implement new protocol in pageserver. In first prototype, maybe just
+  wrap/convert the existing message types into HTTP+protobuf, to keep it simple.
+
+- Implement the C/Rust wrappers needed to launch the communicator as a background
+  worker process, with access to shard memory.
+
+- Implement a simple request / response interface in shared memory between the
+  backends and the communicator.
+
+- Implement a minimalistic communicator: hold one connection to
+  pageserver/shard. No prefetching. Process one request at a time
+
+- Improve the communicator: multiple threads, multiple connections, prefetching

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/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/layer_map.rs

@@ -570,12 +570,8 @@ impl LayerMap {
         self.historic.iter()
     }
 
-    pub fn riter_historic_layers(&self) -> impl '_ + Iterator<Item = Arc<PersistentLayerDesc>> {
-        self.historic.riter()
-    }
-
     /// Get a ref counted pointer for the first in memory layer that matches the provided predicate.
-    pub(crate) fn find_in_memory_layer<Pred>(&self, mut pred: Pred) -> Option<Arc<InMemoryLayer>>
+    pub fn find_in_memory_layer<Pred>(&self, mut pred: Pred) -> Option<Arc<InMemoryLayer>>
     where
         Pred: FnMut(&Arc<InMemoryLayer>) -> bool,
     {
@@ -904,24 +900,6 @@ impl LayerMap {
         Ok(())
     }
 
-    /// Efficiency: this is a single btreemap walk to the end of the map in the common case where
-    /// we are queried for image layers after the start of an ephemeral layer.  In the general case
-    /// where we are called with some arbitrary LSN, this function is O(N) -- so don't use it like that.
-    pub(crate) fn get_newest_image_after(&self, lsn: Lsn) -> Option<Arc<PersistentLayerDesc>> {
-        // TODO: an efficient equivalent, this is a crude placeholder
-        for layer in self.riter_historic_layers() {
-            if !layer.is_delta() && layer.image_layer_lsn() >= lsn {
-                return Some(layer);
-            }
-
-            if layer.lsn_range.start < lsn {
-                // We are past the layers that could possibly intersect with the requested bound
-                break;
-            }
-        }
-        None
-    }
-
     /// `read_points` represent the tip of a timeline and any branch points, i.e. the places
     /// where we expect to serve reads.
     ///

pageserver/src/tenant/layer_map/historic_layer_coverage.rs

@@ -509,18 +509,6 @@ impl<Value: Clone> BufferedHistoricLayerCoverage<Value> {
         self.layers.values().cloned()
     }
 
-    /// Iterate all the layers in reverse order (newest LSNs first)
-    pub fn riter(&self) -> impl '_ + Iterator<Item = Value> {
-        // NOTE we can actually perform this without rebuilding,
-        //      but it's not necessary for now.
-        if !self.buffer.is_empty() {
-            panic!("rebuild pls")
-        }
-
-        // TODO: is cloned() really needed?
-        self.layers.values().rev().cloned()
-    }
-
     /// Return a reference to a queryable map, assuming all updates
     /// have already been processed using self.rebuild()
     pub fn get(&self) -> anyhow::Result<&HistoricLayerCoverage<Value>> {

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],
             });
@@ -3775,14 +3657,12 @@ impl Timeline {
         let mut completed_keyspace = KeySpace::default();
         let mut image_covered_keyspace = KeySpaceRandomAccum::new();
 
-        let mut in_memory_layers_considered = Vec::new();
-
         // Prevent GC from progressing while visiting the current timeline.
         // If we are GC-ing because a new image layer was added while traversing
         // 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;
@@ -3812,34 +3692,12 @@ impl Timeline {
 
                 let in_memory_layer = layers.find_in_memory_layer(|l| {
                     let start_lsn = l.get_lsn_range().start;
-                    !in_memory_layers_considered.contains(&start_lsn) && cont_lsn > start_lsn
+                    cont_lsn > start_lsn
                 });
 
                 match in_memory_layer {
                     Some(l) => {
-                        in_memory_layers_considered.push(l.get_lsn_range().start);
-
-                        // Search for image layers that overlap with the in-memory layer: this is rare but permitted, and
-                        // we must bound the `lsn_range` of this layer to avoid skipping past the image layer.
-                        // TODO: a narrower search that only hits on image layers matching `unmapped_keyspace`
-                        let lsn_range = if let Some(image) =
-                            layers.get_newest_image_after(l.get_lsn_range().start)
-                        {
-                            // Note that this does not guarantee serving a read from an image layer, just that we will
-                            // not skip considering thge image layer in our Fringe.  We can still end up doing walredo work
-                            // in spite of the presence of an image layer, if the inmemory layers we visit contain enough
-                            // information to fully construct a page.  For example:
-                            //  - ephemeral layer contains I1, D1, D2, <LSN X>
-                            //  - image layer at LSN X contains image equal to I2
-                            //  - we will end up doing a walredo of I1+D1+D2, rather than reading from the image layer
-                            //
-                            //  This is not a problem for correctness, and is rare enough that the wasted time doing walredo
-                            //  doesn't matter.
-                            image.get_lsn_range().start + 1..cont_lsn
-                        } else {
-                            l.get_lsn_range().start..cont_lsn
-                        };
-
+                        let lsn_range = l.get_lsn_range().start..cont_lsn;
                         fringe.update(
                             ReadableLayer::InMemoryLayer(l),
                             unmapped_keyspace.clone(),
@@ -4491,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!(
@@ -5719,7 +5577,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,
                 }
             }
@@ -5840,7 +5698,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}",
@@ -5854,7 +5712,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!(
@@ -6794,32 +6652,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();
@@ -6893,7 +6737,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());
 
@@ -6908,144 +6752,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;
-    }
-}