Resolve merge conflicts

Co-authored-by: Heikki Linnakangas <heikki@neon.tech>

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

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

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

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

.github/workflows/_meta.yml

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

.github/workflows/build_and_test.yml

@@ -323,8 +323,6 @@ jobs:
           PERF_TEST_RESULT_CONNSTR: "${{ secrets.PERF_TEST_RESULT_CONNSTR }}"
           TEST_RESULT_CONNSTR: "${{ secrets.REGRESS_TEST_RESULT_CONNSTR_NEW }}"
           PAGESERVER_VIRTUAL_FILE_IO_ENGINE: tokio-epoll-uring
-          PAGESERVER_GET_VECTORED_CONCURRENT_IO: sidecar-task
-          PAGESERVER_VIRTUAL_FILE_IO_MODE: direct
           SYNC_BETWEEN_TESTS: true
       # XXX: no coverage data handling here, since benchmarks are run on release builds,
       # while coverage is currently collected for the debug ones

.github/workflows/fast-forward.yml

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

.github/workflows/pg-clients.yml

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

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

@@ -1,93 +0,0 @@
-name: Random Operations Test
-
-on:
-  schedule:
-    # * is a special character in YAML so you have to quote this string
-    #          ┌───────────── minute (0 - 59)
-    #          │  ┌───────────── hour (0 - 23)
-    #          │  │  ┌───────────── day of the month (1 - 31)
-    #          │  │  │ ┌───────────── month (1 - 12 or JAN-DEC)
-    #          │  │  │ │ ┌───────────── day of the week (0 - 6 or SUN-SAT)
-    - cron:  '23 */2 * * *' # runs every 2 hours
-  workflow_dispatch:
-    inputs:
-      random_seed:
-        type: number
-        description: 'The random seed'
-        required: false
-        default: 0
-      num_operations:
-        type: number
-        description: "The number of operations to test"
-        default: 250
-
-defaults:
-  run:
-    shell: bash -euxo pipefail {0}
-
-permissions: {}
-
-env:
-  DEFAULT_PG_VERSION: 16
-  PLATFORM: neon-captest-new
-  AWS_DEFAULT_REGION: eu-central-1
-
-jobs:
-  run-random-rests:
-    env:
-      POSTGRES_DISTRIB_DIR: /tmp/neon/pg_install
-    runs-on: small
-    permissions:
-      id-token: write
-      statuses: write
-
-    strategy:
-      fail-fast: false
-      matrix:
-        pg-version: [16, 17]
-
-    container:
-      image: ghcr.io/neondatabase/build-tools:pinned-bookworm
-      credentials:
-        username: ${{ github.actor }}
-        password: ${{ secrets.GITHUB_TOKEN }}
-      options: --init
-    steps:
-      - name: Harden the runner (Audit all outbound calls)
-        uses: step-security/harden-runner@4d991eb9b905ef189e4c376166672c3f2f230481 # v2.11.0
-        with:
-          egress-policy: audit
-
-      - uses: actions/checkout@11bd71901bbe5b1630ceea73d27597364c9af683 # v4.2.2
-
-      - name: Download Neon artifact
-        uses: ./.github/actions/download
-        with:
-          name: neon-${{ runner.os }}-${{ runner.arch }}-release-artifact
-          path: /tmp/neon/
-          prefix: latest
-          aws-oicd-role-arn: ${{ vars.DEV_AWS_OIDC_ROLE_ARN }}
-
-      - name: Run tests
-        uses: ./.github/actions/run-python-test-set
-        with:
-          build_type: remote
-          test_selection: random_ops
-          run_in_parallel: false
-          extra_params: -m remote_cluster
-          pg_version: ${{ matrix.pg-version }}
-          aws-oicd-role-arn: ${{ vars.DEV_AWS_OIDC_ROLE_ARN }}
-        env:
-          NEON_API_KEY: ${{ secrets.NEON_STAGING_API_KEY }}
-          RANDOM_SEED: ${{ inputs.random_seed }}
-          NUM_OPERATIONS: ${{ inputs.num_operations }}
-
-      - name: Create Allure report
-        if: ${{ !cancelled() }}
-        id: create-allure-report
-        uses: ./.github/actions/allure-report-generate
-        with:
-          store-test-results-into-db: true
-          aws-oicd-role-arn: ${{ vars.DEV_AWS_OIDC_ROLE_ARN }}
-        env:
-          REGRESS_TEST_RESULT_CONNSTR_NEW: ${{ secrets.REGRESS_TEST_RESULT_CONNSTR_NEW }}

Cargo.lock

@@ -1432,6 +1432,7 @@ dependencies = [
  "pageserver_api",
  "pageserver_client",
  "pem",
+ "pkcs8 0.10.2",
  "postgres_backend",
  "postgres_connection",
  "regex",
@@ -1441,7 +1442,6 @@ dependencies = [
  "serde",
  "serde_json",
  "sha2",
- "spki 0.7.3",
  "storage_broker",
  "thiserror 1.0.69",
  "tokio",
@@ -2424,33 +2424,6 @@ dependencies = [
  "slab",
 ]
 
-[[package]]
-name = "gcp_auth"
-version = "0.12.3"
-source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "dbf67f30198e045a039264c01fb44659ce82402d7771c50938beb41a5ac87733"
-dependencies = [
- "async-trait",
- "base64 0.22.1",
- "bytes",
- "chrono",
- "home",
- "http 1.1.0",
- "http-body-util",
- "hyper 1.4.1",
- "hyper-rustls 0.27.5",
- "hyper-util",
- "ring",
- "rustls-pemfile 2.1.1",
- "serde",
- "serde_json",
- "thiserror 1.0.69",
- "tokio",
- "tracing",
- "tracing-futures",
- "url",
-]
-
 [[package]]
 name = "gen_ops"
 version = "0.4.0"
@@ -2749,15 +2722,6 @@ dependencies = [
  "digest",
 ]
 
-[[package]]
-name = "home"
-version = "0.5.11"
-source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "589533453244b0995c858700322199b2becb13b627df2851f64a2775d024abcf"
-dependencies = [
- "windows-sys 0.59.0",
-]
-
 [[package]]
 name = "hostname"
 version = "0.4.0"
@@ -2987,24 +2951,6 @@ dependencies = [
  "tower-service",
 ]
 
-[[package]]
-name = "hyper-rustls"
-version = "0.27.5"
-source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "2d191583f3da1305256f22463b9bb0471acad48a4e534a5218b9963e9c1f59b2"
-dependencies = [
- "futures-util",
- "http 1.1.0",
- "hyper 1.4.1",
- "hyper-util",
- "rustls 0.23.18",
- "rustls-native-certs 0.8.0",
- "rustls-pki-types",
- "tokio",
- "tokio-rustls 0.26.0",
- "tower-service",
-]
-
 [[package]]
 name = "hyper-timeout"
 version = "0.5.1"
@@ -3760,16 +3706,6 @@ version = "0.3.17"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "6877bb514081ee2a7ff5ef9de3281f14a4dd4bceac4c09388074a6b5df8a139a"
 
-[[package]]
-name = "mime_guess"
-version = "2.0.5"
-source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "f7c44f8e672c00fe5308fa235f821cb4198414e1c77935c1ab6948d3fd78550e"
-dependencies = [
- "mime",
- "unicase",
-]
-
 [[package]]
 name = "minimal-lexical"
 version = "0.2.1"
@@ -4349,7 +4285,6 @@ dependencies = [
  "pageserver_api",
  "pageserver_client",
  "pageserver_compaction",
- "pem",
  "pin-project-lite",
  "postgres-protocol",
  "postgres-types",
@@ -4417,7 +4352,6 @@ dependencies = [
  "humantime-serde",
  "itertools 0.10.5",
  "nix 0.27.1",
- "once_cell",
  "postgres_backend",
  "postgres_ffi",
  "rand 0.8.5",
@@ -5586,11 +5520,8 @@ dependencies = [
  "bytes",
  "camino",
  "camino-tempfile",
- "chrono",
  "futures",
  "futures-util",
- "gcp_auth",
- "http 1.1.0",
  "http-body-util",
  "http-types",
  "humantime-serde",
@@ -5611,9 +5542,7 @@ dependencies = [
  "tokio-util",
  "toml_edit",
  "tracing",
- "url",
  "utils",
- "uuid",
 ]
 
 [[package]]
@@ -5643,7 +5572,6 @@ dependencies = [
  "js-sys",
  "log",
  "mime",
- "mime_guess",
  "once_cell",
  "percent-encoding",
  "pin-project-lite",
@@ -6072,7 +6000,6 @@ dependencies = [
  "once_cell",
  "pageserver_api",
  "parking_lot 0.12.1",
- "pem",
  "postgres-protocol",
  "postgres_backend",
  "postgres_ffi",
@@ -7667,16 +7594,6 @@ dependencies = [
  "tracing-subscriber",
 ]
 
-[[package]]
-name = "tracing-futures"
-version = "0.2.5"
-source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "97d095ae15e245a057c8e8451bab9b3ee1e1f68e9ba2b4fbc18d0ac5237835f2"
-dependencies = [
- "pin-project",
- "tracing",
-]
-
 [[package]]
 name = "tracing-log"
 version = "0.2.0"
@@ -7830,12 +7747,6 @@ dependencies = [
  "libc",
 ]
 
-[[package]]
-name = "unicase"
-version = "2.8.1"
-source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "75b844d17643ee918803943289730bec8aac480150456169e647ed0b576ba539"
-
 [[package]]
 name = "unicode-bidi"
 version = "0.3.17"
@@ -8558,6 +8469,7 @@ dependencies = [
  "once_cell",
  "p256 0.13.2",
  "parquet",
+ "pkcs8 0.10.2",
  "prettyplease",
  "proc-macro2",
  "prost 0.13.3",

Cargo.toml

@@ -143,6 +143,7 @@ parquet_derive = "53"
 pbkdf2 = { version = "0.12.1", features = ["simple", "std"] }
 pem = "3.0.3"
 pin-project-lite = "0.2"
+pkcs8 = "0.10.2"
 pprof = { version = "0.14", features = ["criterion", "flamegraph", "frame-pointer", "prost-codec"] }
 procfs = "0.16"
 prometheus = {version = "0.13", default-features=false, features = ["process"]} # removes protobuf dependency
@@ -175,7 +176,6 @@ signal-hook = "0.3"
 smallvec = "1.11"
 smol_str = { version = "0.2.0", features = ["serde"] }
 socket2 = "0.5"
-spki = "0.7.3"
 strum = "0.26"
 strum_macros = "0.26"
 "subtle"  = "2.5.0"

README.md

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

clippy.toml

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

compute/patches/pg_anon.patch

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

compute/patches/pgvector.patch

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

compute/patches/rum.patch

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

compute/vm-image-spec-bookworm.yaml

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

compute/vm-image-spec-bullseye.yaml

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

compute_tools/src/bin/compute_ctl.rs

@@ -57,13 +57,24 @@ use tracing::{error, info};
 use url::Url;
 use utils::failpoint_support;
 
+// Compatibility hack: if the control plane specified any remote-ext-config
+// use the default value for extension storage proxy gateway.
+// Remove this once the control plane is updated to pass the gateway URL
+fn parse_remote_ext_config(arg: &str) -> Result<String> {
+    if arg.starts_with("http") {
+        Ok(arg.trim_end_matches('/').to_string())
+    } else {
+        Ok("http://pg-ext-s3-gateway".to_string())
+    }
+}
+
 #[derive(Parser)]
 #[command(rename_all = "kebab-case")]
 struct Cli {
     #[arg(short = 'b', long, default_value = "postgres", env = "POSTGRES_PATH")]
     pub pgbin: String,
 
-    #[arg(short = 'r', long)]
+    #[arg(short = 'r', long, value_parser = parse_remote_ext_config)]
     pub remote_ext_config: Option<String>,
 
     /// The port to bind the external listening HTTP server to. Clients running
@@ -105,7 +116,9 @@ struct Cli {
     #[arg(long)]
     pub set_disk_quota_for_fs: Option<String>,
 
-    #[arg(short = 'c', long)]
+    // TODO(tristan957): remove alias after compatibility tests are no longer
+    // an issue
+    #[arg(short = 'c', long, alias = "spec-path")]
     pub config: Option<OsString>,
 
     #[arg(short = 'i', long, group = "compute-id")]

compute_tools/src/compute.rs

@@ -641,26 +641,7 @@ impl ComputeNode {
 
                 let log_directory_path = Path::new(&self.params.pgdata).join("log");
                 let log_directory_path = log_directory_path.to_string_lossy().to_string();
-
-                // Add project_id,endpoint_id tag to identify the logs.
-                //
-                // These ids are passed from cplane,
-                // for backwards compatibility (old computes that don't have them),
-                // we set them to None.
-                // TODO: Clean up this code when all computes have them.
-                let tag: Option<String> = match (
-                    pspec.spec.project_id.as_deref(),
-                    pspec.spec.endpoint_id.as_deref(),
-                ) {
-                    (Some(project_id), Some(endpoint_id)) => {
-                        Some(format!("{project_id}/{endpoint_id}"))
-                    }
-                    (Some(project_id), None) => Some(format!("{project_id}/None")),
-                    (None, Some(endpoint_id)) => Some(format!("None,{endpoint_id}")),
-                    (None, None) => None,
-                };
-
-                configure_audit_rsyslog(log_directory_path.clone(), tag, &remote_endpoint)?;
+                configure_audit_rsyslog(log_directory_path.clone(), "hipaa", &remote_endpoint)?;
 
                 // Launch a background task to clean up the audit logs
                 launch_pgaudit_gc(log_directory_path);
@@ -916,32 +897,28 @@ impl ComputeNode {
         let mut client = config.connect(NoTls)?;
         let pageserver_connect_micros = start_time.elapsed().as_micros() as u64;
 
-        let basebackup_cmd = match lsn {
-            Lsn(0) => {
-                if spec.spec.mode != ComputeMode::Primary {
-                    format!(
-                        "basebackup {} {} --gzip --replica",
-                        spec.tenant_id, spec.timeline_id
-                    )
-                } else {
-                    format!("basebackup {} {} --gzip", spec.tenant_id, spec.timeline_id)
-                }
-            }
-            _ => {
-                if spec.spec.mode != ComputeMode::Primary {
-                    format!(
-                        "basebackup {} {} {} --gzip --replica",
-                        spec.tenant_id, spec.timeline_id, lsn
-                    )
-                } else {
-                    format!(
-                        "basebackup {} {} {} --gzip",
-                        spec.tenant_id, spec.timeline_id, lsn
-                    )
-                }
-            }
-        };
-
+        let tenant_id = spec.tenant_id.to_string();
+        let timeline_id = spec.timeline_id.to_string();
+        let lsn_str = lsn.to_string();
+        let mut cmd = Vec::new();
+        cmd.push("basebackup");
+        cmd.push(&tenant_id);
+        cmd.push(&timeline_id);
+        if lsn != Lsn::INVALID {
+            cmd.push(&lsn_str);
+        }
+        cmd.push("--gzip");
+        if spec.spec.mode != ComputeMode::Primary {
+            cmd.push("--replica");
+        }
+        if spec
+            .spec
+            .features
+            .contains(&ComputeFeature::LazySlruDownload)
+        {
+            cmd.push("--lazy-slru-download")
+        }
+        let basebackup_cmd = cmd.join(" ");
         let copyreader = client.copy_out(basebackup_cmd.as_str())?;
         let mut measured_reader = MeasuredReader::new(copyreader);
         let mut bufreader = std::io::BufReader::new(&mut measured_reader);

compute_tools/src/http/extract/mod.rs

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

compute_tools/src/http/middleware/authorize.rs

@@ -1,7 +1,7 @@
-use std::collections::HashSet;
+use std::{collections::HashSet, net::SocketAddr};
 
 use anyhow::{Result, anyhow};
-use axum::{RequestExt, body::Body};
+use axum::{RequestExt, body::Body, extract::ConnectInfo};
 use axum_extra::{
     TypedHeader,
     headers::{Authorization, authorization::Bearer},
@@ -13,7 +13,7 @@ use jsonwebtoken::{Algorithm, DecodingKey, TokenData, Validation, jwk::JwkSet};
 use tower_http::auth::AsyncAuthorizeRequest;
 use tracing::{debug, warn};
 
-use crate::http::JsonResponse;
+use crate::http::{JsonResponse, extract::RequestId};
 
 #[derive(Clone, Debug)]
 pub(in crate::http) struct Authorize {
@@ -52,6 +52,31 @@ impl AsyncAuthorizeRequest<Body> for Authorize {
         let validation = self.validation.clone();
 
         Box::pin(async move {
+            let request_id = request.extract_parts::<RequestId>().await.unwrap();
+
+            // TODO(tristan957): Remove this stanza after teaching neon_local
+            // and the regression tests to use a JWT + JWKS.
+            //
+            // https://github.com/neondatabase/neon/issues/11316
+            if cfg!(feature = "testing") {
+                warn!(%request_id, "Skipping compute_ctl authorization check");
+
+                return Ok(request);
+            }
+
+            let connect_info = request
+                .extract_parts::<ConnectInfo<SocketAddr>>()
+                .await
+                .unwrap();
+
+            // In the event the request is coming from the loopback interface,
+            // allow all requests
+            if connect_info.ip().is_loopback() {
+                warn!(%request_id, "Bypassed authorization because request is coming from the loopback interface");
+
+                return Ok(request);
+            }
+
             let TypedHeader(Authorization(bearer)) = request
                 .extract_parts::<TypedHeader<Authorization<Bearer>>>()
                 .await

compute_tools/src/rsyslog.rs

@@ -50,13 +50,13 @@ fn restart_rsyslog() -> Result<()> {
 
 pub fn configure_audit_rsyslog(
     log_directory: String,
-    tag: Option<String>,
+    tag: &str,
     remote_endpoint: &str,
 ) -> Result<()> {
     let config_content: String = format!(
         include_str!("config_template/compute_audit_rsyslog_template.conf"),
         log_directory = log_directory,
-        tag = tag.unwrap_or("".to_string()),
+        tag = tag,
         remote_endpoint = remote_endpoint
     );
 

control_plane/Cargo.toml

@@ -16,6 +16,7 @@ jsonwebtoken.workspace = true
 nix.workspace = true
 once_cell.workspace = true
 pem.workspace = true
+pkcs8.workspace = true
 humantime-serde.workspace = true
 hyper0.workspace = true
 regex.workspace = true
@@ -24,7 +25,6 @@ scopeguard.workspace = true
 serde.workspace = true
 serde_json.workspace = true
 sha2.workspace = true
-spki.workspace = true
 thiserror.workspace = true
 toml.workspace = true
 toml_edit.workspace = true

control_plane/src/bin/neon_local.rs

@@ -63,7 +63,7 @@ const DEFAULT_PAGESERVER_ID: NodeId = NodeId(1);
 const DEFAULT_BRANCH_NAME: &str = "main";
 project_git_version!(GIT_VERSION);
 
-const DEFAULT_PG_VERSION: u32 = 17;
+const DEFAULT_PG_VERSION: u32 = 16;
 
 const DEFAULT_PAGESERVER_CONTROL_PLANE_API: &str = "http://127.0.0.1:1234/upcall/v1/";
 
@@ -1544,7 +1544,7 @@ async fn handle_endpoint(subcmd: &EndpointCmd, env: &local_env::LocalEnv) -> Res
                 .with_context(|| format!("postgres endpoint {endpoint_id} is not found"))?;
             let jwt = endpoint.generate_jwt()?;
 
-            print!("{jwt}");
+            println!("{jwt}");
         }
     }
 

control_plane/src/endpoint.rs

@@ -60,12 +60,11 @@ use jsonwebtoken::jwk::{
 use nix::sys::signal::{Signal, kill};
 use pageserver_api::shard::ShardStripeSize;
 use pem::Pem;
+use pkcs8::der::Decode;
 use reqwest::header::CONTENT_TYPE;
 use safekeeper_api::membership::SafekeeperGeneration;
 use serde::{Deserialize, Serialize};
 use sha2::{Digest, Sha256};
-use spki::der::Decode;
-use spki::{SubjectPublicKeyInfo, SubjectPublicKeyInfoRef};
 use tracing::debug;
 use url::Host;
 use utils::id::{NodeId, TenantId, TimelineId};
@@ -148,12 +147,11 @@ impl ComputeControlPlane {
 
     /// Create a JSON Web Key Set. This ideally matches the way we create a JWKS
     /// from the production control plane.
-    fn create_jwks_from_pem(pem: &Pem) -> Result<JwkSet> {
-        let spki: SubjectPublicKeyInfoRef = SubjectPublicKeyInfo::from_der(pem.contents())?;
-        let public_key = spki.subject_public_key.raw_bytes();
+    fn create_jwks_from_pem(pem: Pem) -> Result<JwkSet> {
+        let document = pkcs8::Document::from_der(&pem.into_contents())?;
 
         let mut hasher = Sha256::new();
-        hasher.update(public_key);
+        hasher.update(&document);
         let key_hash = hasher.finalize();
 
         Ok(JwkSet {
@@ -171,7 +169,7 @@ impl ComputeControlPlane {
                 algorithm: AlgorithmParameters::OctetKeyPair(OctetKeyPairParameters {
                     key_type: OctetKeyPairType::OctetKeyPair,
                     curve: EllipticCurve::Ed25519,
-                    x: base64::encode_config(public_key, base64::URL_SAFE_NO_PAD),
+                    x: base64::encode_config(&document, base64::URL_SAFE_NO_PAD),
                 }),
             }],
         })
@@ -195,7 +193,7 @@ impl ComputeControlPlane {
         let external_http_port = external_http_port.unwrap_or_else(|| self.get_port() + 1);
         let internal_http_port = internal_http_port.unwrap_or_else(|| external_http_port + 1);
         let compute_ctl_config = ComputeCtlConfig {
-            jwks: Self::create_jwks_from_pem(&self.env.read_public_key()?)?,
+            jwks: Self::create_jwks_from_pem(self.env.read_public_key()?)?,
             tls: None::<TlsConfig>,
         };
         let ep = Arc::new(Endpoint {
@@ -766,6 +764,10 @@ impl Endpoint {
             }
         };
 
+        // TODO(tristan957): Remove the write to spec.json after compatibility
+        // tests work themselves out
+        let spec_path = self.endpoint_path().join("spec.json");
+        std::fs::write(spec_path, serde_json::to_string_pretty(&config.spec)?)?;
         let config_path = self.endpoint_path().join("config.json");
         std::fs::write(config_path, serde_json::to_string_pretty(&config)?)?;
 
@@ -775,6 +777,16 @@ impl Endpoint {
             .append(true)
             .open(self.endpoint_path().join("compute.log"))?;
 
+        // TODO(tristan957): Remove when compatibility tests are no longer an
+        // issue
+        let old_compute_ctl = {
+            let mut cmd = Command::new(self.env.neon_distrib_dir.join("compute_ctl"));
+            let help_output = cmd.arg("--help").output()?;
+            let help_output = String::from_utf8_lossy(&help_output.stdout);
+
+            !help_output.contains("--config")
+        };
+
         // Launch compute_ctl
         let conn_str = self.connstr("cloud_admin", "postgres");
         println!("Starting postgres node at '{}'", conn_str);
@@ -793,8 +805,19 @@ impl Endpoint {
         ])
         .args(["--pgdata", self.pgdata().to_str().unwrap()])
         .args(["--connstr", &conn_str])
-        .arg("--config")
-        .arg(self.endpoint_path().join("config.json").as_os_str())
+        // TODO(tristan957): Change this to --config when compatibility tests
+        // are no longer an issue
+        .args([
+            "--spec-path",
+            self.endpoint_path()
+                .join(if old_compute_ctl {
+                    "spec.json"
+                } else {
+                    "config.json"
+                })
+                .to_str()
+                .unwrap(),
+        ])
         .args([
             "--pgbin",
             self.env

control_plane/src/local_env.rs

@@ -23,7 +23,7 @@ use crate::object_storage::{OBJECT_STORAGE_REMOTE_STORAGE_DIR, ObjectStorage};
 use crate::pageserver::{PAGESERVER_REMOTE_STORAGE_DIR, PageServerNode};
 use crate::safekeeper::SafekeeperNode;
 
-pub const DEFAULT_PG_VERSION: u32 = 17;
+pub const DEFAULT_PG_VERSION: u32 = 16;
 
 //
 // This data structures represents neon_local CLI config

control_plane/src/pageserver.rs

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

docker-compose/README.md

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

docker-compose/compute_wrapper/private-key.pem

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

docker-compose/compute_wrapper/public-key.pem

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

docker-compose/compute_wrapper/shell/compute.sh

@@ -81,9 +81,19 @@ sed -i "s/TIMELINE_ID/${timeline_id}/" ${CONFIG_FILE}
 
 cat ${CONFIG_FILE}
 
+# TODO(tristan957): Remove these workarounds for backwards compatibility after
+# the next compute release. That includes these next few lines and the
+# --spec-path in the compute_ctl invocation.
+if compute_ctl --help | grep --quiet -- '--config'; then
+  SPEC_PATH="$CONFIG_FILE"
+else
+  jq '.spec' < "$CONFIG_FILE" > /tmp/spec.json
+  SPEC_PATH=/tmp/spec.json
+fi
+
 echo "Start compute node"
 /usr/local/bin/compute_ctl --pgdata /var/db/postgres/compute \
      -C "postgresql://cloud_admin@localhost:55433/postgres"  \
      -b /usr/local/bin/postgres                              \
      --compute-id "compute-$RANDOM"                          \
-     --config "$CONFIG_FILE"
+     --spec-path "$SPEC_PATH"

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

@@ -142,19 +142,7 @@
     },
     "compute_ctl_config": {
         "jwks": {
-            "keys": [
-                {
-                    "use": "sig",
-                    "key_ops": [
-                        "verify"
-                    ],
-                    "alg": "EdDSA",
-                    "kid": "ZGIxMzAzOGY0YWQwODk2ODU1MTk1NzMxMDFkYmUyOWU2NzZkOWNjNjMyMGRkZGJjOWY0MjdjYWVmNzE1MjUyOAo=",
-                    "kty": "OKP",
-                    "crv": "Ed25519",
-                    "x": "MGQ4ZDFhOTdmNTM0NmUwMDc3ZmJkN2Q0MWE0ZmI3M2NhNWE3YjFjOTNkM2IyYzRkZTQzOGM3MjBkZTk3N2E5ZAo="
-                }
-            ]
+            "keys": []
         }
     }
 }

libs/compute_api/src/spec.rs

@@ -183,6 +183,9 @@ pub enum ComputeFeature {
     /// track short-lived connections as user activity.
     ActivityMonitorExperimental,
 
+    /// Download all SLRU files on demand
+    LazySlruDownload,
+
     /// This is a special feature flag that is used to represent unknown feature flags.
     /// Basically all unknown to enum flags are represented as this one. See unit test
     /// `parse_unknown_features()` for more details.

libs/pageserver_api/Cargo.toml

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

libs/pageserver_api/src/config.rs

@@ -379,8 +379,6 @@ pub struct TenantConfigToml {
     /// size exceeds `compaction_upper_limit * checkpoint_distance`.
     pub compaction_upper_limit: usize,
     pub compaction_algorithm: crate::models::CompactionAlgorithmSettings,
-    /// If true, enable shard ancestor compaction (enabled by default).
-    pub compaction_shard_ancestor: bool,
     /// If true, compact down L0 across all tenant timelines before doing regular compaction. L0
     /// compaction must be responsive to avoid read amp during heavy ingestion. Defaults to true.
     pub compaction_l0_first: bool,
@@ -679,13 +677,12 @@ pub mod tenant_conf_defaults {
 
     pub const DEFAULT_COMPACTION_PERIOD: &str = "20 s";
     pub const DEFAULT_COMPACTION_THRESHOLD: usize = 10;
-    pub const DEFAULT_COMPACTION_SHARD_ANCESTOR: bool = true;
 
     // This value needs to be tuned to avoid OOM. We have 3/4*CPUs threads for L0 compaction, that's
-    // 3/4*8=6 on most of our pageservers. Compacting 10 layers requires a maximum of
-    // DEFAULT_CHECKPOINT_DISTANCE*10 memory, that's 2560MB. So with this config, we can get a maximum peak
-    // compaction usage of 15360MB.
-    pub const DEFAULT_COMPACTION_UPPER_LIMIT: usize = 10;
+    // 3/4*16=9 on most of our pageservers. Compacting 20 layers requires about 1 GB memory (could
+    // be reduced later by optimizing L0 hole calculation to avoid loading all keys into memory). So
+    // with this config, we can get a maximum peak compaction usage of 9 GB.
+    pub const DEFAULT_COMPACTION_UPPER_LIMIT: usize = 20;
     // Enable L0 compaction pass and semaphore by default. L0 compaction must be responsive to avoid
     // read amp.
     pub const DEFAULT_COMPACTION_L0_FIRST: bool = true;
@@ -702,11 +699,8 @@ pub mod tenant_conf_defaults {
     // Relevant: https://github.com/neondatabase/neon/issues/3394
     pub const DEFAULT_GC_PERIOD: &str = "1 hr";
     pub const DEFAULT_IMAGE_CREATION_THRESHOLD: usize = 3;
-    // Currently, any value other than 0 will trigger image layer creation preemption immediately with L0 backpressure
-    // without looking at the exact number of L0 layers.
-    // It was expected to have the following behavior:
-    // > If there are more than threshold * compaction_threshold (that is 3 * 10 in the default config) L0 layers, image
-    // > layer creation will end immediately. Set to 0 to disable.
+    // If there are more than threshold * compaction_threshold (that is 3 * 10 in the default config) L0 layers, image
+    // layer creation will end immediately. Set to 0 to disable.
     pub const DEFAULT_IMAGE_CREATION_PREEMPT_THRESHOLD: usize = 3;
     pub const DEFAULT_PITR_INTERVAL: &str = "7 days";
     pub const DEFAULT_WALRECEIVER_CONNECT_TIMEOUT: &str = "10 seconds";
@@ -740,7 +734,6 @@ impl Default for TenantConfigToml {
             compaction_algorithm: crate::models::CompactionAlgorithmSettings {
                 kind: DEFAULT_COMPACTION_ALGORITHM,
             },
-            compaction_shard_ancestor: DEFAULT_COMPACTION_SHARD_ANCESTOR,
             compaction_l0_first: DEFAULT_COMPACTION_L0_FIRST,
             compaction_l0_semaphore: DEFAULT_COMPACTION_L0_SEMAPHORE,
             l0_flush_delay_threshold: None,

libs/pageserver_api/src/models.rs

@@ -526,8 +526,6 @@ pub struct TenantConfigPatch {
     #[serde(skip_serializing_if = "FieldPatch::is_noop")]
     pub compaction_algorithm: FieldPatch<CompactionAlgorithmSettings>,
     #[serde(skip_serializing_if = "FieldPatch::is_noop")]
-    pub compaction_shard_ancestor: FieldPatch<bool>,
-    #[serde(skip_serializing_if = "FieldPatch::is_noop")]
     pub compaction_l0_first: FieldPatch<bool>,
     #[serde(skip_serializing_if = "FieldPatch::is_noop")]
     pub compaction_l0_semaphore: FieldPatch<bool>,
@@ -617,9 +615,6 @@ pub struct TenantConfig {
     #[serde(skip_serializing_if = "Option::is_none")]
     pub compaction_algorithm: Option<CompactionAlgorithmSettings>,
 
-    #[serde(skip_serializing_if = "Option::is_none")]
-    pub compaction_shard_ancestor: Option<bool>,
-
     #[serde(skip_serializing_if = "Option::is_none")]
     pub compaction_l0_first: Option<bool>,
 
@@ -729,7 +724,6 @@ impl TenantConfig {
             mut compaction_threshold,
             mut compaction_upper_limit,
             mut compaction_algorithm,
-            mut compaction_shard_ancestor,
             mut compaction_l0_first,
             mut compaction_l0_semaphore,
             mut l0_flush_delay_threshold,
@@ -778,9 +772,6 @@ impl TenantConfig {
             .compaction_upper_limit
             .apply(&mut compaction_upper_limit);
         patch.compaction_algorithm.apply(&mut compaction_algorithm);
-        patch
-            .compaction_shard_ancestor
-            .apply(&mut compaction_shard_ancestor);
         patch.compaction_l0_first.apply(&mut compaction_l0_first);
         patch
             .compaction_l0_semaphore
@@ -869,7 +860,6 @@ impl TenantConfig {
             compaction_threshold,
             compaction_upper_limit,
             compaction_algorithm,
-            compaction_shard_ancestor,
             compaction_l0_first,
             compaction_l0_semaphore,
             l0_flush_delay_threshold,
@@ -930,9 +920,6 @@ impl TenantConfig {
                 .as_ref()
                 .unwrap_or(&global_conf.compaction_algorithm)
                 .clone(),
-            compaction_shard_ancestor: self
-                .compaction_shard_ancestor
-                .unwrap_or(global_conf.compaction_shard_ancestor),
             compaction_l0_first: self
                 .compaction_l0_first
                 .unwrap_or(global_conf.compaction_l0_first),
@@ -1817,34 +1804,8 @@ pub mod virtual_file {
     }
 
     impl IoMode {
-        pub fn preferred() -> Self {
-            // The default behavior when running Rust unit tests without any further
-            // flags is to use the newest behavior if available on the platform (Direct).
-            // The CI uses the following environment variable to unit tests for all
-            // different modes.
-            // NB: the Python regression & perf tests have their own defaults management
-            // that writes pageserver.toml; they do not use this variable.
-            if cfg!(test) {
-                use once_cell::sync::Lazy;
-                static CACHED: Lazy<IoMode> = Lazy::new(|| {
-                    utils::env::var_serde_json_string(
-                        "NEON_PAGESERVER_UNIT_TEST_VIRTUAL_FILE_IO_MODE",
-                    )
-                    .unwrap_or({
-                        #[cfg(target_os = "linux")]
-                        {
-                            IoMode::Direct
-                        }
-                        #[cfg(not(target_os = "linux"))]
-                        {
-                            IoMode::Buffered
-                        }
-                    })
-                });
-                *CACHED
-            } else {
-                IoMode::Buffered
-            }
+        pub const fn preferred() -> Self {
+            Self::Buffered
         }
     }
 

libs/remote_storage/Cargo.toml

@@ -18,8 +18,7 @@ camino = { workspace = true, features = ["serde1"] }
 humantime-serde.workspace = true
 hyper = { workspace = true, features = ["client"] }
 futures.workspace = true
-reqwest = { workspace = true, features = ["multipart", "stream"] }
-chrono = { version = "0.4", default-features = false, features = ["clock"] }
+reqwest.workspace = true
 serde.workspace = true
 serde_json.workspace = true
 tokio = { workspace = true, features = ["sync", "fs", "io-util"] }
@@ -41,10 +40,6 @@ http-types.workspace = true
 http-body-util.workspace = true
 itertools.workspace = true
 sync_wrapper = { workspace = true, features = ["futures"] }
-gcp_auth = "0.12.3"
-url.workspace = true
-http.workspace = true
-uuid.workspace = true
 
 [dev-dependencies]
 camino-tempfile.workspace = true

libs/remote_storage/src/config.rs

@@ -41,7 +41,6 @@ impl RemoteStorageKind {
             RemoteStorageKind::LocalFs { .. } => None,
             RemoteStorageKind::AwsS3(config) => Some(&config.bucket_name),
             RemoteStorageKind::AzureContainer(config) => Some(&config.container_name),
-            RemoteStorageKind::GCS(config) => Some(&config.bucket_name),
         }
     }
 }
@@ -52,7 +51,6 @@ impl RemoteStorageConfig {
         match &self.storage {
             RemoteStorageKind::LocalFs { .. } => DEFAULT_REMOTE_STORAGE_LOCALFS_CONCURRENCY_LIMIT,
             RemoteStorageKind::AwsS3(c) => c.concurrency_limit.into(),
-            RemoteStorageKind::GCS(c) => c.concurrency_limit.into(),
             RemoteStorageKind::AzureContainer(c) => c.concurrency_limit.into(),
         }
     }
@@ -87,9 +85,6 @@ pub enum RemoteStorageKind {
     /// Azure Blob based storage, storing all files in the container
     /// specified by the config
     AzureContainer(AzureConfig),
-    /// Google Cloud based storage, storing all files in the GCS bucket
-    /// specified by the config
-    GCS(GCSConfig),
 }
 
 /// AWS S3 bucket coordinates and access credentials to manage the bucket contents (read and write).
@@ -159,32 +154,6 @@ impl Debug for S3Config {
     }
 }
 
-#[derive(Clone, PartialEq, Eq, Deserialize, Serialize)]
-pub struct GCSConfig {
-    /// Name of the bucket to connect to.
-    pub bucket_name: String,
-    /// A "subfolder" in the bucket, to use the same bucket separately by multiple remote storage users at once.
-    pub prefix_in_bucket: Option<String>,
-    #[serde(default = "default_remote_storage_s3_concurrency_limit")]
-    pub concurrency_limit: NonZeroUsize,
-    #[serde(default = "default_max_keys_per_list_response")]
-    pub max_keys_per_list_response: Option<i32>,
-}
-
-impl Debug for GCSConfig {
-    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
-        f.debug_struct("GCSConfig")
-            .field("bucket_name", &self.bucket_name)
-            .field("prefix_in_bucket", &self.prefix_in_bucket)
-            .field("concurrency_limit", &self.concurrency_limit)
-            .field(
-                "max_keys_per_list_response",
-                &self.max_keys_per_list_response,
-            )
-            .finish()
-    }
-}
-
 /// Azure  bucket coordinates and access credentials to manage the bucket contents (read and write).
 #[derive(Clone, PartialEq, Eq, Serialize, Deserialize)]
 pub struct AzureConfig {
@@ -299,30 +268,6 @@ timeout = '5s'";
         );
     }
 
-    #[test]
-    fn test_gcs_parsing() {
-        let toml = "\
-    bucket_name = 'foo-bar'
-    prefix_in_bucket = '/pageserver'
-    ";
-
-        let config = parse(toml).unwrap();
-
-        assert_eq!(
-            config,
-            RemoteStorageConfig {
-                storage: RemoteStorageKind::GCS(GCSConfig {
-                    bucket_name: "foo-bar".into(),
-                    prefix_in_bucket: Some("pageserver/".into()),
-                    max_keys_per_list_response: DEFAULT_MAX_KEYS_PER_LIST_RESPONSE,
-                    concurrency_limit: std::num::NonZero::new(100).unwrap(),
-                }),
-                timeout: Duration::from_secs(120),
-                small_timeout: RemoteStorageConfig::DEFAULT_SMALL_TIMEOUT
-            }
-        );
-    }
-
     #[test]
     fn test_s3_parsing() {
         let toml = "\

libs/remote_storage/src/gcs_bucket.rs

@@ -1,978 +0,0 @@
-#![allow(dead_code)]
-#![allow(unused)]
-
-use crate::config::GCSConfig;
-use crate::error::Cancelled;
-pub(super) use crate::metrics::RequestKind;
-use crate::metrics::{AttemptOutcome, start_counting_cancelled_wait, start_measuring_requests};
-use crate::{
-    ConcurrencyLimiter, Download, DownloadError, DownloadOpts, GCS_SCOPES, Listing, ListingMode,
-    ListingObject, MAX_KEYS_PER_DELETE_GCS, REMOTE_STORAGE_PREFIX_SEPARATOR, RemotePath,
-    RemoteStorage, StorageMetadata, TimeTravelError, TimeoutOrCancel,
-};
-use anyhow::Context;
-use azure_core::Etag;
-use bytes::Bytes;
-use bytes::BytesMut;
-use chrono::DateTime;
-use futures::stream::Stream;
-use futures::stream::TryStreamExt;
-use futures_util::StreamExt;
-use gcp_auth::{Token, TokenProvider};
-use http::Method;
-use http::StatusCode;
-use reqwest::{Client, header};
-use scopeguard::ScopeGuard;
-use serde::{Deserialize, Serialize};
-use std::collections::HashMap;
-use std::fmt::Debug;
-use std::num::NonZeroU32;
-use std::pin::{Pin, pin};
-use std::sync::Arc;
-use std::time::Duration;
-use std::time::SystemTime;
-use tokio_util::codec::{BytesCodec, FramedRead};
-use tokio_util::sync::CancellationToken;
-use tracing;
-use url::Url;
-use uuid::Uuid;
-
-// ---------
-pub struct GCSBucket {
-    token_provider: Arc<dyn TokenProvider>,
-    bucket_name: String,
-    prefix_in_bucket: Option<String>,
-    max_keys_per_list_response: Option<i32>,
-    concurrency_limiter: ConcurrencyLimiter,
-    pub timeout: Duration,
-}
-
-struct GetObjectRequest {
-    bucket: String,
-    key: String,
-    etag: Option<String>,
-    range: Option<String>,
-}
-
-// ---------
-
-impl GCSBucket {
-    pub async fn new(remote_storage_config: &GCSConfig, timeout: Duration) -> anyhow::Result<Self> {
-        tracing::debug!(
-            "creating remote storage for gcs bucket {}",
-            remote_storage_config.bucket_name
-        );
-
-        // clean up 'prefix_in_bucket' if user provides '/pageserver' or 'pageserver/'
-        let prefix_in_bucket = remote_storage_config
-            .prefix_in_bucket
-            .as_deref()
-            .map(|prefix| {
-                let mut prefix = prefix;
-                while prefix.starts_with(REMOTE_STORAGE_PREFIX_SEPARATOR) {
-                    prefix = &prefix[1..];
-                }
-
-                let mut prefix = prefix.to_string();
-                if prefix.ends_with(REMOTE_STORAGE_PREFIX_SEPARATOR) {
-                    prefix.pop();
-                }
-
-                prefix
-            });
-
-        // get GOOGLE_APPLICATION_CREDENTIALS
-        let provider = gcp_auth::provider().await?;
-
-        Ok(GCSBucket {
-            token_provider: Arc::clone(&provider),
-            bucket_name: remote_storage_config.bucket_name.clone(),
-            prefix_in_bucket,
-            timeout,
-            max_keys_per_list_response: remote_storage_config.max_keys_per_list_response,
-            concurrency_limiter: ConcurrencyLimiter::new(
-                remote_storage_config.concurrency_limit.get(),
-            ),
-        })
-    }
-
-    // convert `RemotePath` -> `String`
-    pub fn relative_path_to_gcs_object(&self, path: &RemotePath) -> String {
-        let path_string = path.get_path().as_str();
-        match &self.prefix_in_bucket {
-            Some(prefix) => prefix.clone() + "/" + path_string,
-            None => path_string.to_string(),
-        }
-    }
-
-    // convert `String` -> `RemotePath`
-    pub fn gcs_object_to_relative_path(&self, key: &str) -> RemotePath {
-        let relative_path =
-            match key.strip_prefix(self.prefix_in_bucket.as_deref().unwrap_or_default()) {
-                Some(stripped) => stripped,
-                // we rely on GCS to return properly prefixed paths
-                // for requests with a certain prefix
-                None => panic!(
-                    "Key {} does not start with bucket prefix {:?}",
-                    key, self.prefix_in_bucket
-                ),
-            };
-        RemotePath(
-            relative_path
-                .split(REMOTE_STORAGE_PREFIX_SEPARATOR)
-                .collect(),
-        )
-    }
-
-    pub fn bucket_name(&self) -> &str {
-        &self.bucket_name
-    }
-
-    fn max_keys_per_delete(&self) -> usize {
-        MAX_KEYS_PER_DELETE_GCS
-    }
-
-    async fn permit(
-        &self,
-        kind: RequestKind,
-        cancel: &CancellationToken,
-    ) -> Result<tokio::sync::SemaphorePermit<'_>, Cancelled> {
-        let started_at = start_counting_cancelled_wait(kind);
-        let acquire = self.concurrency_limiter.acquire(kind);
-
-        let permit = tokio::select! {
-            permit = acquire => permit.expect("semaphore is never closed"),
-            _ = cancel.cancelled() => return Err(Cancelled),
-        };
-
-        let started_at = ScopeGuard::into_inner(started_at);
-        crate::metrics::BUCKET_METRICS
-            .wait_seconds
-            .observe_elapsed(kind, started_at);
-
-        Ok(permit)
-    }
-
-    async fn owned_permit(
-        &self,
-        kind: RequestKind,
-        cancel: &CancellationToken,
-    ) -> Result<tokio::sync::OwnedSemaphorePermit, Cancelled> {
-        let started_at = start_counting_cancelled_wait(kind);
-        let acquire = self.concurrency_limiter.acquire_owned(kind);
-
-        let permit = tokio::select! {
-            permit = acquire => permit.expect("semaphore is never closed"),
-            _ = cancel.cancelled() => return Err(Cancelled),
-        };
-
-        let started_at = ScopeGuard::into_inner(started_at);
-        crate::metrics::BUCKET_METRICS
-            .wait_seconds
-            .observe_elapsed(kind, started_at);
-        Ok(permit)
-    }
-
-    async fn put_object(
-        &self,
-        byte_stream: impl Stream<Item = std::io::Result<Bytes>> + Send + Sync + 'static,
-        fs_size: usize,
-        to: &RemotePath,
-        cancel: &CancellationToken,
-    ) -> anyhow::Result<()> {
-        // https://cloud.google.com/storage/docs/xml-api/reference-headers#chunked
-        let mut headers = header::HeaderMap::new();
-        headers.insert(
-            header::TRANSFER_ENCODING,
-            header::HeaderValue::from_static("chunked"),
-        );
-
-        // TODO Check if we need type 'multipart/related' file to attach metadata like Neon's S3
-        // `.upload()` does.
-        // https://cloud.google.com/storage/docs/uploading-objects#uploading-an-object
-        let upload_uri = format!(
-            "https://storage.googleapis.com/upload/storage/v1/b/{}/o/?uploadType=media&name={}",
-            self.bucket_name.clone(),
-            self.relative_path_to_gcs_object(to).trim_start_matches("/")
-        );
-
-        let upload = Client::new()
-            .post(upload_uri)
-            .body(reqwest::Body::wrap_stream(byte_stream))
-            .headers(headers)
-            .bearer_auth(self.token_provider.token(GCS_SCOPES).await?.as_str())
-            .send();
-
-        // We await it in a race against the Tokio timeout
-        let upload = tokio::time::timeout(self.timeout, upload);
-        let res = tokio::select! {
-            res = upload => res,
-            _ = cancel.cancelled() => return Err(TimeoutOrCancel::Cancel.into()),
-        };
-
-        match res {
-            Ok(Ok(res)) => {
-                if !res.status().is_success() {
-                    match res.status() {
-                        StatusCode::NOT_FOUND => {
-                            return Err(anyhow::anyhow!("GCS error: not found \n\t {:?}", res));
-                        }
-                        _ => {
-                            return Err(anyhow::anyhow!(
-                                "GCS PUT response contained no response body \n\t {:?}",
-                                res
-                            ));
-                        }
-                    }
-                } else {
-                    Ok(())
-                }
-            }
-            Ok(Err(reqw)) => Err(reqw.into()),
-            Err(_timeout) => Err(TimeoutOrCancel::Timeout.into()),
-        }
-    }
-
-    async fn copy(
-        &self,
-        from: String,
-        to: String,
-        cancel: &CancellationToken,
-    ) -> anyhow::Result<()> {
-        let kind = RequestKind::Copy;
-
-        let _permit = self.permit(kind, cancel).await?;
-
-        let timeout = tokio::time::sleep(self.timeout);
-
-        let started_at = start_measuring_requests(kind);
-
-        let copy_uri = format!(
-            "https://storage.googleapis.com/storage/v1/b/{}/o/{}/copyTo/b/{}/o/{}",
-            self.bucket_name.clone(),
-            &from,
-            self.bucket_name.clone(),
-            &to
-        );
-
-        let op = Client::new()
-            .post(copy_uri)
-            .bearer_auth(self.token_provider.token(GCS_SCOPES).await?.as_str())
-            .send();
-
-        let res = tokio::select! {
-            res = op => res,
-            _ = timeout => return Err(TimeoutOrCancel::Timeout.into()),
-            _ = cancel.cancelled() => return Err(TimeoutOrCancel::Cancel.into()),
-        };
-
-        let started_at = ScopeGuard::into_inner(started_at);
-        crate::metrics::BUCKET_METRICS
-            .req_seconds
-            .observe_elapsed(kind, &res, started_at);
-
-        res?;
-
-        Ok(())
-    }
-
-    async fn delete_oids(
-        &self,
-        delete_objects: &[String],
-        cancel: &CancellationToken,
-        _permit: &tokio::sync::SemaphorePermit<'_>,
-    ) -> anyhow::Result<()> {
-        let kind = RequestKind::Delete;
-        let mut cancel = std::pin::pin!(cancel.cancelled());
-
-        for chunk in delete_objects.chunks(MAX_KEYS_PER_DELETE_GCS) {
-            let started_at = start_measuring_requests(kind);
-
-            // Use this to report keys that didn't delete based on 'content_id'
-            let mut delete_objects_status = HashMap::new();
-
-            let mut form = reqwest::multipart::Form::new();
-            let bulk_uri = "https://storage.googleapis.com/batch/storage/v1";
-
-            for (index, path) in delete_objects.iter().enumerate() {
-                delete_objects_status.insert(index + 1, path.clone());
-
-                let path_to_delete: String =
-                    url::form_urlencoded::byte_serialize(path.trim_start_matches("/").as_bytes())
-                        .collect();
-
-                let delete_req = format!(
-                    "
-                    DELETE /storage/v1/b/{}/o/{} HTTP/1.1\r\n\
-                    Content-Type: application/json\r\n\
-                    accept: application/json\r\n\
-                    content-length: 0\r\n
-                    ",
-                    self.bucket_name.clone(),
-                    path_to_delete
-                )
-                .trim()
-                .to_string();
-
-                let content_id = format!("<{}+{}>", Uuid::new_v4(), index + 1);
-
-                let mut part_headers = header::HeaderMap::new();
-                part_headers.insert(
-                    header::CONTENT_TYPE,
-                    header::HeaderValue::from_static("application/http"),
-                );
-                part_headers.insert(
-                    header::TRANSFER_ENCODING,
-                    header::HeaderValue::from_static("binary"),
-                );
-                part_headers.insert(
-                    header::HeaderName::from_static("content-id"),
-                    header::HeaderValue::from_str(&content_id)?,
-                );
-                let part = reqwest::multipart::Part::text(delete_req).headers(part_headers);
-
-                form = form.part(format!("request-{}", index), part);
-            }
-
-            let mut headers = header::HeaderMap::new();
-            headers.insert(
-                header::CONTENT_TYPE,
-                header::HeaderValue::from_str(&format!(
-                    "multipart/mixed; boundary={}",
-                    form.boundary()
-                ))?,
-            );
-
-            let req = Client::new()
-                .post(bulk_uri)
-                .bearer_auth(self.token_provider.token(GCS_SCOPES).await?.as_str())
-                .multipart(form)
-                .headers(headers)
-                .send();
-
-            let resp = tokio::select! {
-                resp = req => resp,
-                _ = tokio::time::sleep(self.timeout) => return Err(TimeoutOrCancel::Timeout.into()),
-                _ = &mut cancel => return Err(TimeoutOrCancel::Cancel.into()),
-            };
-
-            let started_at = ScopeGuard::into_inner(started_at);
-            crate::metrics::BUCKET_METRICS
-                .req_seconds
-                .observe_elapsed(kind, &resp, started_at);
-
-            let resp = resp.context("request deletion")?;
-
-            crate::metrics::BUCKET_METRICS
-                .deleted_objects_total
-                .inc_by(chunk.len() as u64);
-
-            let res_headers = resp.headers().to_owned();
-
-            let boundary = res_headers
-                .get(header::CONTENT_TYPE)
-                .unwrap()
-                .to_str()?
-                .split("=")
-                .last()
-                .unwrap();
-
-            let res_body = resp.text().await?;
-
-            let parsed: HashMap<String, String> = res_body
-                .split(&format!("--{}", boundary))
-                .filter_map(|c| {
-                    let mut lines = c.lines();
-
-                    let id = lines.find_map(|line| {
-                        line.strip_prefix("Content-ID:")
-                            .and_then(|suf| suf.split('+').last())
-                            .and_then(|suf| suf.split('>').next())
-                            .map(|x| x.trim().to_string())
-                    });
-
-                    let status_code = lines.find_map(|line| {
-                        // Not sure if this protocol version shouldn't be so specific
-                        line.strip_prefix("HTTP/1.1")
-                            .and_then(|x| x.split_whitespace().next())
-                            .map(|x| x.trim().to_string())
-                    });
-
-                    id.zip(status_code)
-                })
-                .collect();
-
-            // Gather failures
-            let errors: HashMap<usize, &String> = parsed
-                .iter()
-                .filter_map(|(x, y)| {
-                    if y.chars().next() != Some('2') {
-                        x.parse::<usize>().ok().map(|v| (v, y))
-                    } else {
-                        None
-                    }
-                })
-                .collect();
-
-            if !errors.is_empty() {
-                // Report 10 of them like S3
-                const LOG_UP_TO_N_ERRORS: usize = 10;
-                for (id, code) in errors.iter().take(LOG_UP_TO_N_ERRORS) {
-                    tracing::warn!(
-                        "DeleteObjects key {} failed with code: {}",
-                        delete_objects_status.get(id).unwrap(),
-                        code
-                    );
-                }
-
-                return Err(anyhow::anyhow!(
-                    "Failed to delete {}/{} objects",
-                    errors.len(),
-                    chunk.len(),
-                ));
-            }
-        }
-
-        Ok(())
-    }
-
-    async fn list_objects_v2(&self, list_uri: String) -> anyhow::Result<reqwest::RequestBuilder> {
-        let res = Client::new()
-            .get(list_uri)
-            .bearer_auth(self.token_provider.token(GCS_SCOPES).await?.as_str());
-        Ok(res)
-    }
-
-    // need a 'bucket', a 'key', and a bytes 'range'.
-    async fn get_object(
-        &self,
-        request: GetObjectRequest,
-        cancel: &CancellationToken,
-    ) -> anyhow::Result<Download, DownloadError> {
-        let kind = RequestKind::Get;
-
-        let permit = self.owned_permit(kind, cancel).await?;
-
-        let started_at = start_measuring_requests(kind);
-
-        let encoded_path: String =
-            url::form_urlencoded::byte_serialize(request.key.as_bytes()).collect();
-
-        /// We do this in two parts:
-        /// 1. Serialize the metadata of the first request to get Etag, last modified, etc
-        /// 2. We do not .await the second request pass on the pinned stream to the 'get_object'
-        ///    caller
-        // 1. Serialize Metadata in initial request
-        let metadata_uri_mod = "alt=json";
-        let download_uri = format!(
-            "https://storage.googleapis.com/storage/v1/b/{}/o/{}?{}",
-            self.bucket_name.clone(),
-            encoded_path,
-            metadata_uri_mod
-        );
-
-        let res = Client::new()
-            .get(download_uri)
-            .bearer_auth(
-                self.token_provider
-                    .token(GCS_SCOPES)
-                    .await
-                    .map_err(|e: gcp_auth::Error| DownloadError::Other(e.into()))?
-                    .as_str(),
-            )
-            .send()
-            .await
-            .map_err(|e: reqwest::Error| DownloadError::Other(e.into()))?;
-
-        if !res.status().is_success() {
-            match res.status() {
-                StatusCode::NOT_FOUND => return Err(DownloadError::NotFound),
-                _ => {
-                    return Err(DownloadError::Other(anyhow::anyhow!(
-                        "GCS GET resposne contained no response body"
-                    )));
-                }
-            }
-        };
-
-        let body = res
-            .text()
-            .await
-            .map_err(|e: reqwest::Error| DownloadError::Other(e.into()))?;
-
-        let resp: GCSObject = serde_json::from_str(&body)
-            .map_err(|e: serde_json::Error| DownloadError::Other(e.into()))?;
-
-        // 2. Byte Stream request
-        let mut headers = header::HeaderMap::new();
-        headers.insert(header::RANGE, header::HeaderValue::from_static("bytes=0-"));
-
-        let encoded_path: String =
-            url::form_urlencoded::byte_serialize(request.key.as_bytes()).collect();
-
-        let stream_uri_mod = "alt=media";
-        let stream_uri = format!(
-            "https://storage.googleapis.com/storage/v1/b/{}/o/{}?{}",
-            self.bucket_name.clone(),
-            encoded_path,
-            stream_uri_mod
-        );
-
-        let mut req = Client::new()
-            .get(stream_uri)
-            .headers(headers)
-            .bearer_auth(
-                self.token_provider
-                    .token(GCS_SCOPES)
-                    .await
-                    .map_err(|e: gcp_auth::Error| DownloadError::Other(e.into()))?
-                    .as_str(),
-            )
-            .send();
-
-        let get_object = tokio::select! {
-            res = req => res,
-            _ = tokio::time::sleep(self.timeout) => return Err(DownloadError::Timeout),
-            _ = cancel.cancelled() => return Err(DownloadError::Cancelled),
-        };
-
-        let started_at = ScopeGuard::into_inner(started_at);
-
-        let object_output = match get_object {
-            Ok(object_output) => {
-                if !object_output.status().is_success() {
-                    match object_output.status() {
-                        StatusCode::NOT_FOUND => return Err(DownloadError::NotFound),
-                        _ => {
-                            return Err(DownloadError::Other(anyhow::anyhow!(
-                                "GCS GET resposne contained no response body"
-                            )));
-                        }
-                    }
-                } else {
-                    object_output
-                }
-            }
-            Err(e) => {
-                crate::metrics::BUCKET_METRICS.req_seconds.observe_elapsed(
-                    kind,
-                    AttemptOutcome::Err,
-                    started_at,
-                );
-
-                return Err(DownloadError::Other(
-                    anyhow::Error::new(e).context("download s3 object"),
-                ));
-            }
-        };
-
-        let remaining = self.timeout.saturating_sub(started_at.elapsed());
-
-        let metadata = resp.metadata.map(StorageMetadata);
-
-        let etag = resp
-            .etag
-            .ok_or(DownloadError::Other(anyhow::anyhow!("Missing ETag header")))?
-            .into();
-
-        let last_modified: SystemTime = resp
-            .updated
-            .and_then(|s| DateTime::parse_from_rfc3339(&s).ok())
-            .map(|s| s.into())
-            .unwrap_or(SystemTime::now());
-
-        // But let data stream pass through
-        Ok(Download {
-            download_stream: Box::pin(object_output.bytes_stream().map(|item| {
-                item.map_err(|e: reqwest::Error| std::io::Error::new(std::io::ErrorKind::Other, e))
-            })),
-            etag,
-            last_modified,
-            metadata,
-        })
-    }
-}
-
-impl RemoteStorage for GCSBucket {
-    // ---------------------------------------
-    // Neon wrappers for GCS client functions
-    // ---------------------------------------
-
-    fn list_streaming(
-        &self,
-        prefix: Option<&RemotePath>,
-        mode: ListingMode,
-        max_keys: Option<NonZeroU32>,
-        cancel: &CancellationToken,
-    ) -> impl Stream<Item = Result<Listing, DownloadError>> {
-        let kind = RequestKind::List;
-
-        let mut max_keys = max_keys.map(|mk| mk.get() as i32);
-
-        let list_prefix = prefix
-            .map(|p| self.relative_path_to_gcs_object(p))
-            .or_else(|| {
-                self.prefix_in_bucket.clone().map(|mut s| {
-                    s.push(REMOTE_STORAGE_PREFIX_SEPARATOR);
-                    s
-                })
-            })
-            .unwrap();
-
-        let request_max_keys = self
-            .max_keys_per_list_response
-            .into_iter()
-            .chain(max_keys.into_iter())
-            .min()
-            // https://cloud.google.com/storage/docs/json_api/v1/objects/list?hl=en#parameters
-            // TODO set this to default
-            .unwrap_or(1000);
-
-        // We pass URI in to `list_objects_v2` as we'll modify it with `NextPageToken`, hence
-        // `mut`
-        let mut list_uri = format!(
-            "https://storage.googleapis.com/storage/v1/b/{}/o?prefix={}&maxResults={}",
-            self.bucket_name.clone(),
-            list_prefix,
-            request_max_keys,
-        );
-
-        // on ListingMode:
-        // https://github.com/neondatabase/neon/blob/edc11253b65e12a10843711bd88ad277511396d7/libs/remote_storage/src/lib.rs#L158C1-L164C2
-        if let ListingMode::WithDelimiter = mode {
-            list_uri.push_str(&format!(
-                "&delimiter={}",
-                REMOTE_STORAGE_PREFIX_SEPARATOR.to_string()
-            ));
-        }
-
-        async_stream::stream! {
-
-            let mut continuation_token = None;
-
-            'outer: loop {
-                let started_at = start_measuring_requests(kind);
-
-                let request = self.list_objects_v2(list_uri.clone())
-                    .await
-                    .map_err(DownloadError::Other)?
-                    .send();
-
-                // this is like `await`
-                let response = tokio::select! {
-                    res = request => Ok(res),
-                    _ = tokio::time::sleep(self.timeout) => Err(DownloadError::Timeout),
-                    _ = cancel.cancelled() => Err(DownloadError::Cancelled),
-                }?;
-
-                // just mapping our `Result' error variant's type.
-                let response = response
-                    .context("Failed to list GCS prefixes")
-                    .map_err(DownloadError::Other);
-
-                let started_at = ScopeGuard::into_inner(started_at);
-
-                crate::metrics::BUCKET_METRICS
-                    .req_seconds
-                    .observe_elapsed(kind, &response, started_at);
-
-                let response = match response {
-                    Ok(response) => response,
-                    Err(e) => {
-                        // The error is potentially retryable, so we must rewind the loop after yielding.
-                        yield Err(e);
-                        continue 'outer;
-                    },
-                };
-
-                let body = response.text()
-                    .await
-                    .map_err(|e: reqwest::Error| DownloadError::Other(e.into()))?;
-
-                let resp: GCSListResponse = serde_json::from_str(&body).map_err(|e: serde_json::Error| DownloadError::Other(e.into()))?;
-
-                let prefixes = resp.common_prefixes();
-                let keys = resp.contents();
-
-                tracing::debug!("list: {} prefixes, {} keys", prefixes.len(), keys.len());
-
-                let mut result = Listing::default();
-
-                for res in keys.iter() {
-
-                   let last_modified: SystemTime = res.updated.clone()
-                       .and_then(|s| DateTime::parse_from_rfc3339(&s).ok())
-                       .map(|s| s.into())
-                       .unwrap_or(SystemTime::now());
-
-                   let size = res.size.clone().unwrap_or("0".to_string()).parse::<u64>().unwrap();
-
-                   let key = res.name.clone();
-
-                   result.keys.push(
-                        ListingObject{
-                            key: self.gcs_object_to_relative_path(&key),
-                            last_modified,
-                            size,
-                        }
-                   );
-
-                   if let Some(mut mk) = max_keys {
-                       assert!(mk > 0);
-                       mk -= 1;
-                       if mk == 0 {
-                          tracing::debug!("reached limit set by max_keys");
-                          yield Ok(result);
-                          break 'outer;
-                       }
-                       max_keys = Some(mk);
-                   };
-                }
-
-                result.prefixes.extend(prefixes.iter().filter_map(|p| {
-                    Some(
-                        self.gcs_object_to_relative_path(
-                            p.trim_end_matches(REMOTE_STORAGE_PREFIX_SEPARATOR)
-                        ),
-                    )
-                }));
-
-                yield Ok(result);
-
-                continuation_token = match resp.next_page_token {
-                    Some(token) => {
-                        list_uri = list_uri + "&pageToken=" + &token;
-                        Some(token)
-                    },
-                    None => break
-                }
-            }
-        }
-    }
-
-    async fn head_object(
-        &self,
-        key: &RemotePath,
-        cancel: &CancellationToken,
-    ) -> Result<ListingObject, DownloadError> {
-        let kind = RequestKind::Head;
-
-        todo!();
-    }
-
-    async fn upload(
-        &self,
-        from: impl Stream<Item = std::io::Result<Bytes>> + Send + Sync + 'static,
-        from_size_bytes: usize,
-        to: &RemotePath,
-        metadata: Option<StorageMetadata>,
-        cancel: &CancellationToken,
-    ) -> anyhow::Result<()> {
-        let kind = RequestKind::Put;
-        let _permit = self.permit(kind, cancel).await?;
-
-        let started_at = start_measuring_requests(kind);
-
-        let upload = self.put_object(from, from_size_bytes, to, cancel);
-
-        let upload = tokio::time::timeout(self.timeout, upload);
-
-        let res = tokio::select! {
-            res = upload => res,
-            _ = cancel.cancelled() => return Err(TimeoutOrCancel::Cancel.into()),
-        };
-
-        if let Ok(inner) = &res {
-            // do not incl. timeouts as errors in metrics but cancellations
-            let started_at = ScopeGuard::into_inner(started_at);
-            crate::metrics::BUCKET_METRICS
-                .req_seconds
-                .observe_elapsed(kind, inner, started_at);
-        }
-
-        match res {
-            Ok(Ok(_put)) => Ok(()),
-            Ok(Err(sdk)) => Err(sdk.into()),
-            Err(_timeout) => Err(TimeoutOrCancel::Timeout.into()),
-        }
-    }
-
-    async fn copy(
-        &self,
-        from: &RemotePath,
-        to: &RemotePath,
-        cancel: &CancellationToken,
-    ) -> anyhow::Result<()> {
-        let kind = RequestKind::Copy;
-        let _permit = self.permit(kind, cancel).await?;
-
-        let timeout = tokio::time::sleep(self.timeout);
-
-        let started_at = start_measuring_requests(kind);
-
-        // we need to specify bucket_name as a prefix
-        let copy_source = format!(
-            "{}/{}",
-            self.bucket_name,
-            self.relative_path_to_gcs_object(from)
-        );
-
-        todo!();
-    }
-
-    async fn download(
-        &self,
-        from: &RemotePath,
-        opts: &DownloadOpts,
-        cancel: &CancellationToken,
-    ) -> Result<Download, DownloadError> {
-        // if prefix is not none then download file `prefix/from`
-        // if prefix is none then download file `from`
-
-        self.get_object(
-            GetObjectRequest {
-                bucket: self.bucket_name.clone(),
-                key: self
-                    .relative_path_to_gcs_object(from)
-                    .trim_start_matches("/")
-                    .to_string(),
-                etag: opts.etag.as_ref().map(|e| e.to_string()),
-                range: opts.byte_range_header(),
-            },
-            cancel,
-        )
-        .await
-    }
-
-    async fn delete_objects(
-        &self,
-        paths: &[RemotePath],
-        cancel: &CancellationToken,
-    ) -> anyhow::Result<()> {
-        let kind = RequestKind::Delete;
-        let permit = self.permit(kind, cancel).await?;
-
-        let mut delete_objects: Vec<String> = Vec::with_capacity(paths.len());
-
-        let delete_objects: Vec<String> = paths
-            .iter()
-            .map(|i| self.relative_path_to_gcs_object(i))
-            .collect();
-
-        self.delete_oids(&delete_objects, cancel, &permit).await
-    }
-
-    fn max_keys_per_delete(&self) -> usize {
-        MAX_KEYS_PER_DELETE_GCS
-    }
-
-    async fn delete(&self, path: &RemotePath, cancel: &CancellationToken) -> anyhow::Result<()> {
-        let paths = std::array::from_ref(path);
-        self.delete_objects(paths, cancel).await
-    }
-
-    async fn time_travel_recover(
-        &self,
-        prefix: Option<&RemotePath>,
-        timestamp: SystemTime,
-        done_if_after: SystemTime,
-        cancel: &CancellationToken,
-    ) -> Result<(), TimeTravelError> {
-        Ok(())
-    }
-}
-
-// ---------
-
-#[derive(Serialize, Deserialize, Debug)]
-#[serde(rename_all = "snake_case")]
-pub struct GCSListResponse {
-    #[serde(rename = "nextPageToken")]
-    pub next_page_token: Option<String>,
-    pub items: Option<Vec<GCSObject>>,
-    pub prefixes: Option<Vec<String>>,
-}
-
-#[derive(Serialize, Deserialize, Debug)]
-#[serde(rename_all = "snake_case")]
-pub struct GCSObject {
-    pub name: String,
-    pub bucket: String,
-    pub generation: String,
-    pub metageneration: String,
-    #[serde(rename = "contentType")]
-    pub content_type: Option<String>,
-    #[serde(rename = "storageClass")]
-    pub storage_class: String,
-    pub size: Option<String>,
-    #[serde(rename = "md5Hash")]
-    pub md5_hash: Option<String>,
-    pub crc32c: String,
-    pub etag: Option<String>,
-    #[serde(rename = "timeCreated")]
-    pub time_created: String,
-    pub updated: Option<String>,
-    #[serde(rename = "timeStorageClassUpdated")]
-    pub time_storage_class_updated: String,
-    #[serde(rename = "timeFinalized")]
-    pub time_finalized: String,
-    pub metadata: Option<HashMap<String, String>>,
-}
-
-impl GCSListResponse {
-    pub fn contents(&self) -> &[GCSObject] {
-        self.items.as_deref().unwrap_or_default()
-    }
-    pub fn common_prefixes(&self) -> &[String] {
-        self.prefixes.as_deref().unwrap_or_default()
-    }
-}
-
-#[cfg(test)]
-mod tests {
-
-    use super::*;
-    use gcp_auth;
-    use std::num::NonZero;
-    use std::pin::pin;
-    use std::sync::Arc;
-
-    const BUFFER_SIZE: usize = 32 * 1024;
-
-    // TODO what does Neon want here for integration tests?
-    const BUCKET: &str = "https://storage.googleapis.com/storage/v1/b/my-test-bucket";
-
-    #[tokio::test]
-    async fn list_returns_keys_from_bucket() {
-        let provider = gcp_auth::provider().await.unwrap();
-        let gcs = GCSBucket {
-            token_provider: Arc::clone(&provider),
-            bucket_name: BUCKET.to_string(),
-            prefix_in_bucket: None,
-            max_keys_per_list_response: Some(100),
-            concurrency_limiter: ConcurrencyLimiter::new(100),
-            timeout: std::time::Duration::from_secs(120),
-        };
-
-        let cancel = CancellationToken::new();
-        let remote_prefix = "box/tiff/2023/TN".to_string();
-        let max_keys: u32 = 100;
-        let mut stream = pin!(gcs.list_streaming(Some(remote_prefix), NonZero::new(max_keys)));
-        let mut combined = stream
-            .next()
-            .await
-            .expect("At least one item required")
-            .unwrap();
-        while let Some(list) = stream.next().await {
-            let list = list.unwrap();
-            combined.keys.extend(list.keys.into_iter());
-            combined.prefixes.extend_from_slice(&list.prefixes);
-        }
-
-        for key in combined.keys.iter() {
-            println!("Item: {} -- {:?}", key.key, key.last_modified);
-        }
-
-        assert_ne!(0, combined.keys.len());
-    }
-}

libs/remote_storage/src/lib.rs

@@ -12,7 +12,6 @@
 mod azure_blob;
 mod config;
 mod error;
-mod gcs_bucket;
 mod local_fs;
 mod metrics;
 mod s3_bucket;
@@ -43,7 +42,6 @@ use tokio_util::sync::CancellationToken;
 use tracing::info;
 
 pub use self::azure_blob::AzureBlobStorage;
-pub use self::gcs_bucket::GCSBucket;
 pub use self::local_fs::LocalFs;
 pub use self::s3_bucket::S3Bucket;
 pub use self::simulate_failures::UnreliableWrapper;
@@ -82,12 +80,8 @@ pub const MAX_KEYS_PER_DELETE_S3: usize = 1000;
 /// <https://learn.microsoft.com/en-us/rest/api/storageservices/blob-batch>
 pub const MAX_KEYS_PER_DELETE_AZURE: usize = 256;
 
-pub const MAX_KEYS_PER_DELETE_GCS: usize = 1000;
-
 const REMOTE_STORAGE_PREFIX_SEPARATOR: char = '/';
 
-const GCS_SCOPES: &[&str] = &["https://www.googleapis.com/auth/cloud-platform"];
-
 /// Path on the remote storage, relative to some inner prefix.
 /// The prefix is an implementation detail, that allows representing local paths
 /// as the remote ones, stripping the local storage prefix away.
@@ -445,7 +439,6 @@ pub enum GenericRemoteStorage<Other: Clone = Arc<UnreliableWrapper>> {
     AwsS3(Arc<S3Bucket>),
     AzureBlob(Arc<AzureBlobStorage>),
     Unreliable(Other),
-    GCS(Arc<GCSBucket>),
 }
 
 impl<Other: RemoteStorage> GenericRemoteStorage<Arc<Other>> {
@@ -462,7 +455,6 @@ impl<Other: RemoteStorage> GenericRemoteStorage<Arc<Other>> {
             Self::AwsS3(s) => s.list(prefix, mode, max_keys, cancel).await,
             Self::AzureBlob(s) => s.list(prefix, mode, max_keys, cancel).await,
             Self::Unreliable(s) => s.list(prefix, mode, max_keys, cancel).await,
-            Self::GCS(s) => s.list(prefix, mode, max_keys, cancel).await,
         }
     }
 
@@ -480,7 +472,6 @@ impl<Other: RemoteStorage> GenericRemoteStorage<Arc<Other>> {
             Self::AwsS3(s) => Box::pin(s.list_streaming(prefix, mode, max_keys, cancel)),
             Self::AzureBlob(s) => Box::pin(s.list_streaming(prefix, mode, max_keys, cancel)),
             Self::Unreliable(s) => Box::pin(s.list_streaming(prefix, mode, max_keys, cancel)),
-            Self::GCS(s) => Box::pin(s.list_streaming(prefix, mode, max_keys, cancel)),
         }
     }
 
@@ -495,7 +486,6 @@ impl<Other: RemoteStorage> GenericRemoteStorage<Arc<Other>> {
             Self::AwsS3(s) => s.head_object(key, cancel).await,
             Self::AzureBlob(s) => s.head_object(key, cancel).await,
             Self::Unreliable(s) => s.head_object(key, cancel).await,
-            Self::GCS(_) => todo!(),
         }
     }
 
@@ -513,7 +503,6 @@ impl<Other: RemoteStorage> GenericRemoteStorage<Arc<Other>> {
             Self::AwsS3(s) => s.upload(from, data_size_bytes, to, metadata, cancel).await,
             Self::AzureBlob(s) => s.upload(from, data_size_bytes, to, metadata, cancel).await,
             Self::Unreliable(s) => s.upload(from, data_size_bytes, to, metadata, cancel).await,
-            Self::GCS(s) => s.upload(from, data_size_bytes, to, metadata, cancel).await,
         }
     }
 
@@ -529,7 +518,6 @@ impl<Other: RemoteStorage> GenericRemoteStorage<Arc<Other>> {
             Self::AwsS3(s) => s.download(from, opts, cancel).await,
             Self::AzureBlob(s) => s.download(from, opts, cancel).await,
             Self::Unreliable(s) => s.download(from, opts, cancel).await,
-            Self::GCS(s) => s.download(from, opts, cancel).await,
         }
     }
 
@@ -544,7 +532,6 @@ impl<Other: RemoteStorage> GenericRemoteStorage<Arc<Other>> {
             Self::AwsS3(s) => s.delete(path, cancel).await,
             Self::AzureBlob(s) => s.delete(path, cancel).await,
             Self::Unreliable(s) => s.delete(path, cancel).await,
-            Self::GCS(s) => s.delete(path, cancel).await,
         }
     }
 
@@ -559,7 +546,6 @@ impl<Other: RemoteStorage> GenericRemoteStorage<Arc<Other>> {
             Self::AwsS3(s) => s.delete_objects(paths, cancel).await,
             Self::AzureBlob(s) => s.delete_objects(paths, cancel).await,
             Self::Unreliable(s) => s.delete_objects(paths, cancel).await,
-            Self::GCS(s) => s.delete_objects(paths, cancel).await,
         }
     }
 
@@ -570,7 +556,6 @@ impl<Other: RemoteStorage> GenericRemoteStorage<Arc<Other>> {
             Self::AwsS3(s) => s.max_keys_per_delete(),
             Self::AzureBlob(s) => s.max_keys_per_delete(),
             Self::Unreliable(s) => s.max_keys_per_delete(),
-            Self::GCS(s) => s.max_keys_per_delete(),
         }
     }
 
@@ -585,7 +570,6 @@ impl<Other: RemoteStorage> GenericRemoteStorage<Arc<Other>> {
             Self::AwsS3(s) => s.delete_prefix(prefix, cancel).await,
             Self::AzureBlob(s) => s.delete_prefix(prefix, cancel).await,
             Self::Unreliable(s) => s.delete_prefix(prefix, cancel).await,
-            Self::GCS(_) => todo!(),
         }
     }
 
@@ -601,7 +585,6 @@ impl<Other: RemoteStorage> GenericRemoteStorage<Arc<Other>> {
             Self::AwsS3(s) => s.copy(from, to, cancel).await,
             Self::AzureBlob(s) => s.copy(from, to, cancel).await,
             Self::Unreliable(s) => s.copy(from, to, cancel).await,
-            Self::GCS(_) => todo!(),
         }
     }
 
@@ -630,25 +613,17 @@ impl<Other: RemoteStorage> GenericRemoteStorage<Arc<Other>> {
                 s.time_travel_recover(prefix, timestamp, done_if_after, cancel)
                     .await
             }
-            Self::GCS(_) => todo!(),
         }
     }
 }
 
 impl GenericRemoteStorage {
     pub async fn from_config(storage_config: &RemoteStorageConfig) -> anyhow::Result<Self> {
-        info!("RemoteStorageConfig: {:?}", storage_config);
-
         let timeout = storage_config.timeout;
 
-        // If someone overrides timeout to be small without adjusting small_timeout, then adjust it automatically
+        // If somkeone overrides timeout to be small without adjusting small_timeout, then adjust it automatically
         let small_timeout = std::cmp::min(storage_config.small_timeout, timeout);
 
-        info!(
-            "RemoteStorageConfig's storage attribute: {:?}",
-            storage_config.storage
-        );
-
         Ok(match &storage_config.storage {
             RemoteStorageKind::LocalFs { local_path: path } => {
                 info!("Using fs root '{path}' as a remote storage");
@@ -686,16 +661,6 @@ impl GenericRemoteStorage {
                     small_timeout,
                 )?))
             }
-            RemoteStorageKind::GCS(gcs_config) => {
-                let google_application_credentials =
-                    std::env::var("GOOGLE_APPLICATION_CREDENTIALS")
-                        .unwrap_or_else(|_| "<none>".into());
-                info!(
-                    "Using gcs bucket '{}' as a remote storage, prefix in bucket: '{:?}', GOOGLE_APPLICATION_CREDENTIALS: {google_application_credentials }",
-                    gcs_config.bucket_name, gcs_config.prefix_in_bucket
-                );
-                Self::GCS(Arc::new(GCSBucket::new(gcs_config, timeout).await?))
-            }
         })
     }
 
@@ -725,7 +690,6 @@ impl GenericRemoteStorage {
             Self::AwsS3(s) => Some(s.bucket_name()),
             Self::AzureBlob(s) => Some(s.container_name()),
             Self::Unreliable(_s) => None,
-            Self::GCS(s) => Some(s.bucket_name()),
         }
     }
 }

libs/remote_storage/src/simulate_failures.rs

@@ -50,7 +50,6 @@ impl UnreliableWrapper {
             GenericRemoteStorage::Unreliable(_s) => {
                 panic!("Can't wrap unreliable wrapper unreliably")
             }
-            GenericRemoteStorage::GCS(_) => todo!(),
         };
         UnreliableWrapper {
             inner,

pageserver/Cargo.toml

@@ -78,7 +78,6 @@ metrics.workspace = true
 pageserver_api.workspace = true
 pageserver_client.workspace = true # for ResponseErrorMessageExt TOOD refactor that
 pageserver_compaction.workspace = true
-pem.workspace = true
 postgres_connection.workspace = true
 postgres_ffi.workspace = true
 pq_proto.workspace = true

pageserver/pagebench/src/cmd/getpage_latest_lsn.rs

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

pageserver/src/basebackup.rs

@@ -73,6 +73,7 @@ impl From<GetVectoredError> for BasebackupError {
 ///  * When working without safekeepers. In this situation it is important to match the lsn
 ///    we are taking basebackup on with the lsn that is used in pageserver's walreceiver
 ///    to start the replication.
+#[allow(clippy::too_many_arguments)]
 pub async fn send_basebackup_tarball<'a, W>(
     write: &'a mut W,
     timeline: &'a Timeline,
@@ -80,6 +81,7 @@ pub async fn send_basebackup_tarball<'a, W>(
     prev_lsn: Option<Lsn>,
     full_backup: bool,
     replica: bool,
+    lazy_slru_download_enabled: bool,
     ctx: &'a RequestContext,
 ) -> Result<(), BasebackupError>
 where
@@ -131,8 +133,8 @@ where
     };
 
     info!(
-        "taking basebackup lsn={}, prev_lsn={} (full_backup={}, replica={})",
-        backup_lsn, prev_lsn, full_backup, replica
+        "taking basebackup lsn={}, prev_lsn={} (full_backup={}, replica={}, lazy_slru_download_enabled={})",
+        backup_lsn, prev_lsn, full_backup, replica, lazy_slru_download_enabled
     );
 
     let basebackup = Basebackup {
@@ -142,6 +144,7 @@ where
         prev_record_lsn: prev_lsn,
         full_backup,
         replica,
+        lazy_slru_download_enabled,
         ctx,
         io_concurrency: IoConcurrency::spawn_from_conf(
             timeline.conf,
@@ -170,6 +173,7 @@ where
     prev_record_lsn: Lsn,
     full_backup: bool,
     replica: bool,
+    lazy_slru_download_enabled: bool,
     ctx: &'a RequestContext,
     io_concurrency: IoConcurrency,
 }
@@ -308,7 +312,10 @@ where
                 self.timeline.pg_version,
             )?;
 
-        let lazy_slru_download = self.timeline.get_lazy_slru_download() && !self.full_backup;
+        let lazy_slru_download = self
+            .timeline
+            .get_lazy_slru_download(self.lazy_slru_download_enabled)
+            && !self.full_backup;
 
         let pgversion = self.timeline.pg_version;
         let subdirs = dispatch_pgversion!(pgversion, &pgv::bindings::PGDATA_SUBDIRS[..]);

pageserver/src/bin/pageserver.rs

@@ -416,18 +416,8 @@ fn start_pageserver(
     // The storage_broker::connect call needs to happen inside a tokio runtime thread.
     let broker_client = WALRECEIVER_RUNTIME
         .block_on(async {
-            let tls_config = storage_broker::ClientTlsConfig::new().ca_certificates(
-                conf.ssl_ca_certs
-                    .iter()
-                    .map(pem::encode)
-                    .map(storage_broker::Certificate::from_pem),
-            );
             // Note: we do not attempt connecting here (but validate endpoints sanity).
-            storage_broker::connect(
-                conf.broker_endpoint.clone(),
-                conf.broker_keepalive_interval,
-                tls_config,
-            )
+            storage_broker::connect(conf.broker_endpoint.clone(), conf.broker_keepalive_interval)
         })
         .with_context(|| {
             format!(

pageserver/src/config.rs

@@ -17,10 +17,9 @@ use once_cell::sync::OnceCell;
 use pageserver_api::config::{DiskUsageEvictionTaskConfig, MaxVectoredReadBytes};
 use pageserver_api::models::ImageCompressionAlgorithm;
 use pageserver_api::shard::TenantShardId;
-use pem::Pem;
 use postgres_backend::AuthType;
 use remote_storage::{RemotePath, RemoteStorageConfig};
-use reqwest::Url;
+use reqwest::{Certificate, Url};
 use storage_broker::Uri;
 use utils::id::{NodeId, TimelineId};
 use utils::logging::{LogFormat, SecretString};
@@ -68,8 +67,8 @@ pub struct PageServerConf {
     /// Period to reload certificate and private key from files.
     /// Default: 60s.
     pub ssl_cert_reload_period: Duration,
-    /// Trusted root CA certificates to use in https APIs in PEM format.
-    pub ssl_ca_certs: Vec<Pem>,
+    /// Trusted root CA certificates to use in https APIs.
+    pub ssl_ca_certs: Vec<Certificate>,
 
     /// Current availability zone. Used for traffic metrics.
     pub availability_zone: Option<String>,
@@ -119,13 +118,13 @@ pub struct PageServerConf {
     /// A lower value implicitly deprioritizes loading such tenants, vs. other work in the system.
     pub concurrent_tenant_warmup: ConfigurableSemaphore,
 
-    /// Number of concurrent [`TenantShard::gather_size_inputs`](crate::tenant::TenantShard::gather_size_inputs) allowed.
+    /// Number of concurrent [`Tenant::gather_size_inputs`](crate::tenant::Tenant::gather_size_inputs) allowed.
     pub concurrent_tenant_size_logical_size_queries: ConfigurableSemaphore,
-    /// Limit of concurrent [`TenantShard::gather_size_inputs`] issued by module `eviction_task`.
+    /// Limit of concurrent [`Tenant::gather_size_inputs`] issued by module `eviction_task`.
     /// The number of permits is the same as `concurrent_tenant_size_logical_size_queries`.
     /// See the comment in `eviction_task` for details.
     ///
-    /// [`TenantShard::gather_size_inputs`]: crate::tenant::TenantShard::gather_size_inputs
+    /// [`Tenant::gather_size_inputs`]: crate::tenant::Tenant::gather_size_inputs
     pub eviction_task_immitated_concurrent_logical_size_queries: ConfigurableSemaphore,
 
     // How often to collect metrics and send them to the metrics endpoint.
@@ -498,10 +497,7 @@ impl PageServerConf {
             ssl_ca_certs: match ssl_ca_file {
                 Some(ssl_ca_file) => {
                     let buf = std::fs::read(ssl_ca_file)?;
-                    pem::parse_many(&buf)?
-                        .into_iter()
-                        .filter(|pem| pem.tag() == "CERTIFICATE")
-                        .collect()
+                    Certificate::from_pem_bundle(&buf)?
                 }
                 None => Vec::new(),
             },
@@ -592,10 +588,10 @@ impl ConfigurableSemaphore {
     /// Initializse using a non-zero amount of permits.
     ///
     /// Require a non-zero initial permits, because using permits == 0 is a crude way to disable a
-    /// feature such as [`TenantShard::gather_size_inputs`]. Otherwise any semaphore using future will
+    /// feature such as [`Tenant::gather_size_inputs`]. Otherwise any semaphore using future will
     /// behave like [`futures::future::pending`], just waiting until new permits are added.
     ///
-    /// [`TenantShard::gather_size_inputs`]: crate::tenant::TenantShard::gather_size_inputs
+    /// [`Tenant::gather_size_inputs`]: crate::tenant::Tenant::gather_size_inputs
     pub fn new(initial_permits: NonZeroUsize) -> Self {
         ConfigurableSemaphore {
             initial_permits,

pageserver/src/consumption_metrics.rs

@@ -24,7 +24,7 @@ use crate::task_mgr::{self, BACKGROUND_RUNTIME, TaskKind};
 use crate::tenant::mgr::TenantManager;
 use crate::tenant::size::CalculateSyntheticSizeError;
 use crate::tenant::tasks::BackgroundLoopKind;
-use crate::tenant::{LogicalSizeCalculationCause, TenantShard};
+use crate::tenant::{LogicalSizeCalculationCause, Tenant};
 
 mod disk_cache;
 mod metrics;
@@ -428,7 +428,7 @@ async fn calculate_synthetic_size_worker(
     }
 }
 
-async fn calculate_and_log(tenant: &TenantShard, cancel: &CancellationToken, ctx: &RequestContext) {
+async fn calculate_and_log(tenant: &Tenant, cancel: &CancellationToken, ctx: &RequestContext) {
     const CAUSE: LogicalSizeCalculationCause =
         LogicalSizeCalculationCause::ConsumptionMetricsSyntheticSize;
 

pageserver/src/consumption_metrics/metrics.rs

@@ -175,9 +175,9 @@ impl MetricsKey {
         .absolute_values()
     }
 
-    /// [`TenantShard::remote_size`]
+    /// [`Tenant::remote_size`]
     ///
-    /// [`TenantShard::remote_size`]: crate::tenant::TenantShard::remote_size
+    /// [`Tenant::remote_size`]: crate::tenant::Tenant::remote_size
     const fn remote_storage_size(tenant_id: TenantId) -> AbsoluteValueFactory {
         MetricsKey {
             tenant_id,
@@ -199,9 +199,9 @@ impl MetricsKey {
         .absolute_values()
     }
 
-    /// [`TenantShard::cached_synthetic_size`] as refreshed by [`calculate_synthetic_size_worker`].
+    /// [`Tenant::cached_synthetic_size`] as refreshed by [`calculate_synthetic_size_worker`].
     ///
-    /// [`TenantShard::cached_synthetic_size`]: crate::tenant::TenantShard::cached_synthetic_size
+    /// [`Tenant::cached_synthetic_size`]: crate::tenant::Tenant::cached_synthetic_size
     /// [`calculate_synthetic_size_worker`]: super::calculate_synthetic_size_worker
     const fn synthetic_size(tenant_id: TenantId) -> AbsoluteValueFactory {
         MetricsKey {
@@ -254,7 +254,7 @@ pub(super) async fn collect_all_metrics(
 
 async fn collect<S>(tenants: S, cache: &Cache, ctx: &RequestContext) -> Vec<NewRawMetric>
 where
-    S: futures::stream::Stream<Item = (TenantId, Arc<crate::tenant::TenantShard>)>,
+    S: futures::stream::Stream<Item = (TenantId, Arc<crate::tenant::Tenant>)>,
 {
     let mut current_metrics: Vec<NewRawMetric> = Vec::new();
 
@@ -308,7 +308,7 @@ impl TenantSnapshot {
     ///
     /// `resident_size` is calculated of the timelines we had access to for other metrics, so we
     /// cannot just list timelines here.
-    fn collect(t: &Arc<crate::tenant::TenantShard>, resident_size: u64) -> Self {
+    fn collect(t: &Arc<crate::tenant::Tenant>, resident_size: u64) -> Self {
         TenantSnapshot {
             resident_size,
             remote_size: t.remote_size(),

pageserver/src/controller_upcall_client.rs

@@ -8,7 +8,6 @@ use pageserver_api::upcall_api::{
     ReAttachRequest, ReAttachResponse, ReAttachResponseTenant, ValidateRequest,
     ValidateRequestTenant, ValidateResponse,
 };
-use reqwest::Certificate;
 use serde::Serialize;
 use serde::de::DeserializeOwned;
 use tokio_util::sync::CancellationToken;
@@ -77,8 +76,8 @@ impl StorageControllerUpcallClient {
             client = client.default_headers(headers);
         }
 
-        for cert in &conf.ssl_ca_certs {
-            client = client.add_root_certificate(Certificate::from_der(cert.contents())?);
+        for ssl_ca_cert in &conf.ssl_ca_certs {
+            client = client.add_root_certificate(ssl_ca_cert.clone());
         }
 
         Ok(Some(Self {

pageserver/src/http/routes.rs

@@ -1873,7 +1873,7 @@ async fn update_tenant_config_handler(
         &ShardParameters::default(),
     );
 
-    crate::tenant::TenantShard::persist_tenant_config(state.conf, &tenant_shard_id, &location_conf)
+    crate::tenant::Tenant::persist_tenant_config(state.conf, &tenant_shard_id, &location_conf)
         .await
         .map_err(|e| ApiError::InternalServerError(anyhow::anyhow!(e)))?;
 
@@ -1917,7 +1917,7 @@ async fn patch_tenant_config_handler(
         &ShardParameters::default(),
     );
 
-    crate::tenant::TenantShard::persist_tenant_config(state.conf, &tenant_shard_id, &location_conf)
+    crate::tenant::Tenant::persist_tenant_config(state.conf, &tenant_shard_id, &location_conf)
         .await
         .map_err(|e| ApiError::InternalServerError(anyhow::anyhow!(e)))?;
 

pageserver/src/lib.rs

@@ -49,7 +49,7 @@ use tracing::{info, info_span};
 /// backwards-compatible changes to the metadata format.
 pub const STORAGE_FORMAT_VERSION: u16 = 3;
 
-pub const DEFAULT_PG_VERSION: u32 = 17;
+pub const DEFAULT_PG_VERSION: u32 = 16;
 
 // Magic constants used to identify different kinds of files
 pub const IMAGE_FILE_MAGIC: u16 = 0x5A60;

pageserver/src/metrics.rs

@@ -1086,7 +1086,7 @@ pub(crate) static TIMELINE_EPHEMERAL_BYTES: Lazy<UIntGauge> = Lazy::new(|| {
     .expect("Failed to register metric")
 });
 
-/// Metrics related to the lifecycle of a [`crate::tenant::TenantShard`] object: things
+/// Metrics related to the lifecycle of a [`crate::tenant::Tenant`] object: things
 /// like how long it took to load.
 ///
 /// Note that these are process-global metrics, _not_ per-tenant metrics.  Per-tenant

pageserver/src/page_service.rs

@@ -76,7 +76,7 @@ use crate::tenant::timeline::{self, WaitLsnError};
 use crate::tenant::{GetTimelineError, PageReconstructError, Timeline};
 use crate::{basebackup, timed_after_cancellation};
 
-/// How long we may wait for a [`crate::tenant::mgr::TenantSlot::InProgress`]` and/or a [`crate::tenant::TenantShard`] which
+/// 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`].
 ///
 /// NB: this is a different value than [`crate::http::routes::ACTIVE_TENANT_TIMEOUT`].
@@ -2394,6 +2394,7 @@ impl PageServerHandler {
         full_backup: bool,
         gzip: bool,
         replica: bool,
+        lazy_slru_download: bool,
         ctx: &RequestContext,
     ) -> Result<(), QueryError>
     where
@@ -2461,6 +2462,7 @@ impl PageServerHandler {
                 prev_lsn,
                 full_backup,
                 replica,
+                lazy_slru_download,
                 &ctx,
             )
             .await
@@ -2484,6 +2486,7 @@ impl PageServerHandler {
                     prev_lsn,
                     full_backup,
                     replica,
+                    lazy_slru_download,
                     &ctx,
                 )
                 .await
@@ -2501,6 +2504,7 @@ impl PageServerHandler {
                     prev_lsn,
                     full_backup,
                     replica,
+                    lazy_slru_download,
                     &ctx,
                 )
                 .await
@@ -2550,7 +2554,7 @@ impl PageServerHandler {
     }
 }
 
-/// `basebackup tenant timeline [lsn] [--gzip] [--replica]`
+/// `basebackup tenant timeline [lsn] [--gzip] [--replica] [--lazy-slru-download]`
 #[derive(Debug, Clone, Eq, PartialEq)]
 struct BaseBackupCmd {
     tenant_id: TenantId,
@@ -2558,6 +2562,7 @@ struct BaseBackupCmd {
     lsn: Option<Lsn>,
     gzip: bool,
     replica: bool,
+    lazy_slru_download: bool,
 }
 
 /// `fullbackup tenant timeline [lsn] [prev_lsn]`
@@ -2690,6 +2695,7 @@ impl BaseBackupCmd {
 
         let mut gzip = false;
         let mut replica = false;
+        let mut lazy_slru_download = false;
 
         for &param in &parameters[flags_parse_from..] {
             match param {
@@ -2705,6 +2711,12 @@ impl BaseBackupCmd {
                     }
                     replica = true
                 }
+                "--lazy-slru-download" => {
+                    if lazy_slru_download {
+                        bail!("duplicate parameter for basebackup command: {param}")
+                    }
+                    lazy_slru_download = true
+                }
                 _ => bail!("invalid parameter for basebackup command: {param}"),
             }
         }
@@ -2714,6 +2726,7 @@ impl BaseBackupCmd {
             lsn,
             gzip,
             replica,
+            lazy_slru_download,
         })
     }
 }
@@ -2928,6 +2941,7 @@ where
                 lsn,
                 gzip,
                 replica,
+                lazy_slru_download,
             }) => {
                 tracing::Span::current()
                     .record("tenant_id", field::display(tenant_id))
@@ -2949,6 +2963,7 @@ where
                         false,
                         gzip,
                         replica,
+                        lazy_slru_download,
                         &ctx,
                     )
                     .await?;
@@ -2986,6 +3001,7 @@ where
                     true,
                     false,
                     false,
+                    false,
                     &ctx,
                 )
                 .await?;
@@ -3120,7 +3136,8 @@ mod tests {
                 timeline_id,
                 lsn: None,
                 gzip: false,
-                replica: false
+                replica: false,
+                lazy_slru_download: false
             })
         );
         let cmd =
@@ -3132,7 +3149,8 @@ mod tests {
                 timeline_id,
                 lsn: None,
                 gzip: true,
-                replica: false
+                replica: false,
+                lazy_slru_download: false
             })
         );
         let cmd =
@@ -3144,7 +3162,8 @@ mod tests {
                 timeline_id,
                 lsn: None,
                 gzip: false,
-                replica: false
+                replica: false,
+                lazy_slru_download: false
             })
         );
         let cmd = PageServiceCmd::parse(&format!("basebackup {tenant_id} {timeline_id} 0/16ABCDE"))
@@ -3156,7 +3175,8 @@ mod tests {
                 timeline_id,
                 lsn: Some(Lsn::from_str("0/16ABCDE").unwrap()),
                 gzip: false,
-                replica: false
+                replica: false,
+                lazy_slru_download: false
             })
         );
         let cmd = PageServiceCmd::parse(&format!(
@@ -3170,7 +3190,23 @@ mod tests {
                 timeline_id,
                 lsn: None,
                 gzip: true,
-                replica: true
+                replica: true,
+                lazy_slru_download: false
+            })
+        );
+        let cmd = PageServiceCmd::parse(&format!(
+            "basebackup {tenant_id} {timeline_id} --replica --gzip --lazy-slru-download"
+        ))
+        .unwrap();
+        assert_eq!(
+            cmd,
+            PageServiceCmd::BaseBackup(BaseBackupCmd {
+                tenant_id,
+                timeline_id,
+                lsn: None,
+                gzip: true,
+                replica: true,
+                lazy_slru_download: true
             })
         );
         let cmd = PageServiceCmd::parse(&format!(
@@ -3184,7 +3220,8 @@ mod tests {
                 timeline_id,
                 lsn: Some(Lsn::from_str("0/16ABCDE").unwrap()),
                 gzip: true,
-                replica: true
+                replica: true,
+                lazy_slru_download: false
             })
         );
         let cmd = PageServiceCmd::parse(&format!("fullbackup {tenant_id} {timeline_id}")).unwrap();

pageserver/src/tenant.rs

@@ -158,7 +158,7 @@ pub struct TenantSharedResources {
     pub l0_flush_global_state: L0FlushGlobalState,
 }
 
-/// A [`TenantShard`] is really an _attached_ tenant.  The configuration
+/// A [`Tenant`] is really an _attached_ tenant.  The configuration
 /// for an attached tenant is a subset of the [`LocationConf`], represented
 /// in this struct.
 #[derive(Clone)]
@@ -245,7 +245,7 @@ pub(crate) enum SpawnMode {
 ///
 /// Tenant consists of multiple timelines. Keep them in a hash table.
 ///
-pub struct TenantShard {
+pub struct Tenant {
     // Global pageserver config parameters
     pub conf: &'static PageServerConf,
 
@@ -267,7 +267,7 @@ pub struct TenantShard {
     shard_identity: ShardIdentity,
 
     /// The remote storage generation, used to protect S3 objects from split-brain.
-    /// Does not change over the lifetime of the [`TenantShard`] object.
+    /// Does not change over the lifetime of the [`Tenant`] object.
     ///
     /// This duplicates the generation stored in LocationConf, but that structure is mutable:
     /// this copy enforces the invariant that generatio doesn't change during a Tenant's lifetime.
@@ -309,7 +309,7 @@ pub struct TenantShard {
     // Access to global deletion queue for when this tenant wants to schedule a deletion
     deletion_queue_client: DeletionQueueClient,
 
-    /// Cached logical sizes updated updated on each [`TenantShard::gather_size_inputs`].
+    /// Cached logical sizes updated updated on each [`Tenant::gather_size_inputs`].
     cached_logical_sizes: tokio::sync::Mutex<HashMap<(TimelineId, Lsn), u64>>,
     cached_synthetic_tenant_size: Arc<AtomicU64>,
 
@@ -337,12 +337,12 @@ pub struct TenantShard {
     // Timelines' cancellation token.
     pub(crate) cancel: CancellationToken,
 
-    // Users of the TenantShard such as the page service must take this Gate to avoid
-    // trying to use a TenantShard which is shutting down.
+    // Users of the Tenant such as the page service must take this Gate to avoid
+    // trying to use a Tenant which is shutting down.
     pub(crate) gate: Gate,
 
     /// Throttle applied at the top of [`Timeline::get`].
-    /// All [`TenantShard::timelines`] of a given [`TenantShard`] instance share the same [`throttle::Throttle`] instance.
+    /// All [`Tenant::timelines`] of a given [`Tenant`] instance share the same [`throttle::Throttle`] instance.
     pub(crate) pagestream_throttle: Arc<throttle::Throttle>,
 
     pub(crate) pagestream_throttle_metrics: Arc<crate::metrics::tenant_throttling::Pagestream>,
@@ -362,7 +362,7 @@ pub struct TenantShard {
 
     l0_flush_global_state: L0FlushGlobalState,
 }
-impl std::fmt::Debug for TenantShard {
+impl std::fmt::Debug for Tenant {
     fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
         write!(f, "{} ({})", self.tenant_shard_id, self.current_state())
     }
@@ -841,7 +841,7 @@ impl Debug for SetStoppingError {
     }
 }
 
-/// Arguments to [`TenantShard::create_timeline`].
+/// Arguments to [`Tenant::create_timeline`].
 ///
 /// Not usable as an idempotency key for timeline creation because if [`CreateTimelineParamsBranch::ancestor_start_lsn`]
 /// is `None`, the result of the timeline create call is not deterministic.
@@ -876,7 +876,7 @@ pub(crate) struct CreateTimelineParamsImportPgdata {
     pub(crate) idempotency_key: import_pgdata::index_part_format::IdempotencyKey,
 }
 
-/// What is used to determine idempotency of a [`TenantShard::create_timeline`] call in  [`TenantShard::start_creating_timeline`] in  [`TenantShard::start_creating_timeline`].
+/// What is used to determine idempotency of a [`Tenant::create_timeline`] call in  [`Tenant::start_creating_timeline`] in  [`Tenant::start_creating_timeline`].
 ///
 /// Each [`Timeline`] object holds [`Self`] as an immutable property in [`Timeline::create_idempotency`].
 ///
@@ -914,7 +914,7 @@ pub(crate) struct CreatingTimelineIdempotencyImportPgdata {
     idempotency_key: import_pgdata::index_part_format::IdempotencyKey,
 }
 
-/// What is returned by [`TenantShard::start_creating_timeline`].
+/// What is returned by [`Tenant::start_creating_timeline`].
 #[must_use]
 enum StartCreatingTimelineResult {
     CreateGuard(TimelineCreateGuard),
@@ -943,13 +943,13 @@ struct TimelineInitAndSyncNeedsSpawnImportPgdata {
     guard: TimelineCreateGuard,
 }
 
-/// What is returned by [`TenantShard::create_timeline`].
+/// What is returned by [`Tenant::create_timeline`].
 enum CreateTimelineResult {
     Created(Arc<Timeline>),
     Idempotent(Arc<Timeline>),
-    /// IMPORTANT: This [`Arc<Timeline>`] object is not in [`TenantShard::timelines`] when
+    /// IMPORTANT: This [`Arc<Timeline>`] object is not in [`Tenant::timelines`] when
     /// we return this result, nor will this concrete object ever be added there.
-    /// Cf method comment on [`TenantShard::create_timeline_import_pgdata`].
+    /// Cf method comment on [`Tenant::create_timeline_import_pgdata`].
     ImportSpawned(Arc<Timeline>),
 }
 
@@ -1082,7 +1082,7 @@ pub(crate) enum LoadConfigError {
     NotFound(Utf8PathBuf),
 }
 
-impl TenantShard {
+impl Tenant {
     /// Yet another helper for timeline initialization.
     ///
     /// - Initializes the Timeline struct and inserts it into the tenant's hash map
@@ -1303,7 +1303,7 @@ impl TenantShard {
         init_order: Option<InitializationOrder>,
         mode: SpawnMode,
         ctx: &RequestContext,
-    ) -> Result<Arc<TenantShard>, GlobalShutDown> {
+    ) -> Result<Arc<Tenant>, GlobalShutDown> {
         let wal_redo_manager =
             WalRedoManager::new(PostgresRedoManager::new(conf, tenant_shard_id))?;
 
@@ -1317,7 +1317,7 @@ impl TenantShard {
         let attach_mode = attached_conf.location.attach_mode;
         let generation = attached_conf.location.generation;
 
-        let tenant = Arc::new(TenantShard::new(
+        let tenant = Arc::new(Tenant::new(
             TenantState::Attaching,
             conf,
             attached_conf,
@@ -1334,7 +1334,7 @@ impl TenantShard {
         let attach_gate_guard = tenant
             .gate
             .enter()
-            .expect("We just created the TenantShard: nothing else can have shut it down yet");
+            .expect("We just created the Tenant: nothing else can have shut it down yet");
 
         // Do all the hard work in the background
         let tenant_clone = Arc::clone(&tenant);
@@ -1362,7 +1362,7 @@ impl TenantShard {
                     }
                 }
 
-                fn make_broken_or_stopping(t: &TenantShard, err: anyhow::Error) {
+                fn make_broken_or_stopping(t: &Tenant, err: anyhow::Error) {
                     t.state.send_modify(|state| match state {
                         // TODO: the old code alluded to DeleteTenantFlow sometimes setting
                         // TenantState::Stopping before we get here, but this may be outdated.
@@ -1627,7 +1627,7 @@ impl TenantShard {
     /// No background tasks are started as part of this routine.
     ///
     async fn attach(
-        self: &Arc<TenantShard>,
+        self: &Arc<Tenant>,
         preload: Option<TenantPreload>,
         ctx: &RequestContext,
     ) -> anyhow::Result<()> {
@@ -1957,7 +1957,7 @@ impl TenantShard {
     }
 
     async fn load_timelines_metadata(
-        self: &Arc<TenantShard>,
+        self: &Arc<Tenant>,
         timeline_ids: HashSet<TimelineId>,
         remote_storage: &GenericRemoteStorage,
         heatmap: Option<(HeatMapTenant, std::time::Instant)>,
@@ -2028,7 +2028,7 @@ impl TenantShard {
     }
 
     fn load_timeline_metadata(
-        self: &Arc<TenantShard>,
+        self: &Arc<Tenant>,
         timeline_id: TimelineId,
         remote_storage: GenericRemoteStorage,
         previous_heatmap: Option<PreviousHeatmap>,
@@ -2429,14 +2429,14 @@ impl TenantShard {
     /// This is used by tests & import-from-basebackup.
     ///
     /// The returned [`UninitializedTimeline`] contains no data nor metadata and it is in
-    /// a state that will fail [`TenantShard::load_remote_timeline`] because `disk_consistent_lsn=Lsn(0)`.
+    /// a state that will fail [`Tenant::load_remote_timeline`] because `disk_consistent_lsn=Lsn(0)`.
     ///
     /// The caller is responsible for getting the timeline into a state that will be accepted
-    /// by [`TenantShard::load_remote_timeline`] / [`TenantShard::attach`].
+    /// by [`Tenant::load_remote_timeline`] / [`Tenant::attach`].
     /// Then they may call [`UninitializedTimeline::finish_creation`] to add the timeline
-    /// to the [`TenantShard::timelines`].
+    /// to the [`Tenant::timelines`].
     ///
-    /// Tests should use `TenantShard::create_test_timeline` to set up the minimum required metadata keys.
+    /// Tests should use `Tenant::create_test_timeline` to set up the minimum required metadata keys.
     pub(crate) async fn create_empty_timeline(
         self: &Arc<Self>,
         new_timeline_id: TimelineId,
@@ -2584,7 +2584,7 @@ impl TenantShard {
     /// the same timeline ID already exists, returns CreateTimelineError::AlreadyExists.
     #[allow(clippy::too_many_arguments)]
     pub(crate) async fn create_timeline(
-        self: &Arc<TenantShard>,
+        self: &Arc<Tenant>,
         params: CreateTimelineParams,
         broker_client: storage_broker::BrokerClientChannel,
         ctx: &RequestContext,
@@ -2751,13 +2751,13 @@ impl TenantShard {
         Ok(activated_timeline)
     }
 
-    /// The returned [`Arc<Timeline>`] is NOT in the [`TenantShard::timelines`] map until the import
+    /// The returned [`Arc<Timeline>`] is NOT in the [`Tenant::timelines`] map until the import
     /// completes in the background. A DIFFERENT [`Arc<Timeline>`] will be inserted into the
-    /// [`TenantShard::timelines`] map when the import completes.
+    /// [`Tenant::timelines`] map when the import completes.
     /// We only return an [`Arc<Timeline>`] here so the API handler can create a [`pageserver_api::models::TimelineInfo`]
     /// for the response.
     async fn create_timeline_import_pgdata(
-        self: &Arc<Self>,
+        self: &Arc<Tenant>,
         params: CreateTimelineParamsImportPgdata,
         activate: ActivateTimelineArgs,
         ctx: &RequestContext,
@@ -2854,7 +2854,7 @@ impl TenantShard {
 
     #[instrument(skip_all, fields(tenant_id=%self.tenant_shard_id.tenant_id, shard_id=%self.tenant_shard_id.shard_slug(), timeline_id=%timeline.timeline_id))]
     async fn create_timeline_import_pgdata_task(
-        self: Arc<TenantShard>,
+        self: Arc<Tenant>,
         timeline: Arc<Timeline>,
         index_part: import_pgdata::index_part_format::Root,
         activate: ActivateTimelineArgs,
@@ -2882,7 +2882,7 @@ impl TenantShard {
     }
 
     async fn create_timeline_import_pgdata_task_impl(
-        self: Arc<TenantShard>,
+        self: Arc<Tenant>,
         timeline: Arc<Timeline>,
         index_part: import_pgdata::index_part_format::Root,
         activate: ActivateTimelineArgs,
@@ -2899,10 +2899,10 @@ impl TenantShard {
         // Reload timeline from remote.
         // This proves that the remote state is attachable, and it reuses the code.
         //
-        // TODO: think about whether this is safe to do with concurrent TenantShard::shutdown.
+        // TODO: think about whether this is safe to do with concurrent Tenant::shutdown.
         // timeline_create_guard hols the tenant gate open, so, shutdown cannot _complete_ until we exit.
-        // But our activate() call might launch new background tasks after TenantShard::shutdown
-        // already went past shutting down the TenantShard::timelines, which this timeline here is no part of.
+        // But our activate() call might launch new background tasks after Tenant::shutdown
+        // already went past shutting down the Tenant::timelines, which this timeline here is no part of.
         // I think the same problem exists with the bootstrap & branch mgmt API tasks (tenant shutting
         // down while bootstrapping/branching + activating), but, the race condition is much more likely
         // to manifest because of the long runtime of this import task.
@@ -2917,7 +2917,7 @@ impl TenantShard {
         // };
         let timeline_id = timeline.timeline_id;
 
-        // load from object storage like TenantShard::attach does
+        // load from object storage like Tenant::attach does
         let resources = self.build_timeline_resources(timeline_id);
         let index_part = resources
             .remote_client
@@ -3938,7 +3938,7 @@ enum ActivateTimelineArgs {
     No,
 }
 
-impl TenantShard {
+impl Tenant {
     pub fn tenant_specific_overrides(&self) -> pageserver_api::models::TenantConfig {
         self.tenant_conf.load().tenant_conf.clone()
     }
@@ -4096,7 +4096,7 @@ impl TenantShard {
         update: F,
     ) -> anyhow::Result<pageserver_api::models::TenantConfig> {
         // Use read-copy-update in order to avoid overwriting the location config
-        // state if this races with [`TenantShard::set_new_location_config`]. Note that
+        // state if this races with [`Tenant::set_new_location_config`]. Note that
         // this race is not possible if both request types come from the storage
         // controller (as they should!) because an exclusive op lock is required
         // on the storage controller side.
@@ -4219,7 +4219,7 @@ impl TenantShard {
         Ok((timeline, timeline_ctx))
     }
 
-    /// [`TenantShard::shutdown`] must be called before dropping the returned [`TenantShard`] object
+    /// [`Tenant::shutdown`] must be called before dropping the returned [`Tenant`] object
     /// to ensure proper cleanup of background tasks and metrics.
     //
     // Allow too_many_arguments because a constructor's argument list naturally grows with the
@@ -4235,7 +4235,7 @@ impl TenantShard {
         remote_storage: GenericRemoteStorage,
         deletion_queue_client: DeletionQueueClient,
         l0_flush_global_state: L0FlushGlobalState,
-    ) -> TenantShard {
+    ) -> Tenant {
         debug_assert!(
             !attached_conf.location.generation.is_none() || conf.control_plane_api.is_none()
         );
@@ -4295,7 +4295,7 @@ impl TenantShard {
             }
         });
 
-        TenantShard {
+        Tenant {
             tenant_shard_id,
             shard_identity,
             generation: attached_conf.location.generation,
@@ -4330,7 +4330,7 @@ impl TenantShard {
             cancel: CancellationToken::default(),
             gate: Gate::default(),
             pagestream_throttle: Arc::new(throttle::Throttle::new(
-                TenantShard::get_pagestream_throttle_config(conf, &attached_conf.tenant_conf),
+                Tenant::get_pagestream_throttle_config(conf, &attached_conf.tenant_conf),
             )),
             pagestream_throttle_metrics: Arc::new(
                 crate::metrics::tenant_throttling::Pagestream::new(&tenant_shard_id),
@@ -4466,11 +4466,11 @@ impl TenantShard {
 
         // Perform GC for each timeline.
         //
-        // Note that we don't hold the `TenantShard::gc_cs` lock here because we don't want to delay the
+        // Note that we don't hold the `Tenant::gc_cs` lock here because we don't want to delay the
         // branch creation task, which requires the GC lock. A GC iteration can run concurrently
         // with branch creation.
         //
-        // See comments in [`TenantShard::branch_timeline`] for more information about why branch
+        // See comments in [`Tenant::branch_timeline`] for more information about why branch
         // creation task can run concurrently with timeline's GC iteration.
         for timeline in gc_timelines {
             if cancel.is_cancelled() {
@@ -4500,7 +4500,7 @@ impl TenantShard {
 
     /// Refreshes the Timeline::gc_info for all timelines, returning the
     /// vector of timelines which have [`Timeline::get_last_record_lsn`] past
-    /// [`TenantShard::get_gc_horizon`].
+    /// [`Tenant::get_gc_horizon`].
     ///
     /// This is usually executed as part of periodic gc, but can now be triggered more often.
     pub(crate) async fn refresh_gc_info(
@@ -5499,7 +5499,7 @@ impl TenantShard {
             }
         }
 
-        // The flushes we did above were just writes, but the TenantShard might have had
+        // The flushes we did above were just writes, but the Tenant might have had
         // pending deletions as well from recent compaction/gc: we want to flush those
         // as well.  This requires flushing the global delete queue.  This is cheap
         // because it's typically a no-op.
@@ -5517,7 +5517,7 @@ impl TenantShard {
 
     /// How much local storage would this tenant like to have?  It can cope with
     /// less than this (via eviction and on-demand downloads), but this function enables
-    /// the TenantShard to advertise how much storage it would prefer to have to provide fast I/O
+    /// the Tenant to advertise how much storage it would prefer to have to provide fast I/O
     /// by keeping important things on local disk.
     ///
     /// This is a heuristic, not a guarantee: tenants that are long-idle will actually use less
@@ -5540,11 +5540,11 @@ impl TenantShard {
     /// manifest in `Self::remote_tenant_manifest`.
     ///
     /// TODO: instead of requiring callers to remember to call `maybe_upload_tenant_manifest` after
-    /// changing any `TenantShard` state that's included in the manifest, consider making the manifest
+    /// changing any `Tenant` state that's included in the manifest, consider making the manifest
     /// the authoritative source of data with an API that automatically uploads on changes. Revisit
     /// this when the manifest is more widely used and we have a better idea of the data model.
     pub(crate) async fn maybe_upload_tenant_manifest(&self) -> Result<(), TenantManifestError> {
-        // Multiple tasks may call this function concurrently after mutating the TenantShard runtime
+        // Multiple tasks may call this function concurrently after mutating the Tenant runtime
         // state, affecting the manifest generated by `build_tenant_manifest`. We use an async mutex
         // to serialize these callers. `eq_ignoring_version` acts as a slightly inefficient but
         // simple coalescing mechanism.
@@ -5812,7 +5812,7 @@ pub(crate) mod harness {
             info_span!("TenantHarness", tenant_id=%self.tenant_shard_id.tenant_id, shard_id=%self.tenant_shard_id.shard_slug())
         }
 
-        pub(crate) async fn load(&self) -> (Arc<TenantShard>, RequestContext) {
+        pub(crate) async fn load(&self) -> (Arc<Tenant>, RequestContext) {
             let ctx = RequestContext::new(TaskKind::UnitTest, DownloadBehavior::Error)
                 .with_scope_unit_test();
             (
@@ -5827,10 +5827,10 @@ pub(crate) mod harness {
         pub(crate) async fn do_try_load(
             &self,
             ctx: &RequestContext,
-        ) -> anyhow::Result<Arc<TenantShard>> {
+        ) -> anyhow::Result<Arc<Tenant>> {
             let walredo_mgr = Arc::new(WalRedoManager::from(TestRedoManager));
 
-            let tenant = Arc::new(TenantShard::new(
+            let tenant = Arc::new(Tenant::new(
                 TenantState::Attaching,
                 self.conf,
                 AttachedTenantConf::try_from(LocationConf::attached_single(
@@ -6046,7 +6046,7 @@ mod tests {
     #[cfg(feature = "testing")]
     #[allow(clippy::too_many_arguments)]
     async fn randomize_timeline(
-        tenant: &Arc<TenantShard>,
+        tenant: &Arc<Tenant>,
         new_timeline_id: TimelineId,
         pg_version: u32,
         spec: TestTimelineSpecification,
@@ -6936,7 +6936,7 @@ mod tests {
     }
 
     async fn bulk_insert_compact_gc(
-        tenant: &TenantShard,
+        tenant: &Tenant,
         timeline: &Arc<Timeline>,
         ctx: &RequestContext,
         lsn: Lsn,
@@ -6948,7 +6948,7 @@ mod tests {
     }
 
     async fn bulk_insert_maybe_compact_gc(
-        tenant: &TenantShard,
+        tenant: &Tenant,
         timeline: &Arc<Timeline>,
         ctx: &RequestContext,
         mut lsn: Lsn,
@@ -7858,7 +7858,7 @@ mod tests {
             let (tline, _ctx) = tenant
                 .create_empty_timeline(TIMELINE_ID, Lsn(0), DEFAULT_PG_VERSION, &ctx)
                 .await?;
-            // Leave the timeline ID in [`TenantShard::timelines_creating`] to exclude attempting to create it again
+            // Leave the timeline ID in [`Tenant::timelines_creating`] to exclude attempting to create it again
             let raw_tline = tline.raw_timeline().unwrap();
             raw_tline
                 .shutdown(super::timeline::ShutdownMode::Hard)

pageserver/src/tenant/blob_io.rs

@@ -37,63 +37,6 @@ pub struct CompressionInfo {
     pub compressed_size: Option<usize>,
 }
 
-/// A blob header, with header+data length and compression info.
-///
-/// TODO: use this more widely, and add an encode() method too.
-/// TODO: document the header format.
-#[derive(Clone, Copy, Default)]
-pub struct Header {
-    pub header_len: usize,
-    pub data_len: usize,
-    pub compression_bits: u8,
-}
-
-impl Header {
-    /// Decodes a header from a byte slice.
-    pub fn decode(bytes: &[u8]) -> Result<Self, std::io::Error> {
-        let Some(&first_header_byte) = bytes.first() else {
-            return Err(std::io::Error::new(
-                std::io::ErrorKind::InvalidData,
-                "zero-length blob header",
-            ));
-        };
-
-        // If the first bit is 0, this is just a 1-byte length prefix up to 128 bytes.
-        if first_header_byte < 0x80 {
-            return Ok(Self {
-                header_len: 1, // by definition
-                data_len: first_header_byte as usize,
-                compression_bits: BYTE_UNCOMPRESSED,
-            });
-        }
-
-        // Otherwise, this is a 4-byte header containing compression information and length.
-        const HEADER_LEN: usize = 4;
-        let mut header_buf: [u8; HEADER_LEN] = bytes[0..HEADER_LEN].try_into().map_err(|_| {
-            std::io::Error::new(
-                std::io::ErrorKind::InvalidData,
-                format!("blob header too short: {bytes:?}"),
-            )
-        })?;
-
-        // TODO: verify the compression bits and convert to an enum.
-        let compression_bits = header_buf[0] & LEN_COMPRESSION_BIT_MASK;
-        header_buf[0] &= !LEN_COMPRESSION_BIT_MASK;
-        let data_len = u32::from_be_bytes(header_buf) as usize;
-
-        Ok(Self {
-            header_len: HEADER_LEN,
-            data_len,
-            compression_bits,
-        })
-    }
-
-    /// Returns the total header+data length.
-    pub fn total_len(&self) -> usize {
-        self.header_len + self.data_len
-    }
-}
-
 impl BlockCursor<'_> {
     /// Read a blob into a new buffer.
     pub async fn read_blob(
@@ -446,34 +389,6 @@ impl<const BUFFERED: bool> BlobWriter<BUFFERED> {
         };
         (srcbuf, res.map(|_| (offset, compression_info)))
     }
-
-    /// Writes a raw blob containing both header and data, returning its offset.
-    pub(crate) async fn write_blob_raw<Buf: IoBuf + Send>(
-        &mut self,
-        raw_with_header: FullSlice<Buf>,
-        ctx: &RequestContext,
-    ) -> (FullSlice<Buf>, Result<u64, Error>) {
-        // Verify the header, to ensure we don't write invalid/corrupt data.
-        let header = match Header::decode(&raw_with_header) {
-            Ok(header) => header,
-            Err(err) => return (raw_with_header, Err(err)),
-        };
-        if raw_with_header.len() != header.total_len() {
-            let header_total_len = header.total_len();
-            let raw_len = raw_with_header.len();
-            return (
-                raw_with_header,
-                Err(std::io::Error::new(
-                    std::io::ErrorKind::InvalidData,
-                    format!("header length mismatch: {header_total_len} != {raw_len}"),
-                )),
-            );
-        }
-
-        let offset = self.offset;
-        let (raw_with_header, result) = self.write_all(raw_with_header, ctx).await;
-        (raw_with_header, result.map(|_| offset))
-    }
 }
 
 impl BlobWriter<true> {

pageserver/src/tenant/metadata.rs

@@ -564,9 +564,8 @@ mod tests {
             Lsn(0),
             Lsn(0),
             Lsn(0),
-            // Updating this version to 17 will cause the test to fail at the
-            // next assert_eq!().
-            16,
+            // Any version will do here, so use the default
+            crate::DEFAULT_PG_VERSION,
         );
         let expected_bytes = vec![
             /* TimelineMetadataHeader */

pageserver/src/tenant/mgr.rs

@@ -52,9 +52,7 @@ use crate::tenant::config::{
 use crate::tenant::span::debug_assert_current_span_has_tenant_id;
 use crate::tenant::storage_layer::inmemory_layer;
 use crate::tenant::timeline::ShutdownMode;
-use crate::tenant::{
-    AttachedTenantConf, GcError, LoadConfigError, SpawnMode, TenantShard, TenantState,
-};
+use crate::tenant::{AttachedTenantConf, GcError, LoadConfigError, SpawnMode, Tenant, TenantState};
 use crate::virtual_file::MaybeFatalIo;
 use crate::{InitializationOrder, TEMP_FILE_SUFFIX};
 
@@ -69,7 +67,7 @@ use crate::{InitializationOrder, TEMP_FILE_SUFFIX};
 /// having a properly acquired generation (Secondary doesn't need a generation)
 #[derive(Clone)]
 pub(crate) enum TenantSlot {
-    Attached(Arc<TenantShard>),
+    Attached(Arc<Tenant>),
     Secondary(Arc<SecondaryTenant>),
     /// In this state, other administrative operations acting on the TenantId should
     /// block, or return a retry indicator equivalent to HTTP 503.
@@ -88,7 +86,7 @@ impl std::fmt::Debug for TenantSlot {
 
 impl TenantSlot {
     /// Return the `Tenant` in this slot if attached, else None
-    fn get_attached(&self) -> Option<&Arc<TenantShard>> {
+    fn get_attached(&self) -> Option<&Arc<Tenant>> {
         match self {
             Self::Attached(t) => Some(t),
             Self::Secondary(_) => None,
@@ -166,7 +164,7 @@ impl TenantStartupMode {
 /// Result type for looking up a TenantId to a specific shard
 pub(crate) enum ShardResolveResult {
     NotFound,
-    Found(Arc<TenantShard>),
+    Found(Arc<Tenant>),
     // Wait for this barrrier, then query again
     InProgress(utils::completion::Barrier),
 }
@@ -175,7 +173,7 @@ impl TenantsMap {
     /// Convenience function for typical usage, where we want to get a `Tenant` object, for
     /// working with attached tenants.  If the TenantId is in the map but in Secondary state,
     /// None is returned.
-    pub(crate) fn get(&self, tenant_shard_id: &TenantShardId) -> Option<&Arc<TenantShard>> {
+    pub(crate) fn get(&self, tenant_shard_id: &TenantShardId) -> Option<&Arc<Tenant>> {
         match self {
             TenantsMap::Initializing => None,
             TenantsMap::Open(m) | TenantsMap::ShuttingDown(m) => {
@@ -412,7 +410,7 @@ fn load_tenant_config(
         return None;
     }
 
-    Some(TenantShard::load_tenant_config(conf, &tenant_shard_id))
+    Some(Tenant::load_tenant_config(conf, &tenant_shard_id))
 }
 
 /// Initial stage of load: walk the local tenants directory, clean up any temp files,
@@ -608,8 +606,7 @@ pub async fn init_tenant_mgr(
         // Presence of a generation number implies attachment: attach the tenant
         // if it wasn't already, and apply the generation number.
         config_write_futs.push(async move {
-            let r =
-                TenantShard::persist_tenant_config(conf, &tenant_shard_id, &location_conf).await;
+            let r = Tenant::persist_tenant_config(conf, &tenant_shard_id, &location_conf).await;
             (tenant_shard_id, location_conf, r)
         });
     }
@@ -697,7 +694,7 @@ fn tenant_spawn(
     init_order: Option<InitializationOrder>,
     mode: SpawnMode,
     ctx: &RequestContext,
-) -> Result<Arc<TenantShard>, GlobalShutDown> {
+) -> Result<Arc<Tenant>, GlobalShutDown> {
     // All these conditions should have been satisfied by our caller: the tenant dir exists, is a well formed
     // path, and contains a configuration file.  Assertions that do synchronous I/O are limited to debug mode
     // to avoid impacting prod runtime performance.
@@ -709,7 +706,7 @@ fn tenant_spawn(
             .unwrap()
     );
 
-    TenantShard::spawn(
+    Tenant::spawn(
         conf,
         tenant_shard_id,
         resources,
@@ -886,12 +883,12 @@ impl TenantManager {
     /// Gets the attached tenant from the in-memory data, erroring if it's absent, in secondary mode, or currently
     /// undergoing a state change (i.e. slot is InProgress).
     ///
-    /// The return TenantShard is not guaranteed to be active: check its status after obtaing it, or
-    /// use [`TenantShard::wait_to_become_active`] before using it if you will do I/O on it.
+    /// The return Tenant is not guaranteed to be active: check its status after obtaing it, or
+    /// use [`Tenant::wait_to_become_active`] before using it if you will do I/O on it.
     pub(crate) fn get_attached_tenant_shard(
         &self,
         tenant_shard_id: TenantShardId,
-    ) -> Result<Arc<TenantShard>, GetTenantError> {
+    ) -> Result<Arc<Tenant>, GetTenantError> {
         let locked = self.tenants.read().unwrap();
 
         let peek_slot = tenant_map_peek_slot(&locked, &tenant_shard_id, TenantSlotPeekMode::Read)?;
@@ -940,12 +937,12 @@ impl TenantManager {
         flush: Option<Duration>,
         mut spawn_mode: SpawnMode,
         ctx: &RequestContext,
-    ) -> Result<Option<Arc<TenantShard>>, UpsertLocationError> {
+    ) -> Result<Option<Arc<Tenant>>, UpsertLocationError> {
         debug_assert_current_span_has_tenant_id();
         info!("configuring tenant location to state {new_location_config:?}");
 
         enum FastPathModified {
-            Attached(Arc<TenantShard>),
+            Attached(Arc<Tenant>),
             Secondary(Arc<SecondaryTenant>),
         }
 
@@ -1002,13 +999,9 @@ impl TenantManager {
         // phase of writing config and/or waiting for flush, before returning.
         match fast_path_taken {
             Some(FastPathModified::Attached(tenant)) => {
-                TenantShard::persist_tenant_config(
-                    self.conf,
-                    &tenant_shard_id,
-                    &new_location_config,
-                )
-                .await
-                .fatal_err("write tenant shard config");
+                Tenant::persist_tenant_config(self.conf, &tenant_shard_id, &new_location_config)
+                    .await
+                    .fatal_err("write tenant shard config");
 
                 // Transition to AttachedStale means we may well hold a valid generation
                 // still, and have been requested to go stale as part of a migration.  If
@@ -1037,13 +1030,9 @@ impl TenantManager {
                 return Ok(Some(tenant));
             }
             Some(FastPathModified::Secondary(_secondary_tenant)) => {
-                TenantShard::persist_tenant_config(
-                    self.conf,
-                    &tenant_shard_id,
-                    &new_location_config,
-                )
-                .await
-                .fatal_err("write tenant shard config");
+                Tenant::persist_tenant_config(self.conf, &tenant_shard_id, &new_location_config)
+                    .await
+                    .fatal_err("write tenant shard config");
 
                 return Ok(None);
             }
@@ -1133,7 +1122,7 @@ impl TenantManager {
         // Before activating either secondary or attached mode, persist the
         // configuration, so that on restart we will re-attach (or re-start
         // secondary) on the tenant.
-        TenantShard::persist_tenant_config(self.conf, &tenant_shard_id, &new_location_config)
+        Tenant::persist_tenant_config(self.conf, &tenant_shard_id, &new_location_config)
             .await
             .fatal_err("write tenant shard config");
 
@@ -1273,7 +1262,7 @@ impl TenantManager {
 
         let tenant_path = self.conf.tenant_path(&tenant_shard_id);
         let timelines_path = self.conf.timelines_path(&tenant_shard_id);
-        let config = TenantShard::load_tenant_config(self.conf, &tenant_shard_id)?;
+        let config = Tenant::load_tenant_config(self.conf, &tenant_shard_id)?;
 
         if drop_cache {
             tracing::info!("Dropping local file cache");
@@ -1308,7 +1297,7 @@ impl TenantManager {
         Ok(())
     }
 
-    pub(crate) fn get_attached_active_tenant_shards(&self) -> Vec<Arc<TenantShard>> {
+    pub(crate) fn get_attached_active_tenant_shards(&self) -> Vec<Arc<Tenant>> {
         let locked = self.tenants.read().unwrap();
         match &*locked {
             TenantsMap::Initializing => Vec::new(),
@@ -1457,7 +1446,7 @@ impl TenantManager {
     #[instrument(skip_all, fields(tenant_id=%tenant.get_tenant_shard_id().tenant_id, shard_id=%tenant.get_tenant_shard_id().shard_slug(), new_shard_count=%new_shard_count.literal()))]
     pub(crate) async fn shard_split(
         &self,
-        tenant: Arc<TenantShard>,
+        tenant: Arc<Tenant>,
         new_shard_count: ShardCount,
         new_stripe_size: Option<ShardStripeSize>,
         ctx: &RequestContext,
@@ -1487,7 +1476,7 @@ impl TenantManager {
 
     pub(crate) async fn do_shard_split(
         &self,
-        tenant: Arc<TenantShard>,
+        tenant: Arc<Tenant>,
         new_shard_count: ShardCount,
         new_stripe_size: Option<ShardStripeSize>,
         ctx: &RequestContext,
@@ -1714,7 +1703,7 @@ impl TenantManager {
     /// For each resident layer in the parent shard, we will hard link it into all of the child shards.
     async fn shard_split_hardlink(
         &self,
-        parent_shard: &TenantShard,
+        parent_shard: &Tenant,
         child_shards: Vec<TenantShardId>,
     ) -> anyhow::Result<()> {
         debug_assert_current_span_has_tenant_id();
@@ -1999,7 +1988,7 @@ impl TenantManager {
             }
 
             let tenant_path = self.conf.tenant_path(&tenant_shard_id);
-            let config = TenantShard::load_tenant_config(self.conf, &tenant_shard_id)
+            let config = Tenant::load_tenant_config(self.conf, &tenant_shard_id)
                 .map_err(|e| Error::DetachReparent(e.into()))?;
 
             let shard_identity = config.shard;

pageserver/src/tenant/remote_timeline_client.rs

@@ -133,7 +133,7 @@
 //! - Initiate upload queue with that [`IndexPart`].
 //! - Reschedule all lost operations by comparing the local filesystem state
 //!   and remote state as per [`IndexPart`]. This is done in
-//!   [`TenantShard::timeline_init_and_sync`].
+//!   [`Tenant::timeline_init_and_sync`].
 //!
 //! Note that if we crash during file deletion between the index update
 //! that removes the file from the list of files, and deleting the remote file,
@@ -171,7 +171,7 @@
 //! If no remote storage configuration is provided, the [`RemoteTimelineClient`] is
 //! not created and the uploads are skipped.
 //!
-//! [`TenantShard::timeline_init_and_sync`]: super::TenantShard::timeline_init_and_sync
+//! [`Tenant::timeline_init_and_sync`]: super::Tenant::timeline_init_and_sync
 //! [`Timeline::load_layer_map`]: super::Timeline::load_layer_map
 
 pub(crate) mod download;
@@ -2743,7 +2743,7 @@ mod tests {
     use crate::tenant::config::AttachmentMode;
     use crate::tenant::harness::{TIMELINE_ID, TenantHarness};
     use crate::tenant::storage_layer::layer::local_layer_path;
-    use crate::tenant::{TenantShard, Timeline};
+    use crate::tenant::{Tenant, Timeline};
 
     pub(super) fn dummy_contents(name: &str) -> Vec<u8> {
         format!("contents for {name}").into()
@@ -2796,7 +2796,7 @@ mod tests {
 
     struct TestSetup {
         harness: TenantHarness,
-        tenant: Arc<TenantShard>,
+        tenant: Arc<Tenant>,
         timeline: Arc<Timeline>,
         tenant_ctx: RequestContext,
     }

pageserver/src/tenant/remote_timeline_client/download.rs

@@ -452,7 +452,7 @@ async fn do_download_index_part(
 /// generation (normal case when migrating/restarting).  Only if both of these return 404 do we fall back
 /// to listing objects.
 ///
-/// * `my_generation`: the value of `[crate::tenant::TenantShard::generation]`
+/// * `my_generation`: the value of `[crate::tenant::Tenant::generation]`
 /// * `what`: for logging, what object are we downloading
 /// * `prefix`: when listing objects, use this prefix (i.e. the part of the object path before the generation)
 /// * `do_download`: a GET of the object in a particular generation, which should **retry indefinitely** unless

pageserver/src/tenant/secondary/heatmap_uploader.rs

@@ -21,7 +21,7 @@ use super::scheduler::{
 use super::{CommandRequest, SecondaryTenantError, UploadCommand};
 use crate::TEMP_FILE_SUFFIX;
 use crate::metrics::SECONDARY_MODE;
-use crate::tenant::TenantShard;
+use crate::tenant::Tenant;
 use crate::tenant::config::AttachmentMode;
 use crate::tenant::mgr::{GetTenantError, TenantManager};
 use crate::tenant::remote_timeline_client::remote_heatmap_path;
@@ -74,7 +74,7 @@ impl RunningJob for WriteInProgress {
 }
 
 struct UploadPending {
-    tenant: Arc<TenantShard>,
+    tenant: Arc<Tenant>,
     last_upload: Option<LastUploadState>,
     target_time: Option<Instant>,
     period: Option<Duration>,
@@ -106,7 +106,7 @@ impl scheduler::Completion for WriteComplete {
 struct UploaderTenantState {
     // This Weak only exists to enable culling idle instances of this type
     // when the Tenant has been deallocated.
-    tenant: Weak<TenantShard>,
+    tenant: Weak<Tenant>,
 
     /// Digest of the serialized heatmap that we last successfully uploaded
     last_upload_state: Option<LastUploadState>,
@@ -357,7 +357,7 @@ struct LastUploadState {
 /// of the object we would have uploaded.
 async fn upload_tenant_heatmap(
     remote_storage: GenericRemoteStorage,
-    tenant: &Arc<TenantShard>,
+    tenant: &Arc<Tenant>,
     last_upload: Option<LastUploadState>,
 ) -> Result<UploadHeatmapOutcome, UploadHeatmapError> {
     debug_assert_current_span_has_tenant_id();

pageserver/src/tenant/secondary/scheduler.rs

@@ -360,7 +360,7 @@ where
 
     /// Periodic execution phase: inspect all attached tenants and schedule any work they require.
     ///
-    /// The type in `tenants` should be a tenant-like structure, e.g. [`crate::tenant::TenantShard`] or [`crate::tenant::secondary::SecondaryTenant`]
+    /// The type in `tenants` should be a tenant-like structure, e.g. [`crate::tenant::Tenant`] or [`crate::tenant::secondary::SecondaryTenant`]
     ///
     /// This function resets the pending list: it is assumed that the caller may change their mind about
     /// which tenants need work between calls to schedule_iteration.

pageserver/src/tenant/size.rs

@@ -12,7 +12,7 @@ use tracing::*;
 use utils::id::TimelineId;
 use utils::lsn::Lsn;
 
-use super::{GcError, LogicalSizeCalculationCause, TenantShard};
+use super::{GcError, LogicalSizeCalculationCause, Tenant};
 use crate::context::RequestContext;
 use crate::pgdatadir_mapping::CalculateLogicalSizeError;
 use crate::tenant::{MaybeOffloaded, Timeline};
@@ -156,7 +156,7 @@ pub struct TimelineInputs {
 ///   initdb_lsn  branchpoints*  next_pitr_cutoff  latest
 /// ```
 pub(super) async fn gather_inputs(
-    tenant: &TenantShard,
+    tenant: &Tenant,
     limit: &Arc<Semaphore>,
     max_retention_period: Option<u64>,
     logical_size_cache: &mut HashMap<(TimelineId, Lsn), u64>,

pageserver/src/tenant/storage_layer/delta_layer.rs

@@ -1620,7 +1620,7 @@ pub(crate) mod test {
     use crate::tenant::harness::{TIMELINE_ID, TenantHarness};
     use crate::tenant::storage_layer::{Layer, ResidentLayer};
     use crate::tenant::vectored_blob_io::StreamingVectoredReadPlanner;
-    use crate::tenant::{TenantShard, Timeline};
+    use crate::tenant::{Tenant, Timeline};
 
     /// Construct an index for a fictional delta layer and and then
     /// traverse in order to plan vectored reads for a query. Finally,
@@ -2209,7 +2209,7 @@ pub(crate) mod test {
     }
 
     pub(crate) async fn produce_delta_layer(
-        tenant: &TenantShard,
+        tenant: &Tenant,
         tline: &Arc<Timeline>,
         mut deltas: Vec<(Key, Lsn, Value)>,
         ctx: &RequestContext,

pageserver/src/tenant/storage_layer/image_layer.rs

@@ -559,12 +559,11 @@ impl ImageLayerInner {
             let view = BufView::new_slice(&blobs_buf.buf);
 
             for meta in blobs_buf.blobs.iter() {
-                // Just read the raw header+data and pass it through to the target layer, without
-                // decoding and recompressing it.
-                let raw = meta.raw_with_header(&view);
+                let img_buf = meta.read(&view).await?;
+
                 key_count += 1;
                 writer
-                    .put_image_raw(meta.meta.key, raw.into_bytes(), ctx)
+                    .put_image(meta.meta.key, img_buf.into_bytes(), ctx)
                     .await
                     .context(format!("Storing key {}", meta.meta.key))?;
             }
@@ -854,41 +853,6 @@ impl ImageLayerWriterInner {
         Ok(())
     }
 
-    ///
-    /// Write the next image to the file, as a raw blob header and data.
-    ///
-    /// The page versions must be appended in blknum order.
-    ///
-    async fn put_image_raw(
-        &mut self,
-        key: Key,
-        raw_with_header: Bytes,
-        ctx: &RequestContext,
-    ) -> anyhow::Result<()> {
-        ensure!(self.key_range.contains(&key));
-
-        // NB: we don't update the (un)compressed metrics, since we can't determine them without
-        // decompressing the image. This seems okay.
-        self.num_keys += 1;
-
-        let (_, res) = self
-            .blob_writer
-            .write_blob_raw(raw_with_header.slice_len(), ctx)
-            .await;
-        let offset = res?;
-
-        let mut keybuf: [u8; KEY_SIZE] = [0u8; KEY_SIZE];
-        key.write_to_byte_slice(&mut keybuf);
-        self.tree.append(&keybuf, offset)?;
-
-        #[cfg(feature = "testing")]
-        {
-            self.last_written_key = key;
-        }
-
-        Ok(())
-    }
-
     ///
     /// Finish writing the image layer.
     ///
@@ -924,13 +888,7 @@ impl ImageLayerWriterInner {
         crate::metrics::COMPRESSION_IMAGE_INPUT_BYTES_CONSIDERED
             .inc_by(self.uncompressed_bytes_eligible);
         crate::metrics::COMPRESSION_IMAGE_INPUT_BYTES_CHOSEN.inc_by(self.uncompressed_bytes_chosen);
-
-        // NB: filter() may pass through raw pages from a different layer, without looking at
-        // whether these are compressed or not. We don't track metrics for these, so avoid
-        // increasing `COMPRESSION_IMAGE_OUTPUT_BYTES` in this case too.
-        if self.uncompressed_bytes > 0 {
-            crate::metrics::COMPRESSION_IMAGE_OUTPUT_BYTES.inc_by(compressed_size);
-        };
+        crate::metrics::COMPRESSION_IMAGE_OUTPUT_BYTES.inc_by(compressed_size);
 
         let mut file = self.blob_writer.into_inner();
 
@@ -1076,25 +1034,6 @@ impl ImageLayerWriter {
         self.inner.as_mut().unwrap().put_image(key, img, ctx).await
     }
 
-    ///
-    /// Write the next value to the file, as a raw header and data. This allows passing through a
-    /// raw, potentially compressed image from a different layer file without recompressing it.
-    ///
-    /// The page versions must be appended in blknum order.
-    ///
-    pub async fn put_image_raw(
-        &mut self,
-        key: Key,
-        raw_with_header: Bytes,
-        ctx: &RequestContext,
-    ) -> anyhow::Result<()> {
-        self.inner
-            .as_mut()
-            .unwrap()
-            .put_image_raw(key, raw_with_header, ctx)
-            .await
-    }
-
     /// Estimated size of the image layer.
     pub(crate) fn estimated_size(&self) -> u64 {
         let inner = self.inner.as_ref().unwrap();
@@ -1228,7 +1167,7 @@ mod test {
     use crate::tenant::harness::{TIMELINE_ID, TenantHarness};
     use crate::tenant::storage_layer::{Layer, ResidentLayer};
     use crate::tenant::vectored_blob_io::StreamingVectoredReadPlanner;
-    use crate::tenant::{TenantShard, Timeline};
+    use crate::tenant::{Tenant, Timeline};
 
     #[tokio::test]
     async fn image_layer_rewrite() {
@@ -1410,7 +1349,7 @@ mod test {
     }
 
     async fn produce_image_layer(
-        tenant: &TenantShard,
+        tenant: &Tenant,
         tline: &Arc<Timeline>,
         mut images: Vec<(Key, Bytes)>,
         lsn: Lsn,

pageserver/src/tenant/tasks.rs

@@ -24,7 +24,7 @@ use crate::task_mgr::{self, BACKGROUND_RUNTIME, TOKIO_WORKER_THREADS, TaskKind};
 use crate::tenant::throttle::Stats;
 use crate::tenant::timeline::CompactionError;
 use crate::tenant::timeline::compaction::CompactionOutcome;
-use crate::tenant::{TenantShard, TenantState};
+use crate::tenant::{Tenant, TenantState};
 
 /// Semaphore limiting concurrent background tasks (across all tenants).
 ///
@@ -117,7 +117,7 @@ pub(crate) async fn acquire_concurrency_permit(
 }
 
 /// Start per tenant background loops: compaction, GC, and ingest housekeeping.
-pub fn start_background_loops(tenant: &Arc<TenantShard>, can_start: Option<&Barrier>) {
+pub fn start_background_loops(tenant: &Arc<Tenant>, can_start: Option<&Barrier>) {
     let tenant_shard_id = tenant.tenant_shard_id;
 
     task_mgr::spawn(
@@ -198,7 +198,7 @@ pub fn start_background_loops(tenant: &Arc<TenantShard>, can_start: Option<&Barr
 }
 
 /// Compaction task's main loop.
-async fn compaction_loop(tenant: Arc<TenantShard>, cancel: CancellationToken) {
+async fn compaction_loop(tenant: Arc<Tenant>, cancel: CancellationToken) {
     const BASE_BACKOFF_SECS: f64 = 1.0;
     const MAX_BACKOFF_SECS: f64 = 300.0;
     const RECHECK_CONFIG_INTERVAL: Duration = Duration::from_secs(10);
@@ -348,7 +348,7 @@ pub(crate) fn log_compaction_error(
 }
 
 /// GC task's main loop.
-async fn gc_loop(tenant: Arc<TenantShard>, cancel: CancellationToken) {
+async fn gc_loop(tenant: Arc<Tenant>, cancel: CancellationToken) {
     const MAX_BACKOFF_SECS: f64 = 300.0;
     let mut error_run = 0; // consecutive errors
 
@@ -432,7 +432,7 @@ async fn gc_loop(tenant: Arc<TenantShard>, cancel: CancellationToken) {
 }
 
 /// Tenant housekeeping's main loop.
-async fn tenant_housekeeping_loop(tenant: Arc<TenantShard>, cancel: CancellationToken) {
+async fn tenant_housekeeping_loop(tenant: Arc<Tenant>, cancel: CancellationToken) {
     let mut last_throttle_flag_reset_at = Instant::now();
     loop {
         if wait_for_active_tenant(&tenant, &cancel).await.is_break() {
@@ -483,7 +483,7 @@ async fn tenant_housekeeping_loop(tenant: Arc<TenantShard>, cancel: Cancellation
 
 /// Waits until the tenant becomes active, or returns `ControlFlow::Break()` to shut down.
 async fn wait_for_active_tenant(
-    tenant: &Arc<TenantShard>,
+    tenant: &Arc<Tenant>,
     cancel: &CancellationToken,
 ) -> ControlFlow<()> {
     if tenant.current_state() == TenantState::Active {

pageserver/src/tenant/timeline.rs

@@ -412,7 +412,7 @@ pub struct Timeline {
     /// Timeline deletion will acquire both compaction and gc locks in whatever order.
     gc_lock: tokio::sync::Mutex<()>,
 
-    /// Cloned from [`super::TenantShard::pagestream_throttle`] on construction.
+    /// Cloned from [`super::Tenant::pagestream_throttle`] on construction.
     pub(crate) pagestream_throttle: Arc<crate::tenant::throttle::Throttle>,
 
     /// Size estimator for aux file v2
@@ -2065,7 +2065,7 @@ impl Timeline {
 
     pub(crate) fn activate(
         self: &Arc<Self>,
-        parent: Arc<crate::tenant::TenantShard>,
+        parent: Arc<crate::tenant::Tenant>,
         broker_client: BrokerClientChannel,
         background_jobs_can_start: Option<&completion::Barrier>,
         ctx: &RequestContext,
@@ -2490,12 +2490,11 @@ impl Timeline {
         tenant_conf.is_gc_blocked_by_lsn_lease_deadline()
     }
 
-    pub(crate) fn get_lazy_slru_download(&self) -> bool {
+    pub(crate) fn get_lazy_slru_download(&self, lazy_slru_download_enabled_by_cp: bool) -> bool {
         let tenant_conf = self.tenant_conf.load();
-        tenant_conf
-            .tenant_conf
-            .lazy_slru_download
-            .unwrap_or(self.conf.default_tenant_conf.lazy_slru_download)
+        tenant_conf.tenant_conf.lazy_slru_download.unwrap_or(
+            lazy_slru_download_enabled_by_cp || self.conf.default_tenant_conf.lazy_slru_download,
+        )
     }
 
     /// Checks if a get page request should get perf tracing
@@ -2702,14 +2701,6 @@ impl Timeline {
             .clone()
     }
 
-    pub fn get_compaction_shard_ancestor(&self) -> bool {
-        let tenant_conf = self.tenant_conf.load();
-        tenant_conf
-            .tenant_conf
-            .compaction_shard_ancestor
-            .unwrap_or(self.conf.default_tenant_conf.compaction_shard_ancestor)
-    }
-
     fn get_eviction_policy(&self) -> EvictionPolicy {
         let tenant_conf = self.tenant_conf.load();
         tenant_conf
@@ -3325,7 +3316,7 @@ impl Timeline {
         //     (1) and (4)
         // TODO: this is basically a no-op now, should we remove it?
         self.remote_client.schedule_barrier()?;
-        // TenantShard::create_timeline will wait for these uploads to happen before returning, or
+        // Tenant::create_timeline will wait for these uploads to happen before returning, or
         // on retry.
 
         // Now that we have the full layer map, we may calculate the visibility of layers within it (a global scan)
@@ -5754,7 +5745,7 @@ impl Timeline {
     /// from our ancestor to be branches of this timeline.
     pub(crate) async fn prepare_to_detach_from_ancestor(
         self: &Arc<Timeline>,
-        tenant: &crate::tenant::TenantShard,
+        tenant: &crate::tenant::Tenant,
         options: detach_ancestor::Options,
         behavior: DetachBehavior,
         ctx: &RequestContext,
@@ -5773,7 +5764,7 @@ impl Timeline {
     /// resetting the tenant.
     pub(crate) async fn detach_from_ancestor_and_reparent(
         self: &Arc<Timeline>,
-        tenant: &crate::tenant::TenantShard,
+        tenant: &crate::tenant::Tenant,
         prepared: detach_ancestor::PreparedTimelineDetach,
         ancestor_timeline_id: TimelineId,
         ancestor_lsn: Lsn,
@@ -5797,7 +5788,7 @@ impl Timeline {
     /// The tenant must've been reset if ancestry was modified previously (in tenant manager).
     pub(crate) async fn complete_detaching_timeline_ancestor(
         self: &Arc<Timeline>,
-        tenant: &crate::tenant::TenantShard,
+        tenant: &crate::tenant::Tenant,
         attempt: detach_ancestor::Attempt,
         ctx: &RequestContext,
     ) -> Result<(), detach_ancestor::Error> {
@@ -6859,14 +6850,14 @@ impl Timeline {
     /// Persistently blocks gc for `Manual` reason.
     ///
     /// Returns true if no such block existed before, false otherwise.
-    pub(crate) async fn block_gc(&self, tenant: &super::TenantShard) -> anyhow::Result<bool> {
+    pub(crate) async fn block_gc(&self, tenant: &super::Tenant) -> anyhow::Result<bool> {
         use crate::tenant::remote_timeline_client::index::GcBlockingReason;
         assert_eq!(self.tenant_shard_id, tenant.tenant_shard_id);
         tenant.gc_block.insert(self, GcBlockingReason::Manual).await
     }
 
     /// Persistently unblocks gc for `Manual` reason.
-    pub(crate) async fn unblock_gc(&self, tenant: &super::TenantShard) -> anyhow::Result<()> {
+    pub(crate) async fn unblock_gc(&self, tenant: &super::Tenant) -> anyhow::Result<()> {
         use crate::tenant::remote_timeline_client::index::GcBlockingReason;
         assert_eq!(self.tenant_shard_id, tenant.tenant_shard_id);
         tenant.gc_block.remove(self, GcBlockingReason::Manual).await
@@ -6884,8 +6875,8 @@ impl Timeline {
 
     /// Force create an image layer and place it into the layer map.
     ///
-    /// DO NOT use this function directly. Use [`TenantShard::branch_timeline_test_with_layers`]
-    /// or [`TenantShard::create_test_timeline_with_layers`] to ensure all these layers are
+    /// DO NOT use this function directly. Use [`Tenant::branch_timeline_test_with_layers`]
+    /// or [`Tenant::create_test_timeline_with_layers`] to ensure all these layers are
     /// placed into the layer map in one run AND be validated.
     #[cfg(test)]
     pub(super) async fn force_create_image_layer(
@@ -6941,8 +6932,8 @@ impl Timeline {
 
     /// Force create a delta layer and place it into the layer map.
     ///
-    /// DO NOT use this function directly. Use [`TenantShard::branch_timeline_test_with_layers`]
-    /// or [`TenantShard::create_test_timeline_with_layers`] to ensure all these layers are
+    /// DO NOT use this function directly. Use [`Tenant::branch_timeline_test_with_layers`]
+    /// or [`Tenant::create_test_timeline_with_layers`] to ensure all these layers are
     /// placed into the layer map in one run AND be validated.
     #[cfg(test)]
     pub(super) async fn force_create_delta_layer(

pageserver/src/tenant/timeline/compaction.rs

@@ -77,7 +77,7 @@ const COMPACTION_DELTA_THRESHOLD: usize = 5;
 /// shard split, which gets expensive for large tenants.
 const ANCESTOR_COMPACTION_REWRITE_THRESHOLD: f64 = 0.3;
 
-#[derive(Default, Debug, Clone, Copy, Hash, PartialEq, Eq, Serialize)]
+#[derive(Debug, Clone, Copy, Hash, PartialEq, Eq)]
 pub struct GcCompactionJobId(pub usize);
 
 impl std::fmt::Display for GcCompactionJobId {
@@ -105,50 +105,6 @@ pub enum GcCompactionQueueItem {
     Notify(GcCompactionJobId, Option<Lsn>),
 }
 
-/// Statistics for gc-compaction meta jobs, which contains several sub compaction jobs.
-#[derive(Debug, Clone, Serialize, Default)]
-pub struct GcCompactionMetaStatistics {
-    /// The total number of sub compaction jobs.
-    pub total_sub_compaction_jobs: usize,
-    /// The total number of sub compaction jobs that failed.
-    pub failed_sub_compaction_jobs: usize,
-    /// The total number of sub compaction jobs that succeeded.
-    pub succeeded_sub_compaction_jobs: usize,
-    /// The layer size before compaction.
-    pub before_compaction_layer_size: u64,
-    /// The layer size after compaction.
-    pub after_compaction_layer_size: u64,
-    /// The start time of the meta job.
-    pub start_time: Option<chrono::DateTime<chrono::Utc>>,
-    /// The end time of the meta job.
-    pub end_time: Option<chrono::DateTime<chrono::Utc>>,
-    /// The duration of the meta job.
-    pub duration_secs: f64,
-    /// The id of the meta job.
-    pub meta_job_id: GcCompactionJobId,
-    /// The LSN below which the layers are compacted, used to compute the statistics.
-    pub below_lsn: Lsn,
-    /// The retention ratio of the meta job (after_compaction_layer_size / before_compaction_layer_size)
-    pub retention_ratio: f64,
-}
-
-impl GcCompactionMetaStatistics {
-    fn finalize(&mut self) {
-        let end_time = chrono::Utc::now();
-        if let Some(start_time) = self.start_time {
-            if end_time > start_time {
-                let delta = end_time - start_time;
-                if let Ok(std_dur) = delta.to_std() {
-                    self.duration_secs = std_dur.as_secs_f64();
-                }
-            }
-        }
-        self.retention_ratio = self.after_compaction_layer_size as f64
-            / (self.before_compaction_layer_size as f64 + 1.0);
-        self.end_time = Some(end_time);
-    }
-}
-
 impl GcCompactionQueueItem {
     pub fn into_compact_info_resp(
         self,
@@ -186,7 +142,6 @@ struct GcCompactionQueueInner {
     queued: VecDeque<(GcCompactionJobId, GcCompactionQueueItem)>,
     guards: HashMap<GcCompactionJobId, GcCompactionGuardItems>,
     last_id: GcCompactionJobId,
-    meta_statistics: Option<GcCompactionMetaStatistics>,
 }
 
 impl GcCompactionQueueInner {
@@ -218,7 +173,6 @@ impl GcCompactionQueue {
                 queued: VecDeque::new(),
                 guards: HashMap::new(),
                 last_id: GcCompactionJobId(0),
-                meta_statistics: None,
             }),
             consumer_lock: tokio::sync::Mutex::new(()),
         }
@@ -403,23 +357,6 @@ impl GcCompactionQueue {
         Ok(())
     }
 
-    async fn collect_layer_below_lsn(
-        &self,
-        timeline: &Arc<Timeline>,
-        lsn: Lsn,
-    ) -> Result<u64, CompactionError> {
-        let guard = timeline.layers.read().await;
-        let layer_map = guard.layer_map()?;
-        let layers = layer_map.iter_historic_layers().collect_vec();
-        let mut size = 0;
-        for layer in layers {
-            if layer.lsn_range.start <= lsn {
-                size += layer.file_size();
-            }
-        }
-        Ok(size)
-    }
-
     /// Notify the caller the job has finished and unblock GC.
     fn notify_and_unblock(&self, id: GcCompactionJobId) {
         info!("compaction job id={} finished", id);
@@ -429,16 +366,6 @@ impl GcCompactionQueue {
                 let _ = tx.send(());
             }
         }
-        if let Some(ref meta_statistics) = guard.meta_statistics {
-            if meta_statistics.meta_job_id == id {
-                if let Ok(stats) = serde_json::to_string(&meta_statistics) {
-                    info!(
-                        "gc-compaction meta statistics for job id = {}: {}",
-                        id, stats
-                    );
-                }
-            }
-        }
     }
 
     fn clear_running_job(&self) {
@@ -478,11 +405,7 @@ impl GcCompactionQueue {
             let mut pending_tasks = Vec::new();
             // gc-compaction might pick more layers or fewer layers to compact. The L2 LSN does not need to be accurate.
             // And therefore, we simply assume the maximum LSN of all jobs is the expected L2 LSN.
-            let expected_l2_lsn = jobs
-                .iter()
-                .map(|job| job.compact_lsn_range.end)
-                .max()
-                .unwrap();
+            let expected_l2_lsn = jobs.iter().map(|job| job.compact_lsn_range.end).max();
             for job in jobs {
                 // Unfortunately we need to convert the `GcCompactJob` back to `CompactionOptions`
                 // until we do further refactors to allow directly call `compact_with_gc`.
@@ -507,13 +430,9 @@ impl GcCompactionQueue {
             if !auto {
                 pending_tasks.push(GcCompactionQueueItem::Notify(id, None));
             } else {
-                pending_tasks.push(GcCompactionQueueItem::Notify(id, Some(expected_l2_lsn)));
+                pending_tasks.push(GcCompactionQueueItem::Notify(id, expected_l2_lsn));
             }
 
-            let layer_size = self
-                .collect_layer_below_lsn(timeline, expected_l2_lsn)
-                .await?;
-
             {
                 let mut guard = self.inner.lock().unwrap();
                 let mut tasks = Vec::new();
@@ -525,16 +444,7 @@ impl GcCompactionQueue {
                 for item in tasks {
                     guard.queued.push_front(item);
                 }
-                guard.meta_statistics = Some(GcCompactionMetaStatistics {
-                    meta_job_id: id,
-                    start_time: Some(chrono::Utc::now()),
-                    before_compaction_layer_size: layer_size,
-                    below_lsn: expected_l2_lsn,
-                    total_sub_compaction_jobs: jobs_len,
-                    ..Default::default()
-                });
             }
-
             info!(
                 "scheduled enhanced gc bottom-most compaction with sub-compaction, split into {} jobs",
                 jobs_len
@@ -663,10 +573,6 @@ impl GcCompactionQueue {
                     Err(err) => {
                         warn!(%err, "failed to run gc-compaction subcompaction job");
                         self.clear_running_job();
-                        let mut guard = self.inner.lock().unwrap();
-                        if let Some(ref mut meta_statistics) = guard.meta_statistics {
-                            meta_statistics.failed_sub_compaction_jobs += 1;
-                        }
                         return Err(err);
                     }
                 };
@@ -676,34 +582,8 @@ impl GcCompactionQueue {
                     // we need to clean things up before returning from the function.
                     yield_for_l0 = true;
                 }
-                {
-                    let mut guard = self.inner.lock().unwrap();
-                    if let Some(ref mut meta_statistics) = guard.meta_statistics {
-                        meta_statistics.succeeded_sub_compaction_jobs += 1;
-                    }
-                }
             }
             GcCompactionQueueItem::Notify(id, l2_lsn) => {
-                let below_lsn = {
-                    let mut guard = self.inner.lock().unwrap();
-                    if let Some(ref mut meta_statistics) = guard.meta_statistics {
-                        meta_statistics.below_lsn
-                    } else {
-                        Lsn::INVALID
-                    }
-                };
-                let layer_size = if below_lsn != Lsn::INVALID {
-                    self.collect_layer_below_lsn(timeline, below_lsn).await?
-                } else {
-                    0
-                };
-                {
-                    let mut guard = self.inner.lock().unwrap();
-                    if let Some(ref mut meta_statistics) = guard.meta_statistics {
-                        meta_statistics.after_compaction_layer_size = layer_size;
-                        meta_statistics.finalize();
-                    }
-                }
                 self.notify_and_unblock(id);
                 if let Some(l2_lsn) = l2_lsn {
                     let current_l2_lsn = timeline
@@ -1359,7 +1239,8 @@ impl Timeline {
         let partition_count = self.partitioning.read().0.0.parts.len();
 
         // 4. Shard ancestor compaction
-        if self.get_compaction_shard_ancestor() && self.shard_identity.count >= ShardCount::new(2) {
+
+        if self.shard_identity.count >= ShardCount::new(2) {
             // Limit the number of layer rewrites to the number of partitions: this means its
             // runtime should be comparable to a full round of image layer creations, rather than
             // being potentially much longer.

pageserver/src/tenant/timeline/delete.rs

@@ -18,8 +18,8 @@ use crate::tenant::remote_timeline_client::{
     PersistIndexPartWithDeletedFlagError, RemoteTimelineClient,
 };
 use crate::tenant::{
-    CreateTimelineCause, DeleteTimelineError, MaybeDeletedIndexPart, TenantManifestError,
-    TenantShard, Timeline, TimelineOrOffloaded,
+    CreateTimelineCause, DeleteTimelineError, MaybeDeletedIndexPart, Tenant, TenantManifestError,
+    Timeline, TimelineOrOffloaded,
 };
 use crate::virtual_file::MaybeFatalIo;
 
@@ -113,7 +113,7 @@ pub(super) async fn delete_local_timeline_directory(
 /// It is important that this gets called when DeletionGuard is being held.
 /// For more context see comments in [`make_timeline_delete_guard`]
 async fn remove_maybe_offloaded_timeline_from_tenant(
-    tenant: &TenantShard,
+    tenant: &Tenant,
     timeline: &TimelineOrOffloaded,
     _: &DeletionGuard, // using it as a witness
 ) -> anyhow::Result<()> {
@@ -192,7 +192,7 @@ impl DeleteTimelineFlow {
     // error out if some of the shutdown tasks have already been completed!
     #[instrument(skip_all)]
     pub async fn run(
-        tenant: &Arc<TenantShard>,
+        tenant: &Arc<Tenant>,
         timeline_id: TimelineId,
     ) -> Result<(), DeleteTimelineError> {
         super::debug_assert_current_span_has_tenant_and_timeline_id();
@@ -288,7 +288,7 @@ impl DeleteTimelineFlow {
     /// Shortcut to create Timeline in stopping state and spawn deletion task.
     #[instrument(skip_all, fields(%timeline_id))]
     pub(crate) async fn resume_deletion(
-        tenant: Arc<TenantShard>,
+        tenant: Arc<Tenant>,
         timeline_id: TimelineId,
         local_metadata: &TimelineMetadata,
         remote_client: RemoteTimelineClient,
@@ -338,7 +338,7 @@ impl DeleteTimelineFlow {
     fn schedule_background(
         guard: DeletionGuard,
         conf: &'static PageServerConf,
-        tenant: Arc<TenantShard>,
+        tenant: Arc<Tenant>,
         timeline: TimelineOrOffloaded,
         remote_client: Arc<RemoteTimelineClient>,
     ) {
@@ -381,7 +381,7 @@ impl DeleteTimelineFlow {
     async fn background(
         mut guard: DeletionGuard,
         conf: &PageServerConf,
-        tenant: &TenantShard,
+        tenant: &Tenant,
         timeline: &TimelineOrOffloaded,
         remote_client: Arc<RemoteTimelineClient>,
     ) -> Result<(), DeleteTimelineError> {
@@ -435,7 +435,7 @@ pub(super) enum TimelineDeleteGuardKind {
 }
 
 pub(super) fn make_timeline_delete_guard(
-    tenant: &TenantShard,
+    tenant: &Tenant,
     timeline_id: TimelineId,
     guard_kind: TimelineDeleteGuardKind,
 ) -> Result<(TimelineOrOffloaded, DeletionGuard), DeleteTimelineError> {

pageserver/src/tenant/timeline/detach_ancestor.rs

@@ -23,7 +23,7 @@ use super::layer_manager::LayerManager;
 use super::{FlushLayerError, Timeline};
 use crate::context::{DownloadBehavior, RequestContext};
 use crate::task_mgr::TaskKind;
-use crate::tenant::TenantShard;
+use crate::tenant::Tenant;
 use crate::tenant::remote_timeline_client::index::GcBlockingReason::DetachAncestor;
 use crate::tenant::storage_layer::layer::local_layer_path;
 use crate::tenant::storage_layer::{
@@ -265,7 +265,7 @@ async fn generate_tombstone_image_layer(
 /// See [`Timeline::prepare_to_detach_from_ancestor`]
 pub(super) async fn prepare(
     detached: &Arc<Timeline>,
-    tenant: &TenantShard,
+    tenant: &Tenant,
     behavior: DetachBehavior,
     options: Options,
     ctx: &RequestContext,
@@ -590,7 +590,7 @@ pub(super) async fn prepare(
 
 async fn start_new_attempt(
     detached: &Timeline,
-    tenant: &TenantShard,
+    tenant: &Tenant,
     ancestor_timeline_id: TimelineId,
     ancestor_lsn: Lsn,
 ) -> Result<Attempt, Error> {
@@ -611,7 +611,7 @@ async fn start_new_attempt(
 
 async fn continue_with_blocked_gc(
     detached: &Timeline,
-    tenant: &TenantShard,
+    tenant: &Tenant,
     ancestor_timeline_id: TimelineId,
     ancestor_lsn: Lsn,
 ) -> Result<Attempt, Error> {
@@ -622,7 +622,7 @@ async fn continue_with_blocked_gc(
 
 fn obtain_exclusive_attempt(
     detached: &Timeline,
-    tenant: &TenantShard,
+    tenant: &Tenant,
     ancestor_timeline_id: TimelineId,
     ancestor_lsn: Lsn,
 ) -> Result<Attempt, Error> {
@@ -655,7 +655,7 @@ fn obtain_exclusive_attempt(
 
 fn reparented_direct_children(
     detached: &Arc<Timeline>,
-    tenant: &TenantShard,
+    tenant: &Tenant,
 ) -> Result<HashSet<TimelineId>, Error> {
     let mut all_direct_children = tenant
         .timelines
@@ -950,7 +950,7 @@ impl DetachingAndReparenting {
 /// See [`Timeline::detach_from_ancestor_and_reparent`].
 pub(super) async fn detach_and_reparent(
     detached: &Arc<Timeline>,
-    tenant: &TenantShard,
+    tenant: &Tenant,
     prepared: PreparedTimelineDetach,
     ancestor_timeline_id: TimelineId,
     ancestor_lsn: Lsn,
@@ -1184,7 +1184,7 @@ pub(super) async fn detach_and_reparent(
 
 pub(super) async fn complete(
     detached: &Arc<Timeline>,
-    tenant: &TenantShard,
+    tenant: &Tenant,
     mut attempt: Attempt,
     _ctx: &RequestContext,
 ) -> Result<(), Error> {
@@ -1258,7 +1258,7 @@ where
 }
 
 fn check_no_archived_children_of_ancestor(
-    tenant: &TenantShard,
+    tenant: &Tenant,
     detached: &Arc<Timeline>,
     ancestor: &Arc<Timeline>,
     ancestor_lsn: Lsn,

pageserver/src/tenant/timeline/eviction_task.rs

@@ -33,7 +33,7 @@ use crate::tenant::size::CalculateSyntheticSizeError;
 use crate::tenant::storage_layer::LayerVisibilityHint;
 use crate::tenant::tasks::{BackgroundLoopKind, BackgroundLoopSemaphorePermit, sleep_random};
 use crate::tenant::timeline::EvictionError;
-use crate::tenant::{LogicalSizeCalculationCause, TenantShard};
+use crate::tenant::{LogicalSizeCalculationCause, Tenant};
 
 #[derive(Default)]
 pub struct EvictionTaskTimelineState {
@@ -48,7 +48,7 @@ pub struct EvictionTaskTenantState {
 impl Timeline {
     pub(super) fn launch_eviction_task(
         self: &Arc<Self>,
-        parent: Arc<TenantShard>,
+        parent: Arc<Tenant>,
         background_tasks_can_start: Option<&completion::Barrier>,
     ) {
         let self_clone = Arc::clone(self);
@@ -75,7 +75,7 @@ impl Timeline {
     }
 
     #[instrument(skip_all, fields(tenant_id = %self.tenant_shard_id.tenant_id, shard_id = %self.tenant_shard_id.shard_slug(), timeline_id = %self.timeline_id))]
-    async fn eviction_task(self: Arc<Self>, tenant: Arc<TenantShard>) {
+    async fn eviction_task(self: Arc<Self>, tenant: Arc<Tenant>) {
         // acquire the gate guard only once within a useful span
         let Ok(guard) = self.gate.enter() else {
             return;
@@ -118,7 +118,7 @@ impl Timeline {
     #[instrument(skip_all, fields(policy_kind = policy.discriminant_str()))]
     async fn eviction_iteration(
         self: &Arc<Self>,
-        tenant: &TenantShard,
+        tenant: &Tenant,
         policy: &EvictionPolicy,
         cancel: &CancellationToken,
         gate: &GateGuard,
@@ -175,7 +175,7 @@ impl Timeline {
 
     async fn eviction_iteration_threshold(
         self: &Arc<Self>,
-        tenant: &TenantShard,
+        tenant: &Tenant,
         p: &EvictionPolicyLayerAccessThreshold,
         cancel: &CancellationToken,
         gate: &GateGuard,
@@ -309,7 +309,7 @@ impl Timeline {
     /// disk usage based eviction task.
     async fn imitiate_only(
         self: &Arc<Self>,
-        tenant: &TenantShard,
+        tenant: &Tenant,
         p: &EvictionPolicyLayerAccessThreshold,
         cancel: &CancellationToken,
         gate: &GateGuard,
@@ -363,7 +363,7 @@ impl Timeline {
     #[instrument(skip_all)]
     async fn imitate_layer_accesses(
         &self,
-        tenant: &TenantShard,
+        tenant: &Tenant,
         p: &EvictionPolicyLayerAccessThreshold,
         cancel: &CancellationToken,
         gate: &GateGuard,
@@ -499,7 +499,7 @@ impl Timeline {
     #[instrument(skip_all)]
     async fn imitate_synthetic_size_calculation_worker(
         &self,
-        tenant: &TenantShard,
+        tenant: &Tenant,
         cancel: &CancellationToken,
         ctx: &RequestContext,
     ) {

pageserver/src/tenant/timeline/import_pgdata/upcall_api.rs

@@ -1,6 +1,6 @@
 //! FIXME: most of this is copy-paste from mgmt_api.rs ; dedupe into a `reqwest_utils::Client` crate.
 use pageserver_client::mgmt_api::{Error, ResponseErrorMessageExt};
-use reqwest::{Certificate, Method};
+use reqwest::Method;
 use serde::{Deserialize, Serialize};
 use tokio_util::sync::CancellationToken;
 use tracing::error;
@@ -34,7 +34,7 @@ impl Client {
         };
         let mut http_client = reqwest::Client::builder();
         for cert in &conf.ssl_ca_certs {
-            http_client = http_client.add_root_certificate(Certificate::from_der(cert.contents())?);
+            http_client = http_client.add_root_certificate(cert.clone());
         }
         let http_client = http_client.build()?;
 

pageserver/src/tenant/timeline/offload.rs

@@ -8,7 +8,7 @@ use crate::span::debug_assert_current_span_has_tenant_and_timeline_id;
 use crate::tenant::remote_timeline_client::ShutdownIfArchivedError;
 use crate::tenant::timeline::delete::{TimelineDeleteGuardKind, make_timeline_delete_guard};
 use crate::tenant::{
-    DeleteTimelineError, OffloadedTimeline, TenantManifestError, TenantShard, TimelineOrOffloaded,
+    DeleteTimelineError, OffloadedTimeline, Tenant, TenantManifestError, TimelineOrOffloaded,
 };
 
 #[derive(thiserror::Error, Debug)]
@@ -33,7 +33,7 @@ impl From<TenantManifestError> for OffloadError {
 }
 
 pub(crate) async fn offload_timeline(
-    tenant: &TenantShard,
+    tenant: &Tenant,
     timeline: &Arc<Timeline>,
 ) -> Result<(), OffloadError> {
     debug_assert_current_span_has_tenant_and_timeline_id();
@@ -123,7 +123,7 @@ pub(crate) async fn offload_timeline(
 ///
 /// Returns the strong count of the timeline `Arc`
 fn remove_timeline_from_tenant(
-    tenant: &TenantShard,
+    tenant: &Tenant,
     timeline: &Timeline,
     _: &DeletionGuard, // using it as a witness
 ) -> usize {

pageserver/src/tenant/timeline/uninit.rs

@@ -15,19 +15,17 @@ use super::Timeline;
 use crate::context::RequestContext;
 use crate::import_datadir;
 use crate::span::debug_assert_current_span_has_tenant_and_timeline_id;
-use crate::tenant::{
-    CreateTimelineError, CreateTimelineIdempotency, TenantShard, TimelineOrOffloaded,
-};
+use crate::tenant::{CreateTimelineError, CreateTimelineIdempotency, Tenant, TimelineOrOffloaded};
 
 /// A timeline with some of its files on disk, being initialized.
 /// This struct ensures the atomicity of the timeline init: it's either properly created and inserted into pageserver's memory, or
 /// its local files are removed.  If we crash while this class exists, then the timeline's local
-/// state is cleaned up during [`TenantShard::clean_up_timelines`], because the timeline's content isn't in remote storage.
+/// state is cleaned up during [`Tenant::clean_up_timelines`], because the timeline's content isn't in remote storage.
 ///
 /// The caller is responsible for proper timeline data filling before the final init.
 #[must_use]
 pub struct UninitializedTimeline<'t> {
-    pub(crate) owning_tenant: &'t TenantShard,
+    pub(crate) owning_tenant: &'t Tenant,
     timeline_id: TimelineId,
     raw_timeline: Option<(Arc<Timeline>, TimelineCreateGuard)>,
     /// Whether we spawned the inner Timeline's tasks such that we must later shut it down
@@ -37,7 +35,7 @@ pub struct UninitializedTimeline<'t> {
 
 impl<'t> UninitializedTimeline<'t> {
     pub(crate) fn new(
-        owning_tenant: &'t TenantShard,
+        owning_tenant: &'t Tenant,
         timeline_id: TimelineId,
         raw_timeline: Option<(Arc<Timeline>, TimelineCreateGuard)>,
     ) -> Self {
@@ -158,7 +156,7 @@ impl<'t> UninitializedTimeline<'t> {
     /// Prepares timeline data by loading it from the basebackup archive.
     pub(crate) async fn import_basebackup_from_tar(
         mut self,
-        tenant: Arc<TenantShard>,
+        tenant: Arc<Tenant>,
         copyin_read: &mut (impl tokio::io::AsyncRead + Send + Sync + Unpin),
         base_lsn: Lsn,
         broker_client: storage_broker::BrokerClientChannel,
@@ -229,17 +227,17 @@ pub(crate) fn cleanup_timeline_directory(create_guard: TimelineCreateGuard) {
             error!("Failed to clean up uninitialized timeline directory {timeline_path:?}: {e:?}")
         }
     }
-    // Having cleaned up, we can release this TimelineId in `[TenantShard::timelines_creating]` to allow other
+    // Having cleaned up, we can release this TimelineId in `[Tenant::timelines_creating]` to allow other
     // timeline creation attempts under this TimelineId to proceed
     drop(create_guard);
 }
 
 /// A guard for timeline creations in process: as long as this object exists, the timeline ID
-/// is kept in `[TenantShard::timelines_creating]` to exclude concurrent attempts to create the same timeline.
+/// is kept in `[Tenant::timelines_creating]` to exclude concurrent attempts to create the same timeline.
 #[must_use]
 pub(crate) struct TimelineCreateGuard {
     pub(crate) _tenant_gate_guard: GateGuard,
-    pub(crate) owning_tenant: Arc<TenantShard>,
+    pub(crate) owning_tenant: Arc<Tenant>,
     pub(crate) timeline_id: TimelineId,
     pub(crate) timeline_path: Utf8PathBuf,
     pub(crate) idempotency: CreateTimelineIdempotency,
@@ -265,7 +263,7 @@ pub(crate) enum TimelineExclusionError {
 
 impl TimelineCreateGuard {
     pub(crate) fn new(
-        owning_tenant: &Arc<TenantShard>,
+        owning_tenant: &Arc<Tenant>,
         timeline_id: TimelineId,
         timeline_path: Utf8PathBuf,
         idempotency: CreateTimelineIdempotency,

pageserver/src/tenant/vectored_blob_io.rs

@@ -26,7 +26,7 @@ use utils::lsn::Lsn;
 use utils::vec_map::VecMap;
 
 use crate::context::RequestContext;
-use crate::tenant::blob_io::{BYTE_UNCOMPRESSED, BYTE_ZSTD, Header};
+use crate::tenant::blob_io::{BYTE_UNCOMPRESSED, BYTE_ZSTD, LEN_COMPRESSION_BIT_MASK};
 use crate::virtual_file::{self, IoBufferMut, VirtualFile};
 
 /// Metadata bundled with the start and end offset of a blob.
@@ -111,20 +111,18 @@ impl From<Bytes> for BufView<'_> {
 pub struct VectoredBlob {
     /// Blob metadata.
     pub meta: BlobMeta,
-    /// Header start offset.
-    header_start: usize,
-    /// Data start offset.
-    data_start: usize,
+    /// Start offset.
+    start: usize,
     /// End offset.
     end: usize,
-    /// Compression used on the data, extracted from the header.
+    /// Compression used on the the blob.
     compression_bits: u8,
 }
 
 impl VectoredBlob {
     /// Reads a decompressed view of the blob.
     pub(crate) async fn read<'a>(&self, buf: &BufView<'a>) -> Result<BufView<'a>, std::io::Error> {
-        let view = buf.view(self.data_start..self.end);
+        let view = buf.view(self.start..self.end);
 
         match self.compression_bits {
             BYTE_UNCOMPRESSED => Ok(view),
@@ -142,18 +140,13 @@ impl VectoredBlob {
                     std::io::ErrorKind::InvalidData,
                     format!(
                         "Failed to decompress blob for {}@{}, {}..{}: invalid compression byte {bits:x}",
-                        self.meta.key, self.meta.lsn, self.data_start, self.end
+                        self.meta.key, self.meta.lsn, self.start, self.end
                     ),
                 );
                 Err(error)
             }
         }
     }
-
-    /// Returns the raw blob including header.
-    pub(crate) fn raw_with_header<'a>(&self, buf: &BufView<'a>) -> BufView<'a> {
-        buf.view(self.header_start..self.end)
-    }
 }
 
 impl std::fmt::Display for VectoredBlob {
@@ -161,7 +154,7 @@ impl std::fmt::Display for VectoredBlob {
         write!(
             f,
             "{}@{}, {}..{}",
-            self.meta.key, self.meta.lsn, self.data_start, self.end
+            self.meta.key, self.meta.lsn, self.start, self.end
         )
     }
 }
@@ -500,28 +493,50 @@ impl<'a> VectoredBlobReader<'a> {
 
         let blobs_at = read.blobs_at.as_slice();
 
-        let mut blobs = Vec::with_capacity(blobs_at.len());
+        let start_offset = read.start;
+
+        let mut metas = Vec::with_capacity(blobs_at.len());
         // Blobs in `read` only provide their starting offset. The end offset
         // of a blob is implicit: the start of the next blob if one exists
         // or the end of the read.
 
-        for (blob_start, meta) in blobs_at.iter().copied() {
-            let header_start = (blob_start - read.start) as usize;
-            let header = Header::decode(&buf[header_start..])?;
-            let data_start = header_start + header.header_len;
-            let end = data_start + header.data_len;
-            let compression_bits = header.compression_bits;
+        for (blob_start, meta) in blobs_at {
+            let blob_start_in_buf = blob_start - start_offset;
+            let first_len_byte = buf[blob_start_in_buf as usize];
 
-            blobs.push(VectoredBlob {
-                header_start,
-                data_start,
+            // Each blob is prefixed by a header containing its size and compression information.
+            // Extract the size and skip that header to find the start of the data.
+            // The size can be 1 or 4 bytes. The most significant bit is 0 in the
+            // 1 byte case and 1 in the 4 byte case.
+            let (size_length, blob_size, compression_bits) = if first_len_byte < 0x80 {
+                (1, first_len_byte as u64, BYTE_UNCOMPRESSED)
+            } else {
+                let mut blob_size_buf = [0u8; 4];
+                let offset_in_buf = blob_start_in_buf as usize;
+
+                blob_size_buf.copy_from_slice(&buf[offset_in_buf..offset_in_buf + 4]);
+                blob_size_buf[0] &= !LEN_COMPRESSION_BIT_MASK;
+
+                let compression_bits = first_len_byte & LEN_COMPRESSION_BIT_MASK;
+                (
+                    4,
+                    u32::from_be_bytes(blob_size_buf) as u64,
+                    compression_bits,
+                )
+            };
+
+            let start = (blob_start_in_buf + size_length) as usize;
+            let end = start + blob_size as usize;
+
+            metas.push(VectoredBlob {
+                start,
                 end,
-                meta,
+                meta: *meta,
                 compression_bits,
             });
         }
 
-        Ok(VectoredBlobsBuf { buf, blobs })
+        Ok(VectoredBlobsBuf { buf, blobs: metas })
     }
 }
 
@@ -982,15 +997,6 @@ mod tests {
                 &read_buf[..],
                 "mismatch for idx={idx} at offset={offset}"
             );
-
-            // Check that raw_with_header returns a valid header.
-            let raw = read_blob.raw_with_header(&view);
-            let header = Header::decode(&raw)?;
-            if !compression || header.header_len == 1 {
-                assert_eq!(header.compression_bits, BYTE_UNCOMPRESSED);
-            }
-            assert_eq!(raw.len(), header.total_len());
-
             buf = result.buf;
         }
         Ok(())

pageserver/src/virtual_file.rs

@@ -1366,8 +1366,7 @@ pub(crate) type IoBuffer = AlignedBuffer<ConstAlign<{ get_io_buffer_alignment()
 pub(crate) type IoPageSlice<'a> =
     AlignedSlice<'a, PAGE_SZ, ConstAlign<{ get_io_buffer_alignment() }>>;
 
-static IO_MODE: once_cell::sync::Lazy<AtomicU8> =
-    once_cell::sync::Lazy::new(|| AtomicU8::new(IoMode::preferred() as u8));
+static IO_MODE: AtomicU8 = AtomicU8::new(IoMode::preferred() as u8);
 
 pub(crate) fn set_io_mode(mode: IoMode) {
     IO_MODE.store(mode as u8, std::sync::atomic::Ordering::Relaxed);

pgxn/neon/communicator.c

@@ -95,7 +95,7 @@ static uint32 local_request_counter;
  * Various settings related to prompt (fast) handling of PageStream responses
  * at any CHECK_FOR_INTERRUPTS point.
  */
-int				readahead_getpage_pull_timeout_ms = 50;
+int				readahead_getpage_pull_timeout_ms = 0;
 static int		PS_TIMEOUT_ID = 0;
 static bool		timeout_set = false;
 static bool		timeout_signaled = false;

pgxn/neon/libpagestore.c

@@ -75,7 +75,7 @@ char	   *neon_auth_token;
 int			readahead_buffer_size = 128;
 int			flush_every_n_requests = 8;
 
-int         neon_protocol_version = 3;
+int         neon_protocol_version = 2;
 
 static int	neon_compute_mode = 0;
 static int	max_reconnect_attempts = 60;
@@ -1362,7 +1362,7 @@ pg_init_libpagestore(void)
 							   "",
 							   PGC_POSTMASTER,
 							   0,	/* no flags required */
-							   NULL, NULL, NULL);
+							   check_neon_id, NULL, NULL);
 	DefineCustomStringVariable("neon.branch_id",
 							   "Neon branch_id the server is running on",
 							   NULL,
@@ -1370,7 +1370,7 @@ pg_init_libpagestore(void)
 							   "",
 							   PGC_POSTMASTER,
 							   0,	/* no flags required */
-							   NULL, NULL, NULL);
+							   check_neon_id, NULL, NULL);
 	DefineCustomStringVariable("neon.endpoint_id",
 							   "Neon endpoint_id the server is running on",
 							   NULL,
@@ -1378,7 +1378,7 @@ pg_init_libpagestore(void)
 							   "",
 							   PGC_POSTMASTER,
 							   0,	/* no flags required */
-							   NULL, NULL, NULL);
+							   check_neon_id, NULL, NULL);
 
 	DefineCustomIntVariable("neon.stripe_size",
 							"sharding stripe size",
@@ -1432,7 +1432,7 @@ pg_init_libpagestore(void)
 							"PageStream connection when we have pages which "
 							"were read ahead but not yet received.",
 							&readahead_getpage_pull_timeout_ms,
-							50, 0, 5 * 60 * 1000,
+							0, 0, 5 * 60 * 1000,
 							PGC_USERSET,
 							GUC_UNIT_MS,
 							NULL, NULL, NULL);
@@ -1440,7 +1440,7 @@ pg_init_libpagestore(void)
 							"Version of compute<->page server protocol",
 							NULL,
 							&neon_protocol_version,
-							3,	/* use protocol version 3 */
+							2,	/* use protocol version 2 */
 							2,	/* min */
 							3,	/* max */
 							PGC_SU_BACKEND,

pgxn/neon/pagestore_smgr.c

@@ -2040,7 +2040,7 @@ neon_finish_unlogged_build_phase_1(SMgrRelation reln)
 /*
  * neon_end_unlogged_build() -- Finish an unlogged rel build.
  *
- * Call this after you have finished WAL-logging a relation that was
+ * Call this after you have finished WAL-logging an relation that was
  * first populated without WAL-logging.
  *
  * This removes the local copy of the rel, since it's now been fully
@@ -2059,35 +2059,14 @@ neon_end_unlogged_build(SMgrRelation reln)
 
 	if (unlogged_build_phase != UNLOGGED_BUILD_NOT_PERMANENT)
 	{
-		XLogRecPtr recptr;
-		BlockNumber nblocks;
-
 		Assert(unlogged_build_phase == UNLOGGED_BUILD_PHASE_2);
 		Assert(reln->smgr_relpersistence == RELPERSISTENCE_UNLOGGED);
 
-		/*
-		 * Update the last-written LSN cache.
-		 *
-		 * The relation is still on local disk so we can get the size by
-		 * calling mdnblocks() directly. For the LSN, GetXLogInsertRecPtr() is
-		 * very conservative. If we could assume that this function is called
-		 * from the same backend that WAL-logged the contents, we could use
-		 * XactLastRecEnd here. But better safe than sorry.
-		 */
-		nblocks = mdnblocks(reln, MAIN_FORKNUM);
-		recptr = GetXLogInsertRecPtr();
-
-		neon_set_lwlsn_block_range(recptr,
-								   InfoFromNInfoB(rinfob),
-								   MAIN_FORKNUM, 0, nblocks);
-		neon_set_lwlsn_relation(recptr,
-								InfoFromNInfoB(rinfob),
-								MAIN_FORKNUM);
-
 		/* Make the relation look permanent again */
 		reln->smgr_relpersistence = RELPERSISTENCE_PERMANENT;
 
 		/* Remove local copy */
+		rinfob = InfoBFromSMgrRel(reln);
 		for (int forknum = 0; forknum <= MAX_FORKNUM; forknum++)
 		{
 			neon_log(SmgrTrace, "forgetting cached relsize for %u/%u/%u.%u",

pgxn/neon/walproposer_pg.c

@@ -890,7 +890,7 @@ libpqwp_connect_start(char *conninfo)
 	 * palloc will exit on failure though, so there's not much we could do if
 	 * it *did* fail.
 	 */
-	conn = (WalProposerConn*)MemoryContextAllocZero(TopMemoryContext, sizeof(WalProposerConn));
+	conn = palloc(sizeof(WalProposerConn));
 	conn->pg_conn = pg_conn;
 	conn->is_nonblocking = false;	/* connections always start in blocking
 									 * mode */

proxy/src/auth/backend/jwt.rs

@@ -776,6 +776,7 @@ impl From<&jose_jwk::Key> for KeyType {
 }
 
 #[cfg(test)]
+#[expect(clippy::unwrap_used)]
 mod tests {
     use std::future::IntoFuture;
     use std::net::SocketAddr;

proxy/src/auth/credentials.rs

@@ -253,6 +253,7 @@ fn project_name_valid(name: &str) -> bool {
 }
 
 #[cfg(test)]
+#[expect(clippy::unwrap_used)]
 mod tests {
     use ComputeUserInfoParseError::*;
     use serde_json::json;

proxy/src/binary/pg_sni_router.rs

@@ -258,7 +258,7 @@ async fn ssl_handshake<S: AsyncRead + AsyncWrite + Unpin>(
                 "unexpected startup packet, rejecting connection"
             );
             stream
-                .throw_error_str(ERR_INSECURE_CONNECTION, crate::error::ErrorKind::User, None)
+                .throw_error_str(ERR_INSECURE_CONNECTION, crate::error::ErrorKind::User)
                 .await?
         }
     }

proxy/src/cache/endpoints.rs

@@ -259,6 +259,7 @@ impl EndpointsCache {
 }
 
 #[cfg(test)]
+#[expect(clippy::unwrap_used)]
 mod tests {
     use super::*;
 

proxy/src/cache/project_info.rs

@@ -585,6 +585,7 @@ impl Cache for ProjectInfoCacheImpl {
 }
 
 #[cfg(test)]
+#[expect(clippy::unwrap_used)]
 mod tests {
     use super::*;
     use crate::scram::ServerSecret;

proxy/src/console_redirect_proxy.rs

@@ -222,7 +222,7 @@ pub(crate) async fn handle_client<S: AsyncRead + AsyncWrite + Unpin>(
     {
         Ok(auth_result) => auth_result,
         Err(e) => {
-            return stream.throw_error(e, Some(ctx)).await?;
+            return stream.throw_error(e).await?;
         }
     };
 
@@ -238,7 +238,7 @@ pub(crate) async fn handle_client<S: AsyncRead + AsyncWrite + Unpin>(
         config.wake_compute_retry_config,
         &config.connect_to_compute,
     )
-    .or_else(|e| stream.throw_error(e, Some(ctx)))
+    .or_else(|e| stream.throw_error(e))
     .await?;
 
     let cancellation_handler_clone = Arc::clone(&cancellation_handler);

proxy/src/context/mod.rs

@@ -63,7 +63,7 @@ struct RequestContextInner {
     success: bool,
     pub(crate) cold_start_info: ColdStartInfo,
     pg_options: Option<StartupMessageParams>,
-    testodrome_query_id: Option<SmolStr>,
+    testodrome_query_id: Option<String>,
 
     // extra
     // This sender is here to keep the request monitoring channel open while requests are taking place.
@@ -219,7 +219,7 @@ impl RequestContext {
             for option in options_str.split_whitespace() {
                 if option.starts_with("neon_query_id:") {
                     if let Some(value) = option.strip_prefix("neon_query_id:") {
-                        this.set_testodrome_id(value.into());
+                        this.set_testodrome_id(value.to_string());
                         break;
                     }
                 }
@@ -272,7 +272,7 @@ impl RequestContext {
             .set_user_agent(user_agent);
     }
 
-    pub(crate) fn set_testodrome_id(&self, query_id: SmolStr) {
+    pub(crate) fn set_testodrome_id(&self, query_id: String) {
         self.0
             .try_lock()
             .expect("should not deadlock")
@@ -378,7 +378,7 @@ impl RequestContext {
             .accumulated()
     }
 
-    pub(crate) fn get_testodrome_id(&self) -> Option<SmolStr> {
+    pub(crate) fn get_testodrome_id(&self) -> Option<String> {
         self.0
             .try_lock()
             .expect("should not deadlock")
@@ -447,7 +447,7 @@ impl RequestContextInner {
         self.user = Some(user);
     }
 
-    fn set_testodrome_id(&mut self, query_id: SmolStr) {
+    fn set_testodrome_id(&mut self, query_id: String) {
         self.testodrome_query_id = Some(query_id);
     }
 

proxy/src/context/parquet.rs

@@ -416,6 +416,7 @@ async fn upload_parquet(
 }
 
 #[cfg(test)]
+#[expect(clippy::unwrap_used)]
 mod tests {
     use std::net::Ipv4Addr;
     use std::num::NonZeroUsize;

proxy/src/intern.rs

@@ -227,6 +227,7 @@ impl From<AccountId> for AccountIdInt {
 }
 
 #[cfg(test)]
+#[expect(clippy::unwrap_used)]
 mod tests {
     use std::sync::OnceLock;
 

proxy/src/logging.rs

@@ -1032,6 +1032,7 @@ impl<const F: usize> serde::ser::Serialize for ExtractedSpanFields<'_, F> {
 }
 
 #[cfg(test)]
+#[allow(clippy::unwrap_used)]
 mod tests {
     use std::marker::PhantomData;
     use std::sync::{Arc, Mutex, MutexGuard};

proxy/src/protocol2.rs

@@ -400,6 +400,7 @@ impl NetworkEndianIpv6 {
 }
 
 #[cfg(test)]
+#[expect(clippy::unwrap_used)]
 mod tests {
     use tokio::io::AsyncReadExt;
 

proxy/src/proxy/copy_bidirectional.rs

@@ -262,6 +262,7 @@ impl CopyBuffer {
 }
 
 #[cfg(test)]
+#[expect(clippy::unwrap_used)]
 mod tests {
     use tokio::io::AsyncWriteExt;
 

proxy/src/proxy/handshake.rs

@@ -196,11 +196,7 @@ pub(crate) async fn handshake<S: AsyncRead + AsyncWrite + Unpin>(
                 // OR we didn't provide it at all (for dev purposes).
                 if tls.is_some() {
                     return stream
-                        .throw_error_str(
-                            ERR_INSECURE_CONNECTION,
-                            crate::error::ErrorKind::User,
-                            None,
-                        )
+                        .throw_error_str(ERR_INSECURE_CONNECTION, crate::error::ErrorKind::User)
                         .await?;
                 }
 

proxy/src/proxy/mod.rs

@@ -329,7 +329,7 @@ pub(crate) async fn handle_client<S: AsyncRead + AsyncWrite + Unpin>(
 
     let user_info = match result {
         Ok(user_info) => user_info,
-        Err(e) => stream.throw_error(e, Some(ctx)).await?,
+        Err(e) => stream.throw_error(e).await?,
     };
 
     let user = user_info.get_user().to_owned();
@@ -349,10 +349,7 @@ pub(crate) async fn handle_client<S: AsyncRead + AsyncWrite + Unpin>(
             let app = params.get("application_name");
             let params_span = tracing::info_span!("", ?user, ?db, ?app);
 
-            return stream
-                .throw_error(e, Some(ctx))
-                .instrument(params_span)
-                .await?;
+            return stream.throw_error(e).instrument(params_span).await?;
         }
     };
 
@@ -377,7 +374,7 @@ pub(crate) async fn handle_client<S: AsyncRead + AsyncWrite + Unpin>(
         config.wake_compute_retry_config,
         &config.connect_to_compute,
     )
-    .or_else(|e| stream.throw_error(e, Some(ctx)))
+    .or_else(|e| stream.throw_error(e))
     .await?;
 
     let cancellation_handler_clone = Arc::clone(&cancellation_handler);

proxy/src/proxy/tests/mod.rs

@@ -1,5 +1,5 @@
 //! A group of high-level tests for connection establishing logic and auth.
-#![allow(clippy::unimplemented)]
+#![allow(clippy::unimplemented, clippy::unwrap_used)]
 
 mod mitm;
 

proxy/src/rate_limiter/leaky_bucket.rs

@@ -83,7 +83,7 @@ impl From<LeakyBucketConfig> for utils::leaky_bucket::LeakyBucketConfig {
 }
 
 #[cfg(test)]
-#[allow(clippy::float_cmp)]
+#[allow(clippy::float_cmp, clippy::unwrap_used)]
 mod tests {
     use std::time::Duration;
 

proxy/src/rate_limiter/limit_algorithm/aimd.rs

@@ -63,6 +63,7 @@ impl LimitAlgorithm for Aimd {
 }
 
 #[cfg(test)]
+#[expect(clippy::unwrap_used)]
 mod tests {
     use std::time::Duration;
 

proxy/src/rate_limiter/limiter.rs

@@ -259,6 +259,7 @@ impl<K: Hash + Eq, R: Rng, S: BuildHasher + Clone> BucketRateLimiter<K, R, S> {
 }
 
 #[cfg(test)]
+#[expect(clippy::unwrap_used)]
 mod tests {
     use std::hash::BuildHasherDefault;
     use std::time::Duration;

proxy/src/sasl/messages.rs

@@ -51,6 +51,7 @@ impl<'a> ServerMessage<&'a str> {
 }
 
 #[cfg(test)]
+#[expect(clippy::unwrap_used)]
 mod tests {
     use super::*;
 

proxy/src/scram/messages.rs

@@ -185,6 +185,7 @@ impl fmt::Debug for OwnedServerFirstMessage {
 }
 
 #[cfg(test)]
+#[expect(clippy::unwrap_used)]
 mod tests {
     use super::*;
 

proxy/src/scram/mod.rs

@@ -57,6 +57,7 @@ fn sha256<'a>(parts: impl IntoIterator<Item = &'a [u8]>) -> [u8; 32] {
 }
 
 #[cfg(test)]
+#[expect(clippy::unwrap_used)]
 mod tests {
     use super::threadpool::ThreadPool;
     use super::{Exchange, ServerSecret};