fix

style
NO_CLOSE_RANGE macro for old glibc/kernel builds
2026-07-07 06:00:38 +00:00 · 2025-06-24 12:50:59 +01:00 · 2025-06-24 12:41:30 +01:00 · 2025-06-24 12:41:30 +01:00
184 changed files with 2231 additions and 5697 deletions
--- a/.dockerignore
+++ b/.dockerignore
@@ -4,7 +4,6 @@
 !Cargo.lock
 !Cargo.toml
 !Makefile
-!postgres.mk
 !rust-toolchain.toml
 !scripts/ninstall.sh
 !docker-compose/run-tests.sh
--- a/.github/workflows/build-macos.yml
+++ b/.github/workflows/build-macos.yml
@@ -94,6 +94,11 @@ jobs:
        run: |
          make "neon-pg-ext-${{ matrix.postgres-version }}" -j$(sysctl -n hw.ncpu)

+      - name: Get postgres headers ${{ matrix.postgres-version }}
+        if: steps.cache_pg.outputs.cache-hit != 'true'
+        run: |
+          make postgres-headers-${{ matrix.postgres-version }} -j$(sysctl -n hw.ncpu)
+
      - name: Upload "pg_install/${{ matrix.postgres-version }}" artifact
        uses: actions/upload-artifact@ea165f8d65b6e75b540449e92b4886f43607fa02 # v4.6.2
        with:
@@ -135,12 +140,6 @@ jobs:
          name: pg_install--v17
          path: pg_install/v17

-      # `actions/download-artifact` doesn't preserve permissions:
-      # https://github.com/actions/download-artifact?tab=readme-ov-file#permission-loss
-      - name: Make pg_install/v*/bin/* executable
-        run: |
-          chmod +x pg_install/v*/bin/*
-
      - name: Cache walproposer-lib
        id: cache_walproposer_lib
        uses: actions/cache@5a3ec84eff668545956fd18022155c47e93e2684 # v4.2.3
@@ -168,7 +167,7 @@ jobs:
      - name: Build walproposer-lib (only for v17)
        if: steps.cache_walproposer_lib.outputs.cache-hit != 'true'
        run:
-          make walproposer-lib -j$(sysctl -n hw.ncpu) PG_INSTALL_CACHED=1
+          make walproposer-lib -j$(sysctl -n hw.ncpu)

      - name: Upload "build/walproposer-lib" artifact
        uses: actions/upload-artifact@ea165f8d65b6e75b540449e92b4886f43607fa02 # v4.6.2
--- a/.github/workflows/build_and_test.yml
+++ b/.github/workflows/build_and_test.yml
@@ -69,7 +69,7 @@ jobs:
          submodules: true

      - name: Check for file changes
-        uses: dorny/paths-filter@de90cc6fb38fc0963ad72b210f1f284cd68cea36  # v3.0.2
+        uses: step-security/paths-filter@v3
        id: files-changed
        with:
          token: ${{ secrets.GITHUB_TOKEN }}
--- a/.github/workflows/large_oltp_benchmark.yml
+++ b/.github/workflows/large_oltp_benchmark.yml
@@ -153,7 +153,7 @@ jobs:
        PERF_TEST_RESULT_CONNSTR: "${{ secrets.PERF_TEST_RESULT_CONNSTR }}"

    - name: Benchmark database maintenance
-      if: ${{ matrix.test_maintenance }}
+      if: ${{ matrix.test_maintenance == 'true' }}
      uses: ./.github/actions/run-python-test-set
      with:
        build_type: ${{ env.BUILD_TYPE }}
--- a/.github/workflows/neon_extra_builds.yml
+++ b/.github/workflows/neon_extra_builds.yml
@@ -53,7 +53,7 @@ jobs:
          submodules: true

      - name: Check for Postgres changes
-        uses: dorny/paths-filter@1441771bbfdd59dcd748680ee64ebd8faab1a242  #v3
+        uses: step-security/paths-filter@v3
        id: files_changed
        with:
          token: ${{ github.token }}
--- a/.github/workflows/pre-merge-checks.yml
+++ b/.github/workflows/pre-merge-checks.yml
@@ -34,7 +34,7 @@ jobs:

      - uses: actions/checkout@11bd71901bbe5b1630ceea73d27597364c9af683 # v4.2.2

-      - uses: tj-actions/changed-files@ed68ef82c095e0d48ec87eccea555d944a631a4c # v46.0.5
+      - uses: step-security/changed-files@3dbe17c78367e7d60f00d78ae6781a35be47b4a1 # v45.0.1
        id: python-src
        with:
          files: |
@@ -45,7 +45,7 @@ jobs:
            poetry.lock
            pyproject.toml

-      - uses: tj-actions/changed-files@ed68ef82c095e0d48ec87eccea555d944a631a4c # v46.0.5
+      - uses: step-security/changed-files@3dbe17c78367e7d60f00d78ae6781a35be47b4a1 # v45.0.1
        id: rust-src
        with:
          files: |
--- a/.github/workflows/proxy-benchmark.yml
+++ b/.github/workflows/proxy-benchmark.yml
@@ -60,23 +60,22 @@ jobs:
        } >> "$GITHUB_ENV"

    - name: Run proxy-bench
-      run: ${PROXY_BENCH_PATH}/run.sh
+      run: ./${PROXY_BENCH_PATH}/run.sh

    - name: Ingest Bench Results # neon repo script
-      if: always()
+      if: success()
      run: |
        mkdir -p $TEST_OUTPUT
        python $NEON_DIR/scripts/proxy_bench_results_ingest.py --out $TEST_OUTPUT

    - name: Push Metrics to Proxy perf database
-      if: always()
+      if: success()
      env:
        PERF_TEST_RESULT_CONNSTR: "${{ secrets.PROXY_TEST_RESULT_CONNSTR }}"
        REPORT_FROM: $TEST_OUTPUT
      run: $NEON_DIR/scripts/generate_and_push_perf_report.sh

    - name: Docker cleanup
-      if: always()
      run: docker compose down

    - name: Notify Failure
--- a/.gitignore
+++ b/.gitignore
@@ -6,7 +6,6 @@
 /tmp_check_cli
 __pycache__/
 test_output/
-neon_previous/
 .vscode
 .idea
 *.swp
--- a/Cargo.lock
+++ b/Cargo.lock
@@ -1279,7 +1279,6 @@ dependencies = [
 "remote_storage",
 "serde",
 "serde_json",
- "url",
 "utils",
 ]

@@ -1305,7 +1304,6 @@ dependencies = [
 "fail",
 "flate2",
 "futures",
- "hostname-validator",
 "http 1.1.0",
 "indexmap 2.9.0",
 "itertools 0.10.5",
@@ -1318,10 +1316,8 @@ dependencies = [
 "opentelemetry",
 "opentelemetry_sdk",
 "p256 0.13.2",
- "pageserver_page_api",
 "postgres",
 "postgres_initdb",
- "postgres_versioninfo",
 "regex",
 "remote_storage",
 "reqwest",
@@ -1338,7 +1334,6 @@ dependencies = [
 "tokio-postgres",
 "tokio-stream",
 "tokio-util",
- "tonic 0.13.1",
 "tower 0.5.2",
 "tower-http",
 "tower-otel",
@@ -2772,12 +2767,6 @@ dependencies = [
 "windows",
 ]

-[[package]]
-name = "hostname-validator"
-version = "1.1.1"
-source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "f558a64ac9af88b5ba400d99b579451af0d39c6d360980045b91aac966d705e2"
-
 [[package]]
 name = "http"
 version = "0.2.9"
@@ -4417,8 +4406,6 @@ dependencies = [
 "once_cell",
 "postgres_backend",
 "postgres_ffi_types",
- "postgres_versioninfo",
- "posthog_client_lite",
 "rand 0.8.5",
 "remote_storage",
 "reqwest",
@@ -4429,7 +4416,6 @@ dependencies = [
 "strum",
 "strum_macros",
 "thiserror 1.0.69",
- "tracing",
 "tracing-utils",
 "utils",
 ]
@@ -4443,7 +4429,6 @@ dependencies = [
 "futures",
 "http-utils",
 "pageserver_api",
- "postgres_versioninfo",
 "reqwest",
 "serde",
 "thiserror 1.0.69",
@@ -4486,14 +4471,12 @@ dependencies = [
 "bytes",
 "futures",
 "pageserver_api",
- "postgres_ffi_types",
+ "postgres_ffi",
 "prost 0.13.5",
- "prost-types 0.13.5",
 "strum",
 "strum_macros",
 "thiserror 1.0.69",
 "tokio",
- "tokio-util",
 "tonic 0.13.1",
 "tonic-build",
 "utils",
@@ -4914,7 +4897,6 @@ dependencies = [
 "once_cell",
 "postgres",
 "postgres_ffi_types",
- "postgres_versioninfo",
 "pprof",
 "regex",
 "serde",
@@ -4937,23 +4919,11 @@ version = "0.1.0"
 dependencies = [
 "anyhow",
 "camino",
- "postgres_versioninfo",
 "thiserror 1.0.69",
 "tokio",
 "workspace_hack",
 ]

-[[package]]
-name = "postgres_versioninfo"
-version = "0.1.0"
-dependencies = [
- "anyhow",
- "serde",
- "serde_repr",
- "thiserror 1.0.69",
- "workspace_hack",
-]
-
 [[package]]
 name = "posthog_client_lite"
 version = "0.1.0"
@@ -5166,7 +5136,7 @@ dependencies = [
 "petgraph",
 "prettyplease",
 "prost 0.13.5",
- "prost-types 0.13.5",
+ "prost-types 0.13.3",
 "regex",
 "syn 2.0.100",
 "tempfile",
@@ -5209,9 +5179,9 @@ dependencies = [

 [[package]]
 name = "prost-types"
-version = "0.13.5"
+version = "0.13.3"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "52c2c1bf36ddb1a1c396b3601a3cec27c2462e45f07c386894ec3ccf5332bd16"
+checksum = "4759aa0d3a6232fb8dbdb97b61de2c20047c68aca932c7ed76da9d788508d670"
 dependencies = [
 "prost 0.13.5",
 ]
@@ -6145,7 +6115,6 @@ dependencies = [
 "postgres-protocol",
 "postgres_backend",
 "postgres_ffi",
- "postgres_versioninfo",
 "pprof",
 "pq_proto",
 "rand 0.8.5",
@@ -6190,7 +6159,6 @@ dependencies = [
 "const_format",
 "pageserver_api",
 "postgres_ffi",
- "postgres_versioninfo",
 "pq_proto",
 "serde",
 "serde_json",
@@ -6513,17 +6481,6 @@ dependencies = [
 "thiserror 1.0.69",
 ]

-[[package]]
-name = "serde_repr"
-version = "0.1.20"
-source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "175ee3e80ae9982737ca543e96133087cbd9a485eecc3bc4de9c1a37b47ea59c"
-dependencies = [
- "proc-macro2",
- "quote",
- "syn 2.0.100",
-]
-
 [[package]]
 name = "serde_spanned"
 version = "0.6.6"
@@ -6818,7 +6775,6 @@ dependencies = [
 "chrono",
 "clap",
 "clashmap",
- "compute_api",
 "control_plane",
 "cron",
 "diesel",
@@ -6830,7 +6786,6 @@ dependencies = [
 "hex",
 "http-utils",
 "humantime",
- "humantime-serde",
 "hyper 0.14.30",
 "itertools 0.10.5",
 "json-structural-diff",
@@ -6841,7 +6796,6 @@ dependencies = [
 "pageserver_api",
 "pageserver_client",
 "postgres_connection",
- "posthog_client_lite",
 "rand 0.8.5",
 "regex",
 "reqwest",
@@ -7652,7 +7606,7 @@ dependencies = [
 "prettyplease",
 "proc-macro2",
 "prost-build 0.13.3",
- "prost-types 0.13.5",
+ "prost-types 0.13.3",
 "quote",
 "syn 2.0.100",
 ]
@@ -7664,7 +7618,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "f9687bd5bfeafebdded2356950f278bba8226f0b32109537c4253406e09aafe1"
 dependencies = [
 "prost 0.13.5",
- "prost-types 0.13.5",
+ "prost-types 0.13.3",
 "tokio",
 "tokio-stream",
 "tonic 0.13.1",
@@ -8693,6 +8647,7 @@ dependencies = [
 "num-iter",
 "num-rational",
 "num-traits",
+ "once_cell",
 "p256 0.13.2",
 "parquet",
 "prettyplease",
--- a/Cargo.toml
+++ b/Cargo.toml
@@ -23,7 +23,6 @@ members = [
    "libs/pageserver_api",
    "libs/postgres_ffi",
    "libs/postgres_ffi_types",
-    "libs/postgres_versioninfo",
    "libs/safekeeper_api",
    "libs/desim",
    "libs/neon-shmem",
@@ -152,7 +151,6 @@ pprof = { version = "0.14", features = ["criterion", "flamegraph", "frame-pointe
 procfs = "0.16"
 prometheus = {version = "0.13", default-features=false, features = ["process"]} # removes protobuf dependency
 prost = "0.13.5"
-prost-types = "0.13.5"
 rand = "0.8"
 redis = { version = "0.29.2", features = ["tokio-rustls-comp", "keep-alive"] }
 regex = "1.10.2"
@@ -176,7 +174,6 @@ serde_json = "1"
 serde_path_to_error = "0.1"
 serde_with = { version = "3", features = [ "base64" ] }
 serde_assert = "0.5.0"
-serde_repr = "0.1.20"
 sha2 = "0.10.2"
 signal-hook = "0.3"
 smallvec = "1.11"
@@ -200,7 +197,7 @@ tokio-postgres-rustls = "0.12.0"
 tokio-rustls = { version = "0.26.0", default-features = false, features = ["tls12", "ring"]}
 tokio-stream = "0.1"
 tokio-tar = "0.3"
-tokio-util = { version = "0.7.10", features = ["io", "io-util", "rt"] }
+tokio-util = { version = "0.7.10", features = ["io", "rt"] }
 toml = "0.8"
 toml_edit = "0.22"
 tonic = { version = "0.13.1", default-features = false, features = ["channel", "codegen", "gzip", "prost", "router", "server", "tls-ring", "tls-native-roots", "zstd"] }
@@ -264,7 +261,6 @@ postgres_backend = { version = "0.1", path = "./libs/postgres_backend/" }
 postgres_connection = { version = "0.1", path = "./libs/postgres_connection/" }
 postgres_ffi = { version = "0.1", path = "./libs/postgres_ffi/" }
 postgres_ffi_types = { version = "0.1", path = "./libs/postgres_ffi_types/" }
-postgres_versioninfo = { version = "0.1", path = "./libs/postgres_versioninfo/" }
 postgres_initdb = { path = "./libs/postgres_initdb" }
 posthog_client_lite = { version = "0.1", path = "./libs/posthog_client_lite" }
 pq_proto = { version = "0.1", path = "./libs/pq_proto/" }
--- a/1
+++ b/1
@@ -40,7 +40,6 @@ COPY --chown=nonroot vendor/postgres-v16 vendor/postgres-v16
 COPY --chown=nonroot vendor/postgres-v17 vendor/postgres-v17
 COPY --chown=nonroot pgxn pgxn
 COPY --chown=nonroot Makefile Makefile
-COPY --chown=nonroot postgres.mk postgres.mk
 COPY --chown=nonroot scripts/ninstall.sh scripts/ninstall.sh

 ENV BUILD_TYPE=release
--- a/131
+++ b/131
@@ -4,14 +4,11 @@ ROOT_PROJECT_DIR := $(dir $(abspath $(lastword $(MAKEFILE_LIST))))
 # managers.
 POSTGRES_INSTALL_DIR ?= $(ROOT_PROJECT_DIR)/pg_install/

-# Supported PostgreSQL versions
-POSTGRES_VERSIONS = v17 v16 v15 v14
-
 # CARGO_BUILD_FLAGS: Extra flags to pass to `cargo build`. `--locked`
 # and `--features testing` are popular examples.
 #
-# CARGO_PROFILE: Set to override the cargo profile to use. By default,
-# it is derived from BUILD_TYPE.
+# CARGO_PROFILE: You can also set to override the cargo profile to
+# use. By default, it is derived from BUILD_TYPE.

 # All intermediate build artifacts are stored here.
 BUILD_DIR := build
@@ -98,24 +95,93 @@ CACHEDIR_TAG_CONTENTS := "Signature: 8a477f597d28d172789f06886806bc55"
 # Top level Makefile to build Neon and PostgreSQL
 #
 .PHONY: all
-all: neon postgres-install neon-pg-ext
+all: neon postgres neon-pg-ext

 ### Neon Rust bits
 #
 # The 'postgres_ffi' depends on the Postgres headers.
 .PHONY: neon
-neon: postgres-headers-install walproposer-lib cargo-target-dir
+neon: postgres-headers walproposer-lib cargo-target-dir
 	+@echo "Compiling Neon"
 	$(CARGO_CMD_PREFIX) cargo build $(CARGO_BUILD_FLAGS) $(CARGO_PROFILE)
-
 .PHONY: cargo-target-dir
 cargo-target-dir:
 	# https://github.com/rust-lang/cargo/issues/14281
 	mkdir -p target
 	test -e target/CACHEDIR.TAG || echo "$(CACHEDIR_TAG_CONTENTS)" > target/CACHEDIR.TAG

+### PostgreSQL parts
+# Some rules are duplicated for Postgres v14 and 15. We may want to refactor
+# to avoid the duplication in the future, but it's tolerable for now.
+#
+$(BUILD_DIR)/%/config.status:
+	mkdir -p $(BUILD_DIR)
+	test -e $(BUILD_DIR)/CACHEDIR.TAG || echo "$(CACHEDIR_TAG_CONTENTS)" > $(BUILD_DIR)/CACHEDIR.TAG
+
+	+@echo "Configuring Postgres $* build"
+	@test -s $(ROOT_PROJECT_DIR)/vendor/postgres-$*/configure || { \
+		echo "\nPostgres submodule not found in $(ROOT_PROJECT_DIR)/vendor/postgres-$*/, execute "; \
+		echo "'git submodule update --init --recursive --depth 2 --progress .' in project root.\n"; \
+		exit 1; }
+	mkdir -p $(BUILD_DIR)/$*
+
+	VERSION=$*; \
+	EXTRA_VERSION=$$(cd $(ROOT_PROJECT_DIR)/vendor/postgres-$$VERSION && git rev-parse HEAD); \
+	(cd $(BUILD_DIR)/$$VERSION && \
+	env PATH="$(EXTRA_PATH_OVERRIDES):$$PATH" $(ROOT_PROJECT_DIR)/vendor/postgres-$$VERSION/configure \
+		CFLAGS='$(PG_CFLAGS)' LDFLAGS='$(PG_LDFLAGS)' \
+		$(PG_CONFIGURE_OPTS) --with-extra-version=" ($$EXTRA_VERSION)" \
+		--prefix=$(abspath $(POSTGRES_INSTALL_DIR))/$$VERSION > configure.log)
+
+# nicer alias to run 'configure'
+# Note: I've been unable to use templates for this part of our configuration.
+# I'm not sure why it wouldn't work, but this is the only place (apart from
+# the "build-all-versions" entry points) where direct mention of PostgreSQL
+# versions is used.
+.PHONY: postgres-configure-v17
+postgres-configure-v17: $(BUILD_DIR)/v17/config.status
+.PHONY: postgres-configure-v16
+postgres-configure-v16: $(BUILD_DIR)/v16/config.status
+.PHONY: postgres-configure-v15
+postgres-configure-v15: $(BUILD_DIR)/v15/config.status
+.PHONY: postgres-configure-v14
+postgres-configure-v14: $(BUILD_DIR)/v14/config.status
+
+# Install the PostgreSQL header files into $(POSTGRES_INSTALL_DIR)/<version>/include
+.PHONY: postgres-headers-%
+postgres-headers-%: postgres-configure-%
+	+@echo "Installing PostgreSQL $* headers"
+	$(MAKE) -C $(BUILD_DIR)/$*/src/include MAKELEVEL=0 install
+
+# Compile and install PostgreSQL
+.PHONY: postgres-%
+postgres-%: postgres-configure-% \
+		  postgres-headers-% # to prevent `make install` conflicts with neon's `postgres-headers`
+	+@echo "Compiling PostgreSQL $*"
+	$(MAKE) -C $(BUILD_DIR)/$* MAKELEVEL=0 install
+	+@echo "Compiling libpq $*"
+	$(MAKE) -C $(BUILD_DIR)/$*/src/interfaces/libpq install
+	+@echo "Compiling pg_prewarm $*"
+	$(MAKE) -C $(BUILD_DIR)/$*/contrib/pg_prewarm install
+	+@echo "Compiling pg_buffercache $*"
+	$(MAKE) -C $(BUILD_DIR)/$*/contrib/pg_buffercache install
+	+@echo "Compiling pg_visibility $*"
+	$(MAKE) -C $(BUILD_DIR)/$*/contrib/pg_visibility install
+	+@echo "Compiling pageinspect $*"
+	$(MAKE) -C $(BUILD_DIR)/$*/contrib/pageinspect install
+	+@echo "Compiling pg_trgm $*"
+	$(MAKE) -C $(BUILD_DIR)/$*/contrib/pg_trgm install
+	+@echo "Compiling amcheck $*"
+	$(MAKE) -C $(BUILD_DIR)/$*/contrib/amcheck install
+	+@echo "Compiling test_decoding $*"
+	$(MAKE) -C $(BUILD_DIR)/$*/contrib/test_decoding install
+
+.PHONY: postgres-check-%
+postgres-check-%: postgres-%
+	$(MAKE) -C $(BUILD_DIR)/$* MAKELEVEL=0 check
+
 .PHONY: neon-pg-ext-%
-neon-pg-ext-%: postgres-install-%
+neon-pg-ext-%: postgres-%
 	+@echo "Compiling neon-specific Postgres extensions for $*"
 	mkdir -p $(BUILD_DIR)/pgxn-$*
 	$(MAKE) PG_CONFIG=$(POSTGRES_INSTALL_DIR)/$*/bin/pg_config COPT='$(COPT)' \
@@ -154,14 +220,39 @@ ifeq ($(UNAME_S),Linux)
 		pg_crc32c.o
 endif

-# Shorthand to call neon-pg-ext-% target for all Postgres versions
 .PHONY: neon-pg-ext
-neon-pg-ext: $(foreach pg_version,$(POSTGRES_VERSIONS),neon-pg-ext-$(pg_version))
+neon-pg-ext: \
+	neon-pg-ext-v14 \
+	neon-pg-ext-v15 \
+	neon-pg-ext-v16 \
+	neon-pg-ext-v17
+
+# shorthand to build all Postgres versions
+.PHONY: postgres
+postgres: \
+	postgres-v14 \
+	postgres-v15 \
+	postgres-v16 \
+	postgres-v17
+
+.PHONY: postgres-headers
+postgres-headers: \
+	postgres-headers-v14 \
+	postgres-headers-v15 \
+	postgres-headers-v16 \
+	postgres-headers-v17
+
+.PHONY: postgres-check
+postgres-check: \
+	postgres-check-v14 \
+	postgres-check-v15 \
+	postgres-check-v16 \
+	postgres-check-v17

 # This removes everything
 .PHONY: distclean
 distclean:
-	$(RM) -r $(POSTGRES_INSTALL_DIR) $(BUILD_DIR)
+	$(RM) -r $(POSTGRES_INSTALL_DIR)
 	$(CARGO_CMD_PREFIX) cargo clean

 .PHONY: fmt
@@ -209,19 +300,3 @@ neon-pgindent: postgres-v17-pg-bsd-indent neon-pg-ext-v17
 .PHONY: setup-pre-commit-hook
 setup-pre-commit-hook:
 	ln -s -f $(ROOT_PROJECT_DIR)/pre-commit.py .git/hooks/pre-commit
-
-# Targets for building PostgreSQL are defined in postgres.mk.
-#
-# But if the caller has indicated that PostgreSQL is already
-# installed, by setting the PG_INSTALL_CACHED variable, skip it.
-ifdef PG_INSTALL_CACHED
-postgres-install: skip-install
-$(foreach pg_version,$(POSTGRES_VERSIONS),postgres-install-$(pg_version)): skip-install
-postgres-headers-install:
-	+@echo "Skipping installation of PostgreSQL headers because PG_INSTALL_CACHED is set"
-skip-install:
-	+@echo "Skipping PostgreSQL installation because PG_INSTALL_CACHED is set"
-
-else
-include postgres.mk
-endif
--- a/build-tools.Dockerfile
+++ b/build-tools.Dockerfile
@@ -165,7 +165,6 @@ RUN curl -fsSL \
    && rm sql_exporter.tar.gz

 # protobuf-compiler (protoc)
-# Keep the version the same as in compute/compute-node.Dockerfile
 ENV PROTOC_VERSION=25.1
 RUN curl -fsSL "https://github.com/protocolbuffers/protobuf/releases/download/v${PROTOC_VERSION}/protoc-${PROTOC_VERSION}-linux-$(uname -m | sed 's/aarch64/aarch_64/g').zip" -o "protoc.zip" \
    && unzip -q protoc.zip -d protoc \
@@ -180,7 +179,7 @@ RUN curl -sL "https://github.com/peak/s5cmd/releases/download/v${S5CMD_VERSION}/
    && mv s5cmd /usr/local/bin/s5cmd

 # LLVM
-ENV LLVM_VERSION=20
+ENV LLVM_VERSION=19
 RUN curl -fsSL 'https://apt.llvm.org/llvm-snapshot.gpg.key' | apt-key add - \
    && echo "deb http://apt.llvm.org/${DEBIAN_VERSION}/ llvm-toolchain-${DEBIAN_VERSION}-${LLVM_VERSION} main" > /etc/apt/sources.list.d/llvm.stable.list \
    && apt update \
@@ -293,7 +292,7 @@ WORKDIR /home/nonroot

 # Rust
 # Please keep the version of llvm (installed above) in sync with rust llvm (`rustc --version --verbose | grep LLVM`)
-ENV RUSTC_VERSION=1.88.0
+ENV RUSTC_VERSION=1.87.0
 ENV RUSTUP_HOME="/home/nonroot/.rustup"
 ENV PATH="/home/nonroot/.cargo/bin:${PATH}"
 ARG RUSTFILT_VERSION=0.2.1
--- a/compute/Makefile
+++ b/compute/Makefile
@@ -22,7 +22,7 @@ sql_exporter.yml: $(jsonnet_files)
 		--output-file etc/$@ \
 		--tla-str collector_name=neon_collector \
 		--tla-str collector_file=neon_collector.yml \
-		--tla-str 'connection_string=postgresql://cloud_admin@127.0.0.1:5432/postgres?sslmode=disable&application_name=sql_exporter&pgaudit.log=none' \
+		--tla-str 'connection_string=postgresql://cloud_admin@127.0.0.1:5432/postgres?sslmode=disable&application_name=sql_exporter' \
 		etc/sql_exporter.jsonnet

 sql_exporter_autoscaling.yml: $(jsonnet_files)
@@ -30,7 +30,7 @@ sql_exporter_autoscaling.yml: $(jsonnet_files)
 		--output-file etc/$@ \
 		--tla-str collector_name=neon_collector_autoscaling \
 		--tla-str collector_file=neon_collector_autoscaling.yml \
-		--tla-str 'connection_string=postgresql://cloud_admin@127.0.0.1:5432/postgres?sslmode=disable&application_name=sql_exporter_autoscaling&pgaudit.log=none' \
+		--tla-str 'connection_string=postgresql://cloud_admin@127.0.0.1:5432/postgres?sslmode=disable&application_name=sql_exporter_autoscaling' \
 		etc/sql_exporter.jsonnet

 .PHONY: clean
--- a/compute/compute-node.Dockerfile
+++ b/compute/compute-node.Dockerfile
@@ -115,9 +115,6 @@ ARG EXTENSIONS=all
 FROM $BASE_IMAGE_SHA AS build-deps
 ARG DEBIAN_VERSION

-# Keep in sync with build-tools.Dockerfile
-ENV PROTOC_VERSION=25.1
-
 # Use strict mode for bash to catch errors early
 SHELL ["/bin/bash", "-euo", "pipefail", "-c"]

@@ -152,14 +149,8 @@ RUN case $DEBIAN_VERSION in \
    libclang-dev \
    jsonnet \
    $VERSION_INSTALLS \
-    && apt clean && rm -rf /var/lib/apt/lists/* \
-    && useradd -ms /bin/bash nonroot -b /home \
-    # Install protoc from binary release, since Debian's versions are too old.
-    && curl -fsSL "https://github.com/protocolbuffers/protobuf/releases/download/v${PROTOC_VERSION}/protoc-${PROTOC_VERSION}-linux-$(uname -m | sed 's/aarch64/aarch_64/g').zip" -o "protoc.zip" \
-    && unzip -q protoc.zip -d protoc \
-    && mv protoc/bin/protoc /usr/local/bin/protoc \
-    && mv protoc/include/google /usr/local/include/google \
-    && rm -rf protoc.zip protoc
+    && apt clean && rm -rf /var/lib/apt/lists/* && \
+    useradd -ms /bin/bash nonroot -b /home

 #########################################################################################
 #
@@ -180,6 +171,9 @@ RUN cd postgres && \
    eval $CONFIGURE_CMD && \
    make MAKELEVEL=0 -j $(getconf _NPROCESSORS_ONLN) -s install && \
    make MAKELEVEL=0 -j $(getconf _NPROCESSORS_ONLN) -s -C contrib/ install && \
+    # Install headers
+    make MAKELEVEL=0 -j $(getconf _NPROCESSORS_ONLN) -s -C src/include install && \
+    make MAKELEVEL=0 -j $(getconf _NPROCESSORS_ONLN) -s -C src/interfaces/libpq install && \
    # Enable some of contrib extensions
    echo 'trusted = true' >> /usr/local/pgsql/share/extension/autoinc.control && \
    echo 'trusted = true' >> /usr/local/pgsql/share/extension/dblink.control && \
@@ -1179,7 +1173,7 @@ COPY --from=pgrag-src /ext-src/ /ext-src/
 # Install it using virtual environment, because Python 3.11 (the default version on Debian 12 (Bookworm)) complains otherwise
 WORKDIR /ext-src/onnxruntime-src
 RUN apt update && apt install --no-install-recommends --no-install-suggests -y \
-    python3 python3-pip python3-venv && \
+    python3 python3-pip python3-venv protobuf-compiler && \
    apt clean && rm -rf /var/lib/apt/lists/* && \
    python3 -m venv venv && \
    . venv/bin/activate && \
@@ -1574,20 +1568,20 @@ ARG PG_VERSION
 WORKDIR /ext-src
 RUN case "${PG_VERSION}" in \
    "v14") \
-    export PGAUDIT_VERSION=1.6.3 \
-    export PGAUDIT_CHECKSUM=37a8f5a7cc8d9188e536d15cf0fdc457fcdab2547caedb54442c37f124110919 \
+    export PGAUDIT_VERSION=1.6.2 \
+    export PGAUDIT_CHECKSUM=1f350d70a0cbf488c0f2b485e3a5c9b11f78ad9e3cbb95ef6904afa1eb3187eb \
    ;; \
    "v15") \
-    export PGAUDIT_VERSION=1.7.1 \
-    export PGAUDIT_CHECKSUM=e9c8e6e092d82b2f901d72555ce0fe7780552f35f8985573796cd7e64b09d4ec \
+    export PGAUDIT_VERSION=1.7.0 \
+    export PGAUDIT_CHECKSUM=8f4a73e451c88c567e516e6cba7dc1e23bc91686bb6f1f77f8f3126d428a8bd8 \
    ;; \
    "v16") \
-    export PGAUDIT_VERSION=16.1 \
-    export PGAUDIT_CHECKSUM=3bae908ab70ba0c6f51224009dbcfff1a97bd6104c6273297a64292e1b921fee \
+    export PGAUDIT_VERSION=16.0 \
+    export PGAUDIT_CHECKSUM=d53ef985f2d0b15ba25c512c4ce967dce07b94fd4422c95bd04c4c1a055fe738 \
    ;; \
    "v17") \
-    export PGAUDIT_VERSION=17.1 \
-    export PGAUDIT_CHECKSUM=9c5f37504d393486cc75d2ced83f75f5899be64fa85f689d6babb833b4361e6c \
+    export PGAUDIT_VERSION=17.0 \
+    export PGAUDIT_CHECKSUM=7d0d08d030275d525f36cd48b38c6455f1023da863385badff0cec44965bfd8c \
    ;; \
    *) \
    echo "pgaudit is not supported on this PostgreSQL version" && exit 1;; \
@@ -1983,7 +1977,7 @@ RUN apt update && \
        locales \
        lsof \
        procps \
-        rsyslog-gnutls \
+        rsyslog \
        screen \
        tcpdump \
        $VERSION_INSTALLS && \
--- a/compute/etc/neon_collector.jsonnet
+++ b/compute/etc/neon_collector.jsonnet
@@ -8,8 +8,6 @@
    import 'sql_exporter/compute_logical_snapshot_files.libsonnet',
    import 'sql_exporter/compute_logical_snapshots_bytes.libsonnet',
    import 'sql_exporter/compute_max_connections.libsonnet',
-    import 'sql_exporter/compute_pg_oldest_frozen_xid_age.libsonnet',
-    import 'sql_exporter/compute_pg_oldest_mxid_age.libsonnet',
    import 'sql_exporter/compute_receive_lsn.libsonnet',
    import 'sql_exporter/compute_subscriptions_count.libsonnet',
    import 'sql_exporter/connection_counts.libsonnet',
--- a/compute/etc/sql_exporter/compute_pg_oldest_frozen_xid_age.libsonnet
+++ b/compute/etc/sql_exporter/compute_pg_oldest_frozen_xid_age.libsonnet
@@ -1,13 +0,0 @@
-{
-  metric_name: 'compute_pg_oldest_frozen_xid_age',
-  type: 'gauge',
-  help: 'Age of oldest XIDs that have not been frozen by VACUUM. An indicator of how long it has been since VACUUM last ran.',
-  key_labels: [
-    'database_name',
-  ],
-  value_label: 'metric',
-  values: [
-    'frozen_xid_age',
-  ],
-  query: importstr 'sql_exporter/compute_pg_oldest_frozen_xid_age.sql',
-}
--- a/compute/etc/sql_exporter/compute_pg_oldest_frozen_xid_age.sql
+++ b/compute/etc/sql_exporter/compute_pg_oldest_frozen_xid_age.sql
@@ -1,4 +0,0 @@
-SELECT datname database_name,
-  age(datfrozenxid) frozen_xid_age
-FROM pg_database
-ORDER BY frozen_xid_age DESC LIMIT 10;
--- a/compute/etc/sql_exporter/compute_pg_oldest_mxid_age.libsonnet
+++ b/compute/etc/sql_exporter/compute_pg_oldest_mxid_age.libsonnet
@@ -1,13 +0,0 @@
-{
-  metric_name: 'compute_pg_oldest_mxid_age',
-  type: 'gauge',
-  help: 'Age of oldest MXIDs that have not been replaced by VACUUM. An indicator of how long it has been since VACUUM last ran.',
-  key_labels: [
-    'database_name',
-  ],
-  value_label: 'metric',
-  values: [
-    'min_mxid_age',
-  ],
-  query: importstr 'sql_exporter/compute_pg_oldest_mxid_age.sql',
-}
--- a/compute/etc/sql_exporter/compute_pg_oldest_mxid_age.sql
+++ b/compute/etc/sql_exporter/compute_pg_oldest_mxid_age.sql
@@ -1,4 +0,0 @@
-SELECT datname database_name,
-  mxid_age(datminmxid) min_mxid_age
-FROM pg_database
-ORDER BY min_mxid_age DESC LIMIT 10;
--- a/compute/patches/anon_v2.patch
+++ b/compute/patches/anon_v2.patch
@@ -1,8 +1,8 @@
 diff --git a/sql/anon.sql b/sql/anon.sql
-index 0cdc769..b450327 100644
+index 0cdc769..f6cc950 100644
 --- a/sql/anon.sql
 +++ b/sql/anon.sql
-@@ -1141,3 +1141,15 @@ $$
+@@ -1141,3 +1141,8 @@ $$
 -- TODO : https://en.wikipedia.org/wiki/L-diversity
 
 -- TODO : https://en.wikipedia.org/wiki/T-closeness
@@ -11,13 +11,6 @@ index 0cdc769..b450327 100644
 +
 +GRANT ALL ON SCHEMA anon to neon_superuser;
 +GRANT ALL ON ALL TABLES IN SCHEMA anon TO neon_superuser;
-+
-+DO $$
-+BEGIN
-+    IF current_setting('server_version_num')::int >= 150000 THEN
-+        GRANT SET ON PARAMETER anon.transparent_dynamic_masking TO neon_superuser;
-+    END IF;
-+END $$;
 diff --git a/sql/init.sql b/sql/init.sql
 index 7da6553..9b6164b 100644
 --- a/sql/init.sql
--- a/compute/vm-image-spec-bookworm.yaml
+++ b/compute/vm-image-spec-bookworm.yaml
@@ -26,7 +26,7 @@ commands:
  - name: postgres-exporter
    user: nobody
    sysvInitAction: respawn
-    shell: 'DATA_SOURCE_NAME="user=cloud_admin sslmode=disable dbname=postgres application_name=postgres-exporter pgaudit.log=none" /bin/postgres_exporter --config.file=/etc/postgres_exporter.yml'
+    shell: 'DATA_SOURCE_NAME="user=cloud_admin sslmode=disable dbname=postgres application_name=postgres-exporter" /bin/postgres_exporter --config.file=/etc/postgres_exporter.yml'
  - name: pgbouncer-exporter
    user: postgres
    sysvInitAction: respawn
@@ -59,7 +59,7 @@ files:
      # the rules use ALL as the hostname. Avoid the pointless lookups and the "unable to
      # resolve host" log messages that they generate.
      Defaults !fqdn
-
+      
      # Allow postgres user (which is what compute_ctl runs as) to run /neonvm/bin/resize-swap
      # and /neonvm/bin/set-disk-quota as root without requiring entering a password (NOPASSWD),
      # regardless of hostname (ALL)
--- a/compute/vm-image-spec-bullseye.yaml
+++ b/compute/vm-image-spec-bullseye.yaml
@@ -26,7 +26,7 @@ commands:
  - name: postgres-exporter
    user: nobody
    sysvInitAction: respawn
-    shell: 'DATA_SOURCE_NAME="user=cloud_admin sslmode=disable dbname=postgres application_name=postgres-exporter pgaudit.log=none" /bin/postgres_exporter --config.file=/etc/postgres_exporter.yml'
+    shell: 'DATA_SOURCE_NAME="user=cloud_admin sslmode=disable dbname=postgres application_name=postgres-exporter" /bin/postgres_exporter --config.file=/etc/postgres_exporter.yml'
  - name: pgbouncer-exporter
    user: postgres
    sysvInitAction: respawn
@@ -59,7 +59,7 @@ files:
      # the rules use ALL as the hostname. Avoid the pointless lookups and the "unable to
      # resolve host" log messages that they generate.
      Defaults !fqdn
-
+      
      # Allow postgres user (which is what compute_ctl runs as) to run /neonvm/bin/resize-swap
      # and /neonvm/bin/set-disk-quota as root without requiring entering a password (NOPASSWD),
      # regardless of hostname (ALL)
--- a/compute_tools/Cargo.toml
+++ b/compute_tools/Cargo.toml
@@ -27,7 +27,6 @@ fail.workspace = true
 flate2.workspace = true
 futures.workspace = true
 http.workspace = true
-hostname-validator = "1.1"
 indexmap.workspace = true
 itertools.workspace = true
 jsonwebtoken.workspace = true
@@ -39,7 +38,6 @@ once_cell.workspace = true
 opentelemetry.workspace = true
 opentelemetry_sdk.workspace = true
 p256 = { version = "0.13", features = ["pem"] }
-pageserver_page_api.workspace = true
 postgres.workspace = true
 regex.workspace = true
 reqwest = { workspace = true, features = ["json"] }
@@ -55,7 +53,6 @@ tokio = { workspace = true, features = ["rt", "rt-multi-thread"] }
 tokio-postgres.workspace = true
 tokio-util.workspace = true
 tokio-stream.workspace = true
-tonic.workspace = true
 tower-otel.workspace = true
 tracing.workspace = true
 tracing-opentelemetry.workspace = true
@@ -67,7 +64,6 @@ uuid.workspace = true
 walkdir.workspace = true
 x509-cert.workspace = true

-postgres_versioninfo.workspace = true
 postgres_initdb.workspace = true
 compute_api.workspace = true
 utils.workspace = true
--- a/compute_tools/src/bin/compute_ctl.rs
+++ b/compute_tools/src/bin/compute_ctl.rs
@@ -36,8 +36,6 @@
 use std::ffi::OsString;
 use std::fs::File;
 use std::process::exit;
-use std::sync::Arc;
-use std::sync::atomic::AtomicU64;
 use std::sync::mpsc;
 use std::thread;
 use std::time::Duration;
@@ -192,9 +190,7 @@ fn main() -> Result<()> {
            cgroup: cli.cgroup,
            #[cfg(target_os = "linux")]
            vm_monitor_addr: cli.vm_monitor_addr,
-            installed_extensions_collection_interval: Arc::new(AtomicU64::new(
-                cli.installed_extensions_collection_interval,
-            )),
+            installed_extensions_collection_interval: cli.installed_extensions_collection_interval,
        },
        config,
    )?;
--- a/compute_tools/src/bin/fast_import.rs
+++ b/compute_tools/src/bin/fast_import.rs
@@ -29,7 +29,7 @@ use anyhow::{Context, bail};
 use aws_config::BehaviorVersion;
 use camino::{Utf8Path, Utf8PathBuf};
 use clap::{Parser, Subcommand};
-use compute_tools::extension_server::get_pg_version;
+use compute_tools::extension_server::{PostgresMajorVersion, get_pg_version};
 use nix::unistd::Pid;
 use std::ops::Not;
 use tracing::{Instrument, error, info, info_span, warn};
@@ -179,8 +179,12 @@ impl PostgresProcess {
            .await
            .context("create pgdata directory")?;

-        let pg_version = get_pg_version(self.pgbin.as_ref());
-
+        let pg_version = match get_pg_version(self.pgbin.as_ref()) {
+            PostgresMajorVersion::V14 => 14,
+            PostgresMajorVersion::V15 => 15,
+            PostgresMajorVersion::V16 => 16,
+            PostgresMajorVersion::V17 => 17,
+        };
        postgres_initdb::do_run_initdb(postgres_initdb::RunInitdbArgs {
            superuser: initdb_user,
            locale: DEFAULT_LOCALE, // XXX: this shouldn't be hard-coded,
--- a/compute_tools/src/compute.rs
+++ b/compute_tools/src/compute.rs
@@ -6,7 +6,7 @@ use compute_api::responses::{
    LfcPrewarmState, TlsConfig,
 };
 use compute_api::spec::{
-    ComputeAudit, ComputeFeature, ComputeMode, ComputeSpec, ExtVersion, PageserverProtocol, PgIdent,
+    ComputeAudit, ComputeFeature, ComputeMode, ComputeSpec, ExtVersion, PgIdent,
 };
 use futures::StreamExt;
 use futures::future::join_all;
@@ -15,17 +15,17 @@ use itertools::Itertools;
 use nix::sys::signal::{Signal, kill};
 use nix::unistd::Pid;
 use once_cell::sync::Lazy;
-use pageserver_page_api::{self as page_api, BaseBackupCompression};
 use postgres;
 use postgres::NoTls;
 use postgres::error::SqlState;
 use remote_storage::{DownloadError, RemotePath};
 use std::collections::{HashMap, HashSet};
+use std::net::SocketAddr;
 use std::os::unix::fs::{PermissionsExt, symlink};
 use std::path::Path;
 use std::process::{Command, Stdio};
 use std::str::FromStr;
-use std::sync::atomic::{AtomicU32, AtomicU64, Ordering};
+use std::sync::atomic::{AtomicU32, Ordering};
 use std::sync::{Arc, Condvar, Mutex, RwLock};
 use std::time::{Duration, Instant};
 use std::{env, fs};
@@ -36,7 +36,6 @@ use utils::id::{TenantId, TimelineId};
 use utils::lsn::Lsn;
 use utils::measured_stream::MeasuredReader;
 use utils::pid_file;
-use utils::shard::{ShardCount, ShardIndex, ShardNumber};

 use crate::configurator::launch_configurator;
 use crate::disk_quota::set_disk_quota;
@@ -70,7 +69,6 @@ pub static BUILD_TAG: Lazy<String> = Lazy::new(|| {
        .unwrap_or(BUILD_TAG_DEFAULT)
        .to_string()
 });
-const DEFAULT_INSTALLED_EXTENSIONS_COLLECTION_INTERVAL: u64 = 3600;

 /// Static configuration params that don't change after startup. These mostly
 /// come from the CLI args, or are derived from them.
@@ -104,7 +102,7 @@ pub struct ComputeNodeParams {
    pub remote_ext_base_url: Option<Url>,

    /// Interval for installed extensions collection
-    pub installed_extensions_collection_interval: Arc<AtomicU64>,
+    pub installed_extensions_collection_interval: u64,
 }

 /// Compute node info shared across several `compute_ctl` threads.
@@ -127,9 +125,6 @@ pub struct ComputeNode {
    // key: ext_archive_name, value: started download time, download_completed?
    pub ext_download_progress: RwLock<HashMap<String, (DateTime<Utc>, bool)>>,
    pub compute_ctl_config: ComputeCtlConfig,
-
-    /// Handle to the extension stats collection task
-    extension_stats_task: Mutex<Option<tokio::task::JoinHandle<()>>>,
 }

 // store some metrics about download size that might impact startup time
@@ -223,8 +218,7 @@ pub struct ParsedSpec {
    pub pageserver_connstr: String,
    pub safekeeper_connstrings: Vec<String>,
    pub storage_auth_token: Option<String>,
-    /// k8s dns name and port
-    pub endpoint_storage_addr: Option<String>,
+    pub endpoint_storage_addr: Option<SocketAddr>,
    pub endpoint_storage_token: Option<String>,
 }

@@ -319,10 +313,13 @@ impl TryFrom<ComputeSpec> for ParsedSpec {
                .or(Err("invalid timeline id"))?
        };

-        let endpoint_storage_addr: Option<String> = spec
+        let endpoint_storage_addr: Option<SocketAddr> = spec
            .endpoint_storage_addr
            .clone()
-            .or_else(|| spec.cluster.settings.find("neon.endpoint_storage_addr"));
+            .or_else(|| spec.cluster.settings.find("neon.endpoint_storage_addr"))
+            .unwrap_or_default()
+            .parse()
+            .ok();
        let endpoint_storage_token = spec
            .endpoint_storage_token
            .clone()
@@ -408,7 +405,7 @@ impl ComputeNode {
        // that can affect `compute_ctl` and prevent it from properly configuring the database schema.
        // Unset them via connection string options before connecting to the database.
        // N.B. keep it in sync with `ZENITH_OPTIONS` in `get_maintenance_client()`.
-        const EXTRA_OPTIONS: &str = "-c role=cloud_admin -c default_transaction_read_only=off -c search_path=public -c statement_timeout=0 -c pgaudit.log=none";
+        const EXTRA_OPTIONS: &str = "-c role=cloud_admin -c default_transaction_read_only=off -c search_path=public -c statement_timeout=0";
        let options = match conn_conf.get_options() {
            // Allow the control plane to override any options set by the
            // compute
@@ -432,7 +429,6 @@ impl ComputeNode {
            state_changed: Condvar::new(),
            ext_download_progress: RwLock::new(HashMap::new()),
            compute_ctl_config: config.compute_ctl_config,
-            extension_stats_task: Mutex::new(None),
        })
    }

@@ -520,9 +516,6 @@ impl ComputeNode {
            None
        };

-        // Terminate the extension stats collection task
-        this.terminate_extension_stats_task();
-
        // Terminate the vm_monitor so it releases the file watcher on
        // /sys/fs/cgroup/neon-postgres.
        // Note: the vm-monitor only runs on linux because it requires cgroups.
@@ -759,15 +752,10 @@ impl ComputeNode {
        // Configure and start rsyslog for compliance audit logging
        match pspec.spec.audit_log_level {
            ComputeAudit::Hipaa | ComputeAudit::Extended | ComputeAudit::Full => {
-                let remote_tls_endpoint =
-                    std::env::var("AUDIT_LOGGING_TLS_ENDPOINT").unwrap_or("".to_string());
-                let remote_plain_endpoint =
+                let remote_endpoint =
                    std::env::var("AUDIT_LOGGING_ENDPOINT").unwrap_or("".to_string());
-
-                if remote_plain_endpoint.is_empty() && remote_tls_endpoint.is_empty() {
-                    anyhow::bail!(
-                        "AUDIT_LOGGING_ENDPOINT and AUDIT_LOGGING_TLS_ENDPOINT are both empty"
-                    );
+                if remote_endpoint.is_empty() {
+                    anyhow::bail!("AUDIT_LOGGING_ENDPOINT is empty");
                }

                let log_directory_path = Path::new(&self.params.pgdata).join("log");
@@ -783,8 +771,7 @@ impl ComputeNode {
                    log_directory_path.clone(),
                    endpoint_id,
                    project_id,
-                    &remote_plain_endpoint,
-                    &remote_tls_endpoint,
+                    &remote_endpoint,
                )?;

                // Launch a background task to clean up the audit logs
@@ -1011,80 +998,13 @@ impl ComputeNode {
        Ok(())
    }

-    /// Fetches a basebackup from the Pageserver using the compute state's Pageserver connstring and
-    /// unarchives it to `pgdata` directory, replacing any existing contents.
+    // Get basebackup from the libpq connection to pageserver using `connstr` and
+    // unarchive it to `pgdata` directory overriding all its previous content.
    #[instrument(skip_all, fields(%lsn))]
    fn try_get_basebackup(&self, compute_state: &ComputeState, lsn: Lsn) -> Result<()> {
        let spec = compute_state.pspec.as_ref().expect("spec must be set");
+        let start_time = Instant::now();

-        let shard0_connstr = spec.pageserver_connstr.split(',').next().unwrap();
-        let started = Instant::now();
-
-        let (connected, size) = match PageserverProtocol::from_connstring(shard0_connstr)? {
-            PageserverProtocol::Libpq => self.try_get_basebackup_libpq(spec, lsn)?,
-            PageserverProtocol::Grpc => self.try_get_basebackup_grpc(spec, lsn)?,
-        };
-
-        let mut state = self.state.lock().unwrap();
-        state.metrics.pageserver_connect_micros =
-            connected.duration_since(started).as_micros() as u64;
-        state.metrics.basebackup_bytes = size as u64;
-        state.metrics.basebackup_ms = started.elapsed().as_millis() as u64;
-
-        Ok(())
-    }
-
-    /// Fetches a basebackup via gRPC. The connstring must use grpc://. Returns the timestamp when
-    /// the connection was established, and the (compressed) size of the basebackup.
-    fn try_get_basebackup_grpc(&self, spec: &ParsedSpec, lsn: Lsn) -> Result<(Instant, usize)> {
-        let shard0_connstr = spec
-            .pageserver_connstr
-            .split(',')
-            .next()
-            .unwrap()
-            .to_string();
-        let shard_index = match spec.pageserver_connstr.split(',').count() as u8 {
-            0 | 1 => ShardIndex::unsharded(),
-            count => ShardIndex::new(ShardNumber(0), ShardCount(count)),
-        };
-
-        let (reader, connected) = tokio::runtime::Handle::current().block_on(async move {
-            let mut client = page_api::Client::new(
-                shard0_connstr,
-                spec.tenant_id,
-                spec.timeline_id,
-                shard_index,
-                spec.storage_auth_token.clone(),
-                None, // NB: base backups use payload compression
-            )
-            .await?;
-            let connected = Instant::now();
-            let reader = client
-                .get_base_backup(page_api::GetBaseBackupRequest {
-                    lsn: (lsn != Lsn(0)).then_some(lsn),
-                    compression: BaseBackupCompression::Gzip,
-                    replica: spec.spec.mode != ComputeMode::Primary,
-                    full: false,
-                })
-                .await?;
-            anyhow::Ok((reader, connected))
-        })?;
-
-        let mut reader = MeasuredReader::new(tokio_util::io::SyncIoBridge::new(reader));
-
-        // Set `ignore_zeros` so that unpack() reads the entire stream and doesn't just stop at the
-        // end-of-archive marker. If the server errors, the tar::Builder drop handler will write an
-        // end-of-archive marker before the error is emitted, and we would not see the error.
-        let mut ar = tar::Archive::new(flate2::read::GzDecoder::new(&mut reader));
-        ar.set_ignore_zeros(true);
-        ar.unpack(&self.params.pgdata)?;
-
-        Ok((connected, reader.get_byte_count()))
-    }
-
-    /// Fetches a basebackup via libpq. The connstring must use postgresql://. Returns the timestamp
-    /// when the connection was established, and the (compressed) size of the basebackup.
-    fn try_get_basebackup_libpq(&self, spec: &ParsedSpec, lsn: Lsn) -> Result<(Instant, usize)> {
        let shard0_connstr = spec.pageserver_connstr.split(',').next().unwrap();
        let mut config = postgres::Config::from_str(shard0_connstr)?;

@@ -1098,14 +1018,16 @@ impl ComputeNode {
        }

        config.application_name("compute_ctl");
-        config.options(&format!(
-            "-c neon.compute_mode={}",
-            spec.spec.mode.to_type_str()
-        ));
+        if let Some(spec) = &compute_state.pspec {
+            config.options(&format!(
+                "-c neon.compute_mode={}",
+                spec.spec.mode.to_type_str()
+            ));
+        }

        // Connect to pageserver
        let mut client = config.connect(NoTls)?;
-        let connected = Instant::now();
+        let pageserver_connect_micros = start_time.elapsed().as_micros() as u64;

        let basebackup_cmd = match lsn {
            Lsn(0) => {
@@ -1142,13 +1064,16 @@ impl ComputeNode {
        // Set `ignore_zeros` so that unpack() reads all the Copy data and
        // doesn't stop at the end-of-archive marker. Otherwise, if the server
        // sends an Error after finishing the tarball, we will not notice it.
-        // The tar::Builder drop handler will write an end-of-archive marker
-        // before emitting the error, and we would not see it otherwise.
        let mut ar = tar::Archive::new(flate2::read::GzDecoder::new(&mut bufreader));
        ar.set_ignore_zeros(true);
        ar.unpack(&self.params.pgdata)?;

-        Ok((connected, measured_reader.get_byte_count()))
+        // Report metrics
+        let mut state = self.state.lock().unwrap();
+        state.metrics.pageserver_connect_micros = pageserver_connect_micros;
+        state.metrics.basebackup_bytes = measured_reader.get_byte_count() as u64;
+        state.metrics.basebackup_ms = start_time.elapsed().as_millis() as u64;
+        Ok(())
    }

    // Gets the basebackup in a retry loop
@@ -1685,8 +1610,6 @@ impl ComputeNode {
            tls_config = self.compute_ctl_config.tls.clone();
        }

-        self.update_installed_extensions_collection_interval(&spec);
-
        let max_concurrent_connections = self.max_service_connections(compute_state, &spec);

        // Merge-apply spec & changes to PostgreSQL state.
@@ -1751,8 +1674,6 @@ impl ComputeNode {

        let tls_config = self.tls_config(&spec);

-        self.update_installed_extensions_collection_interval(&spec);
-
        if let Some(ref pgbouncer_settings) = spec.pgbouncer_settings {
            info!("tuning pgbouncer");

@@ -2357,20 +2278,10 @@ LIMIT 100",
    }

    pub fn spawn_extension_stats_task(&self) {
-        // Cancel any existing task
-        if let Some(handle) = self.extension_stats_task.lock().unwrap().take() {
-            handle.abort();
-        }
-
        let conf = self.tokio_conn_conf.clone();
-        let atomic_interval = self.params.installed_extensions_collection_interval.clone();
-        let mut installed_extensions_collection_interval =
-            2 * atomic_interval.load(std::sync::atomic::Ordering::SeqCst);
-        info!(
-            "[NEON_EXT_SPAWN] Spawning background installed extensions worker with Timeout: {}",
-            installed_extensions_collection_interval
-        );
-        let handle = tokio::spawn(async move {
+        let installed_extensions_collection_interval =
+            self.params.installed_extensions_collection_interval;
+        tokio::spawn(async move {
            // An initial sleep is added to ensure that two collections don't happen at the same time.
            // The first collection happens during compute startup.
            tokio::time::sleep(tokio::time::Duration::from_secs(
@@ -2383,48 +2294,8 @@ LIMIT 100",
            loop {
                interval.tick().await;
                let _ = installed_extensions(conf.clone()).await;
-                // Acquire a read lock on the compute spec and then update the interval if necessary
-                interval = tokio::time::interval(tokio::time::Duration::from_secs(std::cmp::max(
-                    installed_extensions_collection_interval,
-                    2 * atomic_interval.load(std::sync::atomic::Ordering::SeqCst),
-                )));
-                installed_extensions_collection_interval = interval.period().as_secs();
            }
        });
-
-        // Store the new task handle
-        *self.extension_stats_task.lock().unwrap() = Some(handle);
-    }
-
-    fn terminate_extension_stats_task(&self) {
-        if let Some(handle) = self.extension_stats_task.lock().unwrap().take() {
-            handle.abort();
-        }
-    }
-
-    fn update_installed_extensions_collection_interval(&self, spec: &ComputeSpec) {
-        // Update the interval for collecting installed extensions statistics
-        // If the value is -1, we never suspend so set the value to default collection.
-        // If the value is 0, it means default, we will just continue to use the default.
-        if spec.suspend_timeout_seconds == -1 || spec.suspend_timeout_seconds == 0 {
-            info!(
-                "[NEON_EXT_INT_UPD] Spec Timeout: {}, New Timeout: {}",
-                spec.suspend_timeout_seconds, DEFAULT_INSTALLED_EXTENSIONS_COLLECTION_INTERVAL
-            );
-            self.params.installed_extensions_collection_interval.store(
-                DEFAULT_INSTALLED_EXTENSIONS_COLLECTION_INTERVAL,
-                std::sync::atomic::Ordering::SeqCst,
-            );
-        } else {
-            info!(
-                "[NEON_EXT_INT_UPD] Spec Timeout: {}",
-                spec.suspend_timeout_seconds
-            );
-            self.params.installed_extensions_collection_interval.store(
-                spec.suspend_timeout_seconds as u64,
-                std::sync::atomic::Ordering::SeqCst,
-            );
-        }
    }
 }

--- a/compute_tools/src/config_template/compute_audit_rsyslog_template.conf
+++ b/compute_tools/src/config_template/compute_audit_rsyslog_template.conf
@@ -10,13 +10,7 @@ input(type="imfile" File="{log_directory}/*.log"
  startmsg.regex="^[[:digit:]]{{4}}-[[:digit:]]{{2}}-[[:digit:]]{{2}} [[:digit:]]{{2}}:[[:digit:]]{{2}}:[[:digit:]]{{2}}.[[:digit:]]{{3}} GMT,")

 # the directory to store rsyslog state files
-global(
-  workDirectory="/var/log/rsyslog"
-  DefaultNetstreamDriverCAFile="/etc/ssl/certs/ca-certificates.crt"
-)
-
-# Whether the remote syslog receiver uses tls
-set $.remote_syslog_tls = "{remote_syslog_tls}";
+global(workDirectory="/var/log/rsyslog")

 # Construct json, endpoint_id and project_id as additional metadata
 set $.json_log!endpoint_id = "{endpoint_id}";
@@ -27,29 +21,5 @@ set $.json_log!msg = $msg;
 template(name="PgAuditLog" type="string"
    string="<%PRI%>1 %TIMESTAMP:::date-rfc3339% %HOSTNAME% - - - - %$.json_log%")

-# Forward to remote syslog receiver (over TLS)
-if ( $syslogtag == 'pgaudit_log' ) then {{
-  if ( $.remote_syslog_tls == 'true' ) then {{
-    action(type="omfwd" target="{remote_syslog_host}" port="{remote_syslog_port}" protocol="tcp"
-      template="PgAuditLog"
-      queue.type="linkedList"
-      queue.size="1000"
-      action.ResumeRetryCount="10"
-      StreamDriver="gtls"
-      StreamDriverMode="1"
-      StreamDriverAuthMode="x509/name"
-      StreamDriverPermittedPeers="{remote_syslog_host}"
-      StreamDriver.CheckExtendedKeyPurpose="on"
-      StreamDriver.PermitExpiredCerts="off"
-    )
-    stop
-  }} else {{
-    action(type="omfwd" target="{remote_syslog_host}" port="{remote_syslog_port}" protocol="tcp"
-      template="PgAuditLog"
-      queue.type="linkedList"
-      queue.size="1000"
-      action.ResumeRetryCount="10"
-    )
-    stop
-  }}
-}}
+# Forward to remote syslog receiver (@@<hostname>:<port>;format
+local5.info @@{remote_endpoint};PgAuditLog
--- a/compute_tools/src/extension_server.rs
+++ b/compute_tools/src/extension_server.rs
@@ -74,11 +74,9 @@ More specifically, here is an example ext_index.json
 use std::path::Path;
 use std::str;

-use crate::metrics::{REMOTE_EXT_REQUESTS_TOTAL, UNKNOWN_HTTP_STATUS};
 use anyhow::{Context, Result, bail};
 use bytes::Bytes;
 use compute_api::spec::RemoteExtSpec;
-use postgres_versioninfo::PgMajorVersion;
 use regex::Regex;
 use remote_storage::*;
 use reqwest::StatusCode;
@@ -88,6 +86,8 @@ use tracing::log::warn;
 use url::Url;
 use zstd::stream::read::Decoder;

+use crate::metrics::{REMOTE_EXT_REQUESTS_TOTAL, UNKNOWN_HTTP_STATUS};
+
 fn get_pg_config(argument: &str, pgbin: &str) -> String {
    // gives the result of `pg_config [argument]`
    // where argument is a flag like `--version` or `--sharedir`
@@ -106,7 +106,7 @@ fn get_pg_config(argument: &str, pgbin: &str) -> String {
        .to_string()
 }

-pub fn get_pg_version(pgbin: &str) -> PgMajorVersion {
+pub fn get_pg_version(pgbin: &str) -> PostgresMajorVersion {
    // pg_config --version returns a (platform specific) human readable string
    // such as "PostgreSQL 15.4". We parse this to v14/v15/v16 etc.
    let human_version = get_pg_config("--version", pgbin);
@@ -114,11 +114,25 @@ pub fn get_pg_version(pgbin: &str) -> PgMajorVersion {
 }

 pub fn get_pg_version_string(pgbin: &str) -> String {
-    get_pg_version(pgbin).v_str()
+    match get_pg_version(pgbin) {
+        PostgresMajorVersion::V14 => "v14",
+        PostgresMajorVersion::V15 => "v15",
+        PostgresMajorVersion::V16 => "v16",
+        PostgresMajorVersion::V17 => "v17",
+    }
+    .to_owned()
 }

-fn parse_pg_version(human_version: &str) -> PgMajorVersion {
-    use PgMajorVersion::*;
+#[derive(Copy, Clone, Debug, PartialEq, Eq)]
+pub enum PostgresMajorVersion {
+    V14,
+    V15,
+    V16,
+    V17,
+}
+
+fn parse_pg_version(human_version: &str) -> PostgresMajorVersion {
+    use PostgresMajorVersion::*;
    // Normal releases have version strings like "PostgreSQL 15.4". But there
    // are also pre-release versions like "PostgreSQL 17devel" or "PostgreSQL
    // 16beta2" or "PostgreSQL 17rc1". And with the --with-extra-version
@@ -129,10 +143,10 @@ fn parse_pg_version(human_version: &str) -> PgMajorVersion {
        .captures(human_version)
    {
        Some(captures) if captures.len() == 2 => match &captures["major"] {
-            "14" => return PG14,
-            "15" => return PG15,
-            "16" => return PG16,
-            "17" => return PG17,
+            "14" => return V14,
+            "15" => return V15,
+            "16" => return V16,
+            "17" => return V17,
            _ => {}
        },
        _ => {}
@@ -329,25 +343,25 @@ mod tests {

    #[test]
    fn test_parse_pg_version() {
-        use postgres_versioninfo::PgMajorVersion::*;
-        assert_eq!(parse_pg_version("PostgreSQL 15.4"), PG15);
-        assert_eq!(parse_pg_version("PostgreSQL 15.14"), PG15);
+        use super::PostgresMajorVersion::*;
+        assert_eq!(parse_pg_version("PostgreSQL 15.4"), V15);
+        assert_eq!(parse_pg_version("PostgreSQL 15.14"), V15);
        assert_eq!(
            parse_pg_version("PostgreSQL 15.4 (Ubuntu 15.4-0ubuntu0.23.04.1)"),
-            PG15
+            V15
        );

-        assert_eq!(parse_pg_version("PostgreSQL 14.15"), PG14);
-        assert_eq!(parse_pg_version("PostgreSQL 14.0"), PG14);
+        assert_eq!(parse_pg_version("PostgreSQL 14.15"), V14);
+        assert_eq!(parse_pg_version("PostgreSQL 14.0"), V14);
        assert_eq!(
            parse_pg_version("PostgreSQL 14.9 (Debian 14.9-1.pgdg120+1"),
-            PG14
+            V14
        );

-        assert_eq!(parse_pg_version("PostgreSQL 16devel"), PG16);
-        assert_eq!(parse_pg_version("PostgreSQL 16beta1"), PG16);
-        assert_eq!(parse_pg_version("PostgreSQL 16rc2"), PG16);
-        assert_eq!(parse_pg_version("PostgreSQL 16extra"), PG16);
+        assert_eq!(parse_pg_version("PostgreSQL 16devel"), V16);
+        assert_eq!(parse_pg_version("PostgreSQL 16beta1"), V16);
+        assert_eq!(parse_pg_version("PostgreSQL 16rc2"), V16);
+        assert_eq!(parse_pg_version("PostgreSQL 16extra"), V16);
    }

    #[test]
--- a/compute_tools/src/lsn_lease.rs
+++ b/compute_tools/src/lsn_lease.rs
@@ -4,9 +4,7 @@ use std::thread;
 use std::time::{Duration, SystemTime};

 use anyhow::{Result, bail};
-use compute_api::spec::{ComputeMode, PageserverProtocol};
-use itertools::Itertools as _;
-use pageserver_page_api as page_api;
+use compute_api::spec::ComputeMode;
 use postgres::{NoTls, SimpleQueryMessage};
 use tracing::{info, warn};
 use utils::id::{TenantId, TimelineId};
@@ -78,17 +76,25 @@ fn acquire_lsn_lease_with_retry(

    loop {
        // Note: List of pageservers is dynamic, need to re-read configs before each attempt.
-        let (connstrings, auth) = {
+        let configs = {
            let state = compute.state.lock().unwrap();
+
            let spec = state.pspec.as_ref().expect("spec must be set");
-            (
-                spec.pageserver_connstr.clone(),
-                spec.storage_auth_token.clone(),
-            )
+
+            let conn_strings = spec.pageserver_connstr.split(',');
+
+            conn_strings
+                .map(|connstr| {
+                    let mut config = postgres::Config::from_str(connstr).expect("Invalid connstr");
+                    if let Some(storage_auth_token) = &spec.storage_auth_token {
+                        config.password(storage_auth_token.clone());
+                    }
+                    config
+                })
+                .collect::<Vec<_>>()
        };

-        let result =
-            try_acquire_lsn_lease(&connstrings, auth.as_deref(), tenant_id, timeline_id, lsn);
+        let result = try_acquire_lsn_lease(tenant_id, timeline_id, lsn, &configs);
        match result {
            Ok(Some(res)) => {
                return Ok(res);
@@ -110,104 +116,68 @@ fn acquire_lsn_lease_with_retry(
    }
 }

-/// Tries to acquire LSN leases on all Pageserver shards.
+/// Tries to acquire an LSN lease through PS page_service API.
 fn try_acquire_lsn_lease(
-    connstrings: &str,
-    auth: Option<&str>,
    tenant_id: TenantId,
    timeline_id: TimelineId,
    lsn: Lsn,
+    configs: &[postgres::Config],
 ) -> Result<Option<SystemTime>> {
-    let connstrings = connstrings.split(',').collect_vec();
-    let shard_count = connstrings.len();
-    let mut leases = Vec::new();
-
-    for (shard_number, &connstring) in connstrings.iter().enumerate() {
-        let tenant_shard_id = match shard_count {
-            0 | 1 => TenantShardId::unsharded(tenant_id),
-            shard_count => TenantShardId {
-                tenant_id,
-                shard_number: ShardNumber(shard_number as u8),
-                shard_count: ShardCount::new(shard_count as u8),
-            },
+    fn get_valid_until(
+        config: &postgres::Config,
+        tenant_shard_id: TenantShardId,
+        timeline_id: TimelineId,
+        lsn: Lsn,
+    ) -> Result<Option<SystemTime>> {
+        let mut client = config.connect(NoTls)?;
+        let cmd = format!("lease lsn {tenant_shard_id} {timeline_id} {lsn} ");
+        let res = client.simple_query(&cmd)?;
+        let msg = match res.first() {
+            Some(msg) => msg,
+            None => bail!("empty response"),
+        };
+        let row = match msg {
+            SimpleQueryMessage::Row(row) => row,
+            _ => bail!("error parsing lsn lease response"),
        };

-        let lease = match PageserverProtocol::from_connstring(connstring)? {
-            PageserverProtocol::Libpq => {
-                acquire_lsn_lease_libpq(connstring, auth, tenant_shard_id, timeline_id, lsn)?
-            }
-            PageserverProtocol::Grpc => {
-                acquire_lsn_lease_grpc(connstring, auth, tenant_shard_id, timeline_id, lsn)?
-            }
-        };
-        leases.push(lease);
+        // Note: this will be None if a lease is explicitly not granted.
+        let valid_until_str = row.get("valid_until");
+
+        let valid_until = valid_until_str.map(|s| {
+            SystemTime::UNIX_EPOCH
+                .checked_add(Duration::from_millis(u128::from_str(s).unwrap() as u64))
+                .expect("Time larger than max SystemTime could handle")
+        });
+        Ok(valid_until)
    }

-    Ok(leases.into_iter().min().flatten())
-}
+    let shard_count = configs.len();

-/// Acquires an LSN lease on a single shard, using the libpq API. The connstring must use a
-/// postgresql:// scheme.
-fn acquire_lsn_lease_libpq(
-    connstring: &str,
-    auth: Option<&str>,
-    tenant_shard_id: TenantShardId,
-    timeline_id: TimelineId,
-    lsn: Lsn,
-) -> Result<Option<SystemTime>> {
-    let mut config = postgres::Config::from_str(connstring)?;
-    if let Some(auth) = auth {
-        config.password(auth);
-    }
-    let mut client = config.connect(NoTls)?;
-    let cmd = format!("lease lsn {tenant_shard_id} {timeline_id} {lsn} ");
-    let res = client.simple_query(&cmd)?;
-    let msg = match res.first() {
-        Some(msg) => msg,
-        None => bail!("empty response"),
-    };
-    let row = match msg {
-        SimpleQueryMessage::Row(row) => row,
-        _ => bail!("error parsing lsn lease response"),
+    let valid_until = if shard_count > 1 {
+        configs
+            .iter()
+            .enumerate()
+            .map(|(shard_number, config)| {
+                let tenant_shard_id = TenantShardId {
+                    tenant_id,
+                    shard_count: ShardCount::new(shard_count as u8),
+                    shard_number: ShardNumber(shard_number as u8),
+                };
+                get_valid_until(config, tenant_shard_id, timeline_id, lsn)
+            })
+            .collect::<Result<Vec<Option<SystemTime>>>>()?
+            .into_iter()
+            .min()
+            .unwrap()
+    } else {
+        get_valid_until(
+            &configs[0],
+            TenantShardId::unsharded(tenant_id),
+            timeline_id,
+            lsn,
+        )?
    };

-    // Note: this will be None if a lease is explicitly not granted.
-    let valid_until_str = row.get("valid_until");
-
-    let valid_until = valid_until_str.map(|s| {
-        SystemTime::UNIX_EPOCH
-            .checked_add(Duration::from_millis(u128::from_str(s).unwrap() as u64))
-            .expect("Time larger than max SystemTime could handle")
-    });
    Ok(valid_until)
 }
-
-/// Acquires an LSN lease on a single shard, using the gRPC API. The connstring must use a
-/// grpc:// scheme.
-fn acquire_lsn_lease_grpc(
-    connstring: &str,
-    auth: Option<&str>,
-    tenant_shard_id: TenantShardId,
-    timeline_id: TimelineId,
-    lsn: Lsn,
-) -> Result<Option<SystemTime>> {
-    tokio::runtime::Handle::current().block_on(async move {
-        let mut client = page_api::Client::new(
-            connstring.to_string(),
-            tenant_shard_id.tenant_id,
-            timeline_id,
-            tenant_shard_id.to_index(),
-            auth.map(String::from),
-            None,
-        )
-        .await?;
-
-        let req = page_api::LeaseLsnRequest { lsn };
-        match client.lease_lsn(req).await {
-            Ok(expires) => Ok(Some(expires)),
-            // Lease couldn't be acquired because the LSN has been garbage collected.
-            Err(err) if err.code() == tonic::Code::FailedPrecondition => Ok(None),
-            Err(err) => Err(err.into()),
-        }
-    })
-}
--- a/compute_tools/src/rsyslog.rs
+++ b/compute_tools/src/rsyslog.rs
@@ -4,10 +4,8 @@ use std::path::Path;
 use std::process::Command;
 use std::time::Duration;
 use std::{fs::OpenOptions, io::Write};
-use url::{Host, Url};

 use anyhow::{Context, Result, anyhow};
-use hostname_validator;
 use tracing::{error, info, instrument, warn};

 const POSTGRES_LOGS_CONF_PATH: &str = "/etc/rsyslog.d/postgres_logs.conf";
@@ -84,84 +82,18 @@ fn restart_rsyslog() -> Result<()> {
    Ok(())
 }

-fn parse_audit_syslog_address(
-    remote_plain_endpoint: &str,
-    remote_tls_endpoint: &str,
-) -> Result<(String, u16, String)> {
-    let tls;
-    let remote_endpoint = if !remote_tls_endpoint.is_empty() {
-        tls = "true".to_string();
-        remote_tls_endpoint
-    } else {
-        tls = "false".to_string();
-        remote_plain_endpoint
-    };
-    // Urlify the remote_endpoint, so parsing can be done with url::Url.
-    let url_str = format!("http://{remote_endpoint}");
-    let url = Url::parse(&url_str).map_err(|err| {
-        anyhow!("Error parsing {remote_endpoint}, expected host:port, got {err:?}")
-    })?;
-
-    let is_valid = url.scheme() == "http"
-        && url.path() == "/"
-        && url.query().is_none()
-        && url.fragment().is_none()
-        && url.username() == ""
-        && url.password().is_none();
-
-    if !is_valid {
-        return Err(anyhow!(
-            "Invalid address format {remote_endpoint}, expected host:port"
-        ));
-    }
-    let host = match url.host() {
-        Some(Host::Domain(h)) if hostname_validator::is_valid(h) => h.to_string(),
-        Some(Host::Ipv4(ip4)) => ip4.to_string(),
-        Some(Host::Ipv6(ip6)) => ip6.to_string(),
-        _ => return Err(anyhow!("Invalid host")),
-    };
-    let port = url
-        .port()
-        .ok_or_else(|| anyhow!("Invalid port in {remote_endpoint}"))?;
-
-    Ok((host, port, tls))
-}
-
-fn generate_audit_rsyslog_config(
-    log_directory: String,
-    endpoint_id: &str,
-    project_id: &str,
-    remote_syslog_host: &str,
-    remote_syslog_port: u16,
-    remote_syslog_tls: &str,
-) -> String {
-    format!(
-        include_str!("config_template/compute_audit_rsyslog_template.conf"),
-        log_directory = log_directory,
-        endpoint_id = endpoint_id,
-        project_id = project_id,
-        remote_syslog_host = remote_syslog_host,
-        remote_syslog_port = remote_syslog_port,
-        remote_syslog_tls = remote_syslog_tls
-    )
-}
-
 pub fn configure_audit_rsyslog(
    log_directory: String,
    endpoint_id: &str,
    project_id: &str,
    remote_endpoint: &str,
-    remote_tls_endpoint: &str,
 ) -> Result<()> {
-    let (remote_syslog_host, remote_syslog_port, remote_syslog_tls) =
-        parse_audit_syslog_address(remote_endpoint, remote_tls_endpoint).unwrap();
-    let config_content = generate_audit_rsyslog_config(
-        log_directory,
-        endpoint_id,
-        project_id,
-        &remote_syslog_host,
-        remote_syslog_port,
-        &remote_syslog_tls,
+    let config_content: String = format!(
+        include_str!("config_template/compute_audit_rsyslog_template.conf"),
+        log_directory = log_directory,
+        endpoint_id = endpoint_id,
+        project_id = project_id,
+        remote_endpoint = remote_endpoint
    );

    info!("rsyslog config_content: {}", config_content);
@@ -326,8 +258,6 @@ pub fn launch_pgaudit_gc(log_directory: String) {
 mod tests {
    use crate::rsyslog::PostgresLogsRsyslogConfig;

-    use super::{generate_audit_rsyslog_config, parse_audit_syslog_address};
-
    #[test]
    fn test_postgres_logs_config() {
        {
@@ -357,146 +287,4 @@ mod tests {
            assert!(res.is_err());
        }
    }
-
-    #[test]
-    fn test_parse_audit_syslog_address() {
-        {
-            // host:port format (plaintext)
-            let parsed = parse_audit_syslog_address("collector.host.tld:5555", "");
-            assert!(parsed.is_ok());
-            assert_eq!(
-                parsed.unwrap(),
-                (
-                    String::from("collector.host.tld"),
-                    5555,
-                    String::from("false")
-                )
-            );
-        }
-
-        {
-            // host:port format with ipv4 ip address (plaintext)
-            let parsed = parse_audit_syslog_address("10.0.0.1:5555", "");
-            assert!(parsed.is_ok());
-            assert_eq!(
-                parsed.unwrap(),
-                (String::from("10.0.0.1"), 5555, String::from("false"))
-            );
-        }
-
-        {
-            // host:port format with ipv6 ip address (plaintext)
-            let parsed =
-                parse_audit_syslog_address("[7e60:82ed:cb2e:d617:f904:f395:aaca:e252]:5555", "");
-            assert_eq!(
-                parsed.unwrap(),
-                (
-                    String::from("7e60:82ed:cb2e:d617:f904:f395:aaca:e252"),
-                    5555,
-                    String::from("false")
-                )
-            );
-        }
-
-        {
-            // Only TLS host:port defined
-            let parsed = parse_audit_syslog_address("", "tls.host.tld:5556");
-            assert_eq!(
-                parsed.unwrap(),
-                (String::from("tls.host.tld"), 5556, String::from("true"))
-            );
-        }
-
-        {
-            // tls host should take precedence, when both defined
-            let parsed = parse_audit_syslog_address("plaintext.host.tld:5555", "tls.host.tld:5556");
-            assert_eq!(
-                parsed.unwrap(),
-                (String::from("tls.host.tld"), 5556, String::from("true"))
-            );
-        }
-
-        {
-            // host without port (plaintext)
-            let parsed = parse_audit_syslog_address("collector.host.tld", "");
-            assert!(parsed.is_err());
-        }
-
-        {
-            // port without host
-            let parsed = parse_audit_syslog_address(":5555", "");
-            assert!(parsed.is_err());
-        }
-
-        {
-            // valid host with invalid port
-            let parsed = parse_audit_syslog_address("collector.host.tld:90001", "");
-            assert!(parsed.is_err());
-        }
-
-        {
-            // invalid hostname with valid port
-            let parsed = parse_audit_syslog_address("-collector.host.tld:5555", "");
-            assert!(parsed.is_err());
-        }
-
-        {
-            // parse error
-            let parsed = parse_audit_syslog_address("collector.host.tld:::5555", "");
-            assert!(parsed.is_err());
-        }
-    }
-
-    #[test]
-    fn test_generate_audit_rsyslog_config() {
-        {
-            // plaintext version
-            let log_directory = "/tmp/log".to_string();
-            let endpoint_id = "ep-test-endpoint-id";
-            let project_id = "test-project-id";
-            let remote_syslog_host = "collector.host.tld";
-            let remote_syslog_port = 5555;
-            let remote_syslog_tls = "false";
-
-            let conf_str = generate_audit_rsyslog_config(
-                log_directory,
-                endpoint_id,
-                project_id,
-                remote_syslog_host,
-                remote_syslog_port,
-                remote_syslog_tls,
-            );
-
-            assert!(conf_str.contains(r#"set $.remote_syslog_tls = "false";"#));
-            assert!(conf_str.contains(r#"type="omfwd""#));
-            assert!(conf_str.contains(r#"target="collector.host.tld""#));
-            assert!(conf_str.contains(r#"port="5555""#));
-            assert!(conf_str.contains(r#"StreamDriverPermittedPeers="collector.host.tld""#));
-        }
-
-        {
-            // TLS version
-            let log_directory = "/tmp/log".to_string();
-            let endpoint_id = "ep-test-endpoint-id";
-            let project_id = "test-project-id";
-            let remote_syslog_host = "collector.host.tld";
-            let remote_syslog_port = 5556;
-            let remote_syslog_tls = "true";
-
-            let conf_str = generate_audit_rsyslog_config(
-                log_directory,
-                endpoint_id,
-                project_id,
-                remote_syslog_host,
-                remote_syslog_port,
-                remote_syslog_tls,
-            );
-
-            assert!(conf_str.contains(r#"set $.remote_syslog_tls = "true";"#));
-            assert!(conf_str.contains(r#"type="omfwd""#));
-            assert!(conf_str.contains(r#"target="collector.host.tld""#));
-            assert!(conf_str.contains(r#"port="5556""#));
-            assert!(conf_str.contains(r#"StreamDriverPermittedPeers="collector.host.tld""#));
-        }
-    }
 }
--- a/compute_tools/tests/cluster_spec.json
+++ b/compute_tools/tests/cluster_spec.json
@@ -3,8 +3,7 @@

  "timestamp": "2021-05-23T18:25:43.511Z",
  "operation_uuid": "0f657b36-4b0f-4a2d-9c2e-1dcd615e7d8b",
-  "suspend_timeout_seconds": 3600,
-  
+
  "cluster": {
    "cluster_id": "test-cluster-42",
    "name": "Zenith Test",
--- a/control_plane/src/bin/neon_local.rs
+++ b/control_plane/src/bin/neon_local.rs
@@ -16,9 +16,9 @@ use std::time::Duration;
 use anyhow::{Context, Result, anyhow, bail};
 use clap::Parser;
 use compute_api::requests::ComputeClaimsScope;
-use compute_api::spec::{ComputeMode, PageserverProtocol};
+use compute_api::spec::ComputeMode;
 use control_plane::broker::StorageBroker;
-use control_plane::endpoint::{ComputeControlPlane, EndpointTerminateMode};
+use control_plane::endpoint::{ComputeControlPlane, EndpointTerminateMode, PageserverProtocol};
 use control_plane::endpoint_storage::{ENDPOINT_STORAGE_DEFAULT_ADDR, EndpointStorage};
 use control_plane::local_env;
 use control_plane::local_env::{
@@ -48,7 +48,7 @@ use postgres_connection::parse_host_port;
 use safekeeper_api::membership::{SafekeeperGeneration, SafekeeperId};
 use safekeeper_api::{
    DEFAULT_HTTP_LISTEN_PORT as DEFAULT_SAFEKEEPER_HTTP_PORT,
-    DEFAULT_PG_LISTEN_PORT as DEFAULT_SAFEKEEPER_PG_PORT, PgMajorVersion, PgVersionId,
+    DEFAULT_PG_LISTEN_PORT as DEFAULT_SAFEKEEPER_PG_PORT,
 };
 use storage_broker::DEFAULT_LISTEN_ADDR as DEFAULT_BROKER_ADDR;
 use tokio::task::JoinSet;
@@ -64,7 +64,7 @@ const DEFAULT_PAGESERVER_ID: NodeId = NodeId(1);
 const DEFAULT_BRANCH_NAME: &str = "main";
 project_git_version!(GIT_VERSION);

-const DEFAULT_PG_VERSION: PgMajorVersion = PgMajorVersion::PG17;
+const DEFAULT_PG_VERSION: u32 = 17;

 const DEFAULT_PAGESERVER_CONTROL_PLANE_API: &str = "http://127.0.0.1:1234/upcall/v1/";

@@ -169,7 +169,7 @@ struct TenantCreateCmdArgs {

    #[arg(default_value_t = DEFAULT_PG_VERSION)]
    #[clap(long, help = "Postgres version to use for the initial timeline")]
-    pg_version: PgMajorVersion,
+    pg_version: u32,

    #[clap(
        long,
@@ -292,7 +292,7 @@ struct TimelineCreateCmdArgs {

    #[arg(default_value_t = DEFAULT_PG_VERSION)]
    #[clap(long, help = "Postgres version")]
-    pg_version: PgMajorVersion,
+    pg_version: u32,
 }

 #[derive(clap::Args)]
@@ -324,7 +324,7 @@ struct TimelineImportCmdArgs {

    #[arg(default_value_t = DEFAULT_PG_VERSION)]
    #[clap(long, help = "Postgres version of the backup being imported")]
-    pg_version: PgMajorVersion,
+    pg_version: u32,
 }

 #[derive(clap::Subcommand)]
@@ -603,7 +603,7 @@ struct EndpointCreateCmdArgs {

    #[arg(default_value_t = DEFAULT_PG_VERSION)]
    #[clap(long, help = "Postgres version")]
-    pg_version: PgMajorVersion,
+    pg_version: u32,

    /// Use gRPC to communicate with Pageservers, by generating grpc:// connstrings.
    ///
@@ -1295,7 +1295,7 @@ async fn handle_timeline(cmd: &TimelineCmd, env: &mut local_env::LocalEnv) -> Re
                    },
                    new_members: None,
                };
-                let pg_version = PgVersionId::from(args.pg_version);
+                let pg_version = args.pg_version * 10000;
                let req = safekeeper_api::models::TimelineCreateRequest {
                    tenant_id,
                    timeline_id,
@@ -1649,9 +1649,7 @@ async fn handle_endpoint(subcmd: &EndpointCmd, env: &local_env::LocalEnv) -> Res
            // If --safekeepers argument is given, use only the listed
            // safekeeper nodes; otherwise all from the env.
            let safekeepers = parse_safekeepers(&args.safekeepers)?;
-            endpoint
-                .reconfigure(Some(pageservers), None, safekeepers, None)
-                .await?;
+            endpoint.reconfigure(pageservers, None, safekeepers).await?;
        }
        EndpointCmd::Stop(args) => {
            let endpoint_id = &args.endpoint_id;
--- a/control_plane/src/endpoint.rs
+++ b/control_plane/src/endpoint.rs
@@ -56,8 +56,8 @@ use compute_api::responses::{
    TlsConfig,
 };
 use compute_api::spec::{
-    Cluster, ComputeAudit, ComputeFeature, ComputeMode, ComputeSpec, Database, PageserverProtocol,
-    PgIdent, RemoteExtSpec, Role,
+    Cluster, ComputeAudit, ComputeFeature, ComputeMode, ComputeSpec, Database, PgIdent,
+    RemoteExtSpec, Role,
 };
 use jsonwebtoken::jwk::{
    AlgorithmParameters, CommonParameters, EllipticCurve, Jwk, JwkSet, KeyAlgorithm, KeyOperations,
@@ -67,7 +67,6 @@ use nix::sys::signal::{Signal, kill};
 use pageserver_api::shard::ShardStripeSize;
 use pem::Pem;
 use reqwest::header::CONTENT_TYPE;
-use safekeeper_api::PgMajorVersion;
 use safekeeper_api::membership::SafekeeperGeneration;
 use serde::{Deserialize, Serialize};
 use sha2::{Digest, Sha256};
@@ -90,7 +89,7 @@ pub struct EndpointConf {
    pg_port: u16,
    external_http_port: u16,
    internal_http_port: u16,
-    pg_version: PgMajorVersion,
+    pg_version: u32,
    grpc: bool,
    skip_pg_catalog_updates: bool,
    reconfigure_concurrency: usize,
@@ -193,7 +192,7 @@ impl ComputeControlPlane {
        pg_port: Option<u16>,
        external_http_port: Option<u16>,
        internal_http_port: Option<u16>,
-        pg_version: PgMajorVersion,
+        pg_version: u32,
        mode: ComputeMode,
        grpc: bool,
        skip_pg_catalog_updates: bool,
@@ -313,7 +312,7 @@ pub struct Endpoint {
    pub internal_http_address: SocketAddr,

    // postgres major version in the format: 14, 15, etc.
-    pg_version: PgMajorVersion,
+    pg_version: u32,

    // These are not part of the endpoint as such, but the environment
    // the endpoint runs in.
@@ -373,6 +372,29 @@ impl std::fmt::Display for EndpointTerminateMode {
    }
 }

+/// Protocol used to connect to a Pageserver.
+#[derive(Clone, Copy, Debug)]
+pub enum PageserverProtocol {
+    Libpq,
+    Grpc,
+}
+
+impl PageserverProtocol {
+    /// Returns the URL scheme for the protocol, used in connstrings.
+    pub fn scheme(&self) -> &'static str {
+        match self {
+            Self::Libpq => "postgresql",
+            Self::Grpc => "grpc",
+        }
+    }
+}
+
+impl Display for PageserverProtocol {
+    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
+        f.write_str(self.scheme())
+    }
+}
+
 impl Endpoint {
    fn from_dir_entry(entry: std::fs::DirEntry, env: &LocalEnv) -> Result<Endpoint> {
        if !entry.file_type()?.is_dir() {
@@ -535,7 +557,7 @@ impl Endpoint {
                conf.append("hot_standby", "on");
                // prefetching of blocks referenced in WAL doesn't make sense for us
                // Neon hot standby ignores pages that are not in the shared_buffers
-                if self.pg_version >= PgMajorVersion::PG15 {
+                if self.pg_version >= 15 {
                    conf.append("recovery_prefetch", "off");
                }
            }
@@ -780,7 +802,6 @@ impl Endpoint {
                endpoint_storage_addr: Some(endpoint_storage_addr),
                endpoint_storage_token: Some(endpoint_storage_token),
                autoprewarm: false,
-                suspend_timeout_seconds: -1, // Only used in neon_local.
            };

            // this strange code is needed to support respec() in tests
@@ -975,11 +996,12 @@ impl Endpoint {

    pub async fn reconfigure(
        &self,
-        pageservers: Option<Vec<(PageserverProtocol, Host, u16)>>,
+        pageservers: Vec<(PageserverProtocol, Host, u16)>,
        stripe_size: Option<ShardStripeSize>,
        safekeepers: Option<Vec<NodeId>>,
-        safekeeper_generation: Option<SafekeeperGeneration>,
    ) -> Result<()> {
+        anyhow::ensure!(!pageservers.is_empty(), "no pageservers provided");
+
        let (mut spec, compute_ctl_config) = {
            let config_path = self.endpoint_path().join("config.json");
            let file = std::fs::File::open(config_path)?;
@@ -991,24 +1013,16 @@ impl Endpoint {
        let postgresql_conf = self.read_postgresql_conf()?;
        spec.cluster.postgresql_conf = Some(postgresql_conf);

-        // If pageservers are not specified, don't change them.
-        if let Some(pageservers) = pageservers {
-            anyhow::ensure!(!pageservers.is_empty(), "no pageservers provided");
-
-            let pageserver_connstr = Self::build_pageserver_connstr(&pageservers);
-            spec.pageserver_connstring = Some(pageserver_connstr);
-            if stripe_size.is_some() {
-                spec.shard_stripe_size = stripe_size.map(|s| s.0 as usize);
-            }
+        let pageserver_connstr = Self::build_pageserver_connstr(&pageservers);
+        spec.pageserver_connstring = Some(pageserver_connstr);
+        if stripe_size.is_some() {
+            spec.shard_stripe_size = stripe_size.map(|s| s.0 as usize);
        }

        // If safekeepers are not specified, don't change them.
        if let Some(safekeepers) = safekeepers {
            let safekeeper_connstrings = self.build_safekeepers_connstrs(safekeepers)?;
            spec.safekeeper_connstrings = safekeeper_connstrings;
-            if let Some(g) = safekeeper_generation {
-                spec.safekeepers_generation = Some(g.into_inner());
-            }
        }

        let client = reqwest::Client::builder()
@@ -1046,24 +1060,6 @@ impl Endpoint {
        }
    }

-    pub async fn reconfigure_pageservers(
-        &self,
-        pageservers: Vec<(PageserverProtocol, Host, u16)>,
-        stripe_size: Option<ShardStripeSize>,
-    ) -> Result<()> {
-        self.reconfigure(Some(pageservers), stripe_size, None, None)
-            .await
-    }
-
-    pub async fn reconfigure_safekeepers(
-        &self,
-        safekeepers: Vec<NodeId>,
-        generation: SafekeeperGeneration,
-    ) -> Result<()> {
-        self.reconfigure(None, None, Some(safekeepers), Some(generation))
-            .await
-    }
-
    pub async fn stop(
        &self,
        mode: EndpointTerminateMode,
--- a/control_plane/src/local_env.rs
+++ b/control_plane/src/local_env.rs
@@ -12,11 +12,9 @@ use std::{env, fs};

 use anyhow::{Context, bail};
 use clap::ValueEnum;
-use pageserver_api::config::PostHogConfig;
 use pem::Pem;
 use postgres_backend::AuthType;
 use reqwest::{Certificate, Url};
-use safekeeper_api::PgMajorVersion;
 use serde::{Deserialize, Serialize};
 use utils::auth::encode_from_key_file;
 use utils::id::{NodeId, TenantId, TenantTimelineId, TimelineId};
@@ -212,9 +210,7 @@ pub struct NeonStorageControllerConf {

    pub use_local_compute_notifications: bool,

-    pub timeline_safekeeper_count: Option<usize>,
-
-    pub posthog_config: Option<PostHogConfig>,
+    pub timeline_safekeeper_count: Option<i64>,

    pub kick_secondary_downloads: Option<bool>,
 }
@@ -248,7 +244,6 @@ impl Default for NeonStorageControllerConf {
            use_https_safekeeper_api: false,
            use_local_compute_notifications: true,
            timeline_safekeeper_count: None,
-            posthog_config: None,
            kick_secondary_downloads: None,
        }
    }
@@ -429,21 +424,25 @@ impl LocalEnv {
        self.pg_distrib_dir.clone()
    }

-    pub fn pg_distrib_dir(&self, pg_version: PgMajorVersion) -> anyhow::Result<PathBuf> {
+    pub fn pg_distrib_dir(&self, pg_version: u32) -> anyhow::Result<PathBuf> {
        let path = self.pg_distrib_dir.clone();

-        Ok(path.join(pg_version.v_str()))
+        #[allow(clippy::manual_range_patterns)]
+        match pg_version {
+            14 | 15 | 16 | 17 => Ok(path.join(format!("v{pg_version}"))),
+            _ => bail!("Unsupported postgres version: {}", pg_version),
+        }
    }

-    pub fn pg_dir(&self, pg_version: PgMajorVersion, dir_name: &str) -> anyhow::Result<PathBuf> {
+    pub fn pg_dir(&self, pg_version: u32, dir_name: &str) -> anyhow::Result<PathBuf> {
        Ok(self.pg_distrib_dir(pg_version)?.join(dir_name))
    }

-    pub fn pg_bin_dir(&self, pg_version: PgMajorVersion) -> anyhow::Result<PathBuf> {
+    pub fn pg_bin_dir(&self, pg_version: u32) -> anyhow::Result<PathBuf> {
        self.pg_dir(pg_version, "bin")
    }

-    pub fn pg_lib_dir(&self, pg_version: PgMajorVersion) -> anyhow::Result<PathBuf> {
+    pub fn pg_lib_dir(&self, pg_version: u32) -> anyhow::Result<PathBuf> {
        self.pg_dir(pg_version, "lib")
    }

--- a/control_plane/src/pageserver.rs
+++ b/control_plane/src/pageserver.rs
@@ -22,7 +22,6 @@ use pageserver_api::shard::TenantShardId;
 use pageserver_client::mgmt_api;
 use postgres_backend::AuthType;
 use postgres_connection::{PgConnectionConfig, parse_host_port};
-use safekeeper_api::PgMajorVersion;
 use utils::auth::{Claims, Scope};
 use utils::id::{NodeId, TenantId, TimelineId};
 use utils::lsn::Lsn;
@@ -608,7 +607,7 @@ impl PageServerNode {
        timeline_id: TimelineId,
        base: (Lsn, PathBuf),
        pg_wal: Option<(Lsn, PathBuf)>,
-        pg_version: PgMajorVersion,
+        pg_version: u32,
    ) -> anyhow::Result<()> {
        // Init base reader
        let (start_lsn, base_tarfile_path) = base;
--- a/control_plane/src/storage_controller.rs
+++ b/control_plane/src/storage_controller.rs
@@ -6,8 +6,6 @@ use std::str::FromStr;
 use std::sync::OnceLock;
 use std::time::{Duration, Instant};

-use crate::background_process;
-use crate::local_env::{LocalEnv, NeonStorageControllerConf};
 use camino::{Utf8Path, Utf8PathBuf};
 use hyper0::Uri;
 use nix::unistd::Pid;
@@ -24,7 +22,6 @@ use pageserver_client::mgmt_api::ResponseErrorMessageExt;
 use pem::Pem;
 use postgres_backend::AuthType;
 use reqwest::{Method, Response};
-use safekeeper_api::PgMajorVersion;
 use serde::de::DeserializeOwned;
 use serde::{Deserialize, Serialize};
 use tokio::process::Command;
@@ -34,6 +31,9 @@ use utils::auth::{Claims, Scope, encode_from_key_file};
 use utils::id::{NodeId, TenantId};
 use whoami::username;

+use crate::background_process;
+use crate::local_env::{LocalEnv, NeonStorageControllerConf};
+
 pub struct StorageController {
    env: LocalEnv,
    private_key: Option<Pem>,
@@ -48,7 +48,7 @@ pub struct StorageController {

 const COMMAND: &str = "storage_controller";

-const STORAGE_CONTROLLER_POSTGRES_VERSION: PgMajorVersion = PgMajorVersion::PG16;
+const STORAGE_CONTROLLER_POSTGRES_VERSION: u32 = 16;

 const DB_NAME: &str = "storage_controller";

@@ -184,15 +184,9 @@ impl StorageController {
    /// to other versions if that one isn't found.  Some automated tests create circumstances
    /// where only one version is available in pg_distrib_dir, such as `test_remote_extensions`.
    async fn get_pg_dir(&self, dir_name: &str) -> anyhow::Result<Utf8PathBuf> {
-        const PREFER_VERSIONS: [PgMajorVersion; 5] = [
-            STORAGE_CONTROLLER_POSTGRES_VERSION,
-            PgMajorVersion::PG16,
-            PgMajorVersion::PG15,
-            PgMajorVersion::PG14,
-            PgMajorVersion::PG17,
-        ];
+        let prefer_versions = [STORAGE_CONTROLLER_POSTGRES_VERSION, 16, 15, 14];

-        for v in PREFER_VERSIONS {
+        for v in prefer_versions {
            let path = Utf8PathBuf::from_path_buf(self.env.pg_dir(v, dir_name)?).unwrap();
            if tokio::fs::try_exists(&path).await? {
                return Ok(path);
@@ -638,28 +632,10 @@ impl StorageController {
            args.push("--timelines-onto-safekeepers".to_string());
        }

-        // neon_local is used in test environments where we often have less than 3 safekeepers.
-        if self.config.timeline_safekeeper_count.is_some() || self.env.safekeepers.len() < 3 {
-            let sk_cnt = self
-                .config
-                .timeline_safekeeper_count
-                .unwrap_or(self.env.safekeepers.len());
-
+        if let Some(sk_cnt) = self.config.timeline_safekeeper_count {
            args.push(format!("--timeline-safekeeper-count={sk_cnt}"));
        }

-        let mut envs = vec![
-            ("LD_LIBRARY_PATH".to_owned(), pg_lib_dir.to_string()),
-            ("DYLD_LIBRARY_PATH".to_owned(), pg_lib_dir.to_string()),
-        ];
-
-        if let Some(posthog_config) = &self.config.posthog_config {
-            envs.push((
-                "POSTHOG_CONFIG".to_string(),
-                serde_json::to_string(posthog_config)?,
-            ));
-        }
-
        println!("Starting storage controller");

        background_process::start_process(
@@ -667,7 +643,10 @@ impl StorageController {
            &instance_dir,
            &self.env.storage_controller_bin(),
            args,
-            envs,
+            vec![
+                ("LD_LIBRARY_PATH".to_owned(), pg_lib_dir.to_string()),
+                ("DYLD_LIBRARY_PATH".to_owned(), pg_lib_dir.to_string()),
+            ],
            background_process::InitialPidFile::Create(self.pid_file(start_args.instance_id)),
            &start_args.start_timeout,
            || async {
--- a/docker-compose/compute_wrapper/var/db/postgres/configs/config.json
+++ b/docker-compose/compute_wrapper/var/db/postgres/configs/config.json
@@ -4,7 +4,6 @@

        "timestamp": "2022-10-12T18:00:00.000Z",
        "operation_uuid": "0f657b36-4b0f-4a2d-9c2e-1dcd615e7d8c",
-        "suspend_timeout_seconds": -1,

        "cluster": {
            "cluster_id": "docker_compose",
--- a/docs/rfcs/040-Endpoint-Persistent-Unlogged-Files-Storage.md
+++ b/docs/rfcs/040-Endpoint-Persistent-Unlogged-Files-Storage.md
@@ -1,396 +0,0 @@
-# Memo: Endpoint Persistent Unlogged Files Storage
-Created on 2024-11-05
-Implemented on N/A
-
-## Summary
-A design for a storage system that allows storage of files required to make
-Neon's Endpoints have a better experience at or after a reboot.
-
-## Motivation
-Several systems inside PostgreSQL (and Neon) need some persistent storage for
-optimal workings across reboots and restarts, but still work without.
-Examples are the query-level statistics files of `pg_stat_statements` in
-`pg_stat/pg_stat_statements.stat`, and `pg_prewarm`'s `autoprewarm.blocks`.
-We need a storage system that can store and manage these files for each
-Endpoint, without necessarily granting users access to an unlimited storage
-device.
-
-## Goals
- Store known files for Endpoints with reasonable persistence.  
-  _Data loss in this service, while annoying and bad for UX, won't lose any
-  customer's data._
-
-## Non Goals (if relevant)
- This storage system does not need branching, file versioning, or other such
-  features. The files are as ephemeral to the timeline of the data as the
-  Endpoints that host the data.
- This storage system does not need to store _all_ user files, only 'known'
-  user files.
- This storage system does not need to be hosted fully inside Computes.  
-  _Instead, this will be a separate component similar to Pageserver,
-  SafeKeeper, the S3 proxy used for dynamically loaded extensions, etc._
-
-## Impacted components
- Compute needs new code to load and store these files in its lifetime.
- Control Plane needs to consider this new storage system when signalling
-  the deletion of an Endpoint, Timeline, or Tenant.
- Control Plane needs to consider this new storage system when it resets
-  or re-assigns an endpoint's timeline/branch state.
-
-A new service is created: the Endpoint Persistent Unlogged Files Storage
-service.  This could be integrated in e.g. Pageserver or Control Plane, or a
-separately hosted service.
-
-## Proposed implementation
-Endpoint-related data files are managed by a newly designed service (which
-optionally is integrated in an existing service like Pageserver or Control
-Plane), which stores data directly into S3 or any blob storage of choice.
-
-Upon deletion of the Endpoint, or reassignment of the endpoint to a different
-branch, this ephemeral data is dropped: the data stored may not match the
-state of the branch's data after reassignment, and on endpoint deletion the
-data won't have any use to the user.
-
-Compute gets credentials (JWT token with Tenant, Timeline & Endpoint claims)
-which it can use to authenticate to this new service and retrieve and store
-data associated with this endpoint.  This limited scope reduces leaks of data
-across endpoints and timeline resets, and limits the ability of endpoints to
-mess with other endpoints' data.
-
-The path of this endpoint data in S3 is initially as follows:
-
-    s3://<regional-epufs-bucket>/
-      tenants/
-        <hex-tenant-id>/
-          tenants/
-            <hex-timeline-id>/
-              endpoints/
-                <endpoint-id>/
-                  pgdata/
-                    <file_path_in_pgdatadir>
-
-For other blob storages an equivalent or similar path can be constructed.
-
-### Reliability, failure modes and corner cases (if relevant)
-Reliability is important, but not critical to the workings of Neon.  The data
-stored in this service will, when lost, reduce performance, but won't be a
-cause of permanent data loss - only operational metadata is stored.
-
-Most, if not all, blob storage services have sufficiently high persistence
-guarantees to cater our need for persistence and uptime. The only concern with
-blob storages is that the access latency is generally higher than local disk,
-but for the object types stored (cache state, ...) I don't think this will be
-much of an issue.
-
-### Interaction/Sequence diagram (if relevant)
-
-In these diagrams you can replace S3 with any persistent storage device of
-choice, but S3 is chosen as representative name: The well-known and short name
-of AWS' blob storage. Azure Blob Storage should work too, but it has a much
-longer name making it less practical for the diagrams.
-
-Write data:
-
-```http
-POST /tenants/<tenant-id>/timelines/<tl-id>/endpoints/<endpoint-id>/pgdata/<the-pgdata-path>
-Host: epufs.svc.neon.local
-
-<<<
-
-200 OK
-{
-  "version": "<opaque>", # opaque file version token, changes when the file contents change
-  "size": <bytes>,
-}
-```
-
-```mermaid
-sequenceDiagram
-    autonumber
-    participant co as Compute
-    participant ep as EPUFS
-    participant s3 as Blob Storage
-
-    co-->ep: Connect with credentials
-    co->>+ep: Store Unlogged Persistent File
-    opt is authenticated
-        ep->>s3: Write UPF to S3
-    end
-    ep->>-co: OK / Failure / Auth Failure
-    co-->ep: Cancel connection
-```
-
-Read data: (optional with cache-relevant request parameters, e.g. If-Modified-Since)
-```http
-GET /tenants/<tenant-id>/timelines/<tl-id>/endpoints/<endpoint-id>/pgdata/<the-pgdata-path>
-Host: epufs.svc.neon.local
-
-<<<
-
-200 OK
-
-<file data>
-```
-
-```mermaid
-sequenceDiagram
-    autonumber
-    participant co as Compute
-    participant ep as EPUFS
-    participant s3 as Blob Storage
-
-    co->>+ep: Read Unlogged Persistent File
-    opt is authenticated
-        ep->>+s3: Request UPF from storage
-        s3->>-ep: Receive UPF from storage
-    end
-    ep->>-co: OK(response) / Failure(storage, auth, ...)
-```
-
-Compute Startup:
-```mermaid
-sequenceDiagram
-    autonumber
-    participant co as Compute
-    participant ps as Pageserver
-    participant ep as EPUFS
-    participant es as Extension server
-
-    note over co: Bind endpoint ep-xxx
-    par Get basebackup
-        co->>+ps: Request basebackup @ LSN
-        ps-)ps: Construct basebackup
-        ps->>-co: Receive basebackup TAR @ LSN
-    and Get startup-critical Unlogged Persistent Files
-        co->>+ep: Get all UPFs of endpoint ep-xxx
-        ep-)ep: Retrieve and gather all UPFs
-        ep->>-co: TAR of UPFs
-    and Get startup-critical extensions
-        loop For every startup-critical extension
-            co->>es: Get critical extension
-            es->>co: Receive critical extension
-        end
-    end
-    note over co: Start compute
-```
-
-CPlane ops:
-```http
-DELETE /tenants/<tenant-id>/timelines/<timeline-id>/endpoints/<endpoint-id>
-Host: epufs.svc.neon.local
-
-<<<
-
-200 OK
-{
-  "tenant": "<tenant-id>",
-  "timeline": "<timeline-id>",
-  "endpoint": "<endpoint-id>",
-  "deleted": {
-    "files": <count>,
-    "bytes": <count>,
-  },
-}
-```
-
-```http
-DELETE /tenants/<tenant-id>/timelines/<timeline-id>
-Host: epufs.svc.neon.local
-
-<<<
-
-200 OK
-{
-  "tenant": "<tenant-id>",
-  "timeline": "<timeline-id>",
-  "deleted": {
-    "files": <count>,
-    "bytes": <count>,
-  },
-}
-```
-
-```http
-DELETE /tenants/<tenant-id>
-Host: epufs.svc.neon.local
-
-<<<
-
-200 OK
-{
-  "tenant": "<tenant-id>",
-  "deleted": {
-    "files": <count>,
-    "bytes": <count>,
-  },
-}
-```
-
-```mermaid
-sequenceDiagram
-    autonumber
-    participant cp as Control Plane
-    participant ep as EPUFS
-    participant s3 as Blob Storage
-
-    alt Tenant deleted
-        cp-)ep: Tenant deleted
-        loop For every object associated with removed tenant
-            ep->>s3: Remove data of deleted tenant from Storage
-        end
-        opt
-            ep-)cp: Tenant cleanup complete
-        end
-    alt Timeline deleted
-        cp-)ep: Timeline deleted
-        loop For every object associated with removed timeline
-            ep->>s3: Remove data of deleted timeline from Storage
-        end
-        opt
-            ep-)cp: Timeline cleanup complete
-        end
-    else Endpoint reassigned or removed
-        cp->>+ep: Endpoint reassigned
-        loop For every object associated with reassigned/removed endpoint
-            ep->>s3: Remove data from Storage
-        end
-        ep->>-cp: Cleanup complete
-    end
-```
-
-### Scalability (if relevant)
-
-Provisionally:  As this service is going to be part of compute startup, this
-service should be able to quickly respond to all requests.  Therefore this
-service is deployed to every AZ we host Computes in, and Computes communicate
-(generally) only to the EPUFS endpoint of the AZ they're hosted in.
-
-Local caching of frequently restarted endpoints' data or metadata may be
-needed for best performance.  However, due to the regional nature of stored
-data but zonal nature of the service deployment, we should be careful when we
-implement any local caching, as it is possible that computes in AZ 1 will
-update data originally written and thus cached by AZ 2.  Cache version tests
-and invalidation is therefore required if we want to roll out caching to this
-service, which is too broad a scope for an MVC.  This is why caching is left
-out of scope for this RFC, and should be considered separately after this RFC
-is implemented.
-
-### Security implications (if relevant)
-This service must be able to authenticate users at least by Tenant ID,
-Timeline ID and Endpoint ID. This will use the existing JWT infrastructure of
-Compute, which will be upgraded to the extent needed to support Timeline- and
-Endpoint-based claims.
-
-The service requires unlimited access to (a prefix of) a blob storage bucket,
-and thus must be hosted outside the Compute VM sandbox.
-
-A service that generates pre-signed request URLs for Compute to download the
-data from that URL is likely problematic, too:  Compute would be able to write
-unlimited data to the bucket, or exfiltrate this signed URL to get read/write
-access to specific objects in this bucket, which would still effectively give
-users access to the S3 bucket (but with improved access logging).
-
-There may be a use case for transferring data associated with one endpoint to
-another endpoint (e.g. to make one endpoint warm its caches with the state of
-another endpoint), but that's not currently in scope, and specific needs may
-be solved through out-of-line communication of data or pre-signed URLs.
-
-### Unresolved questions (if relevant)
-Caching of files is not in the implementation scope of the document, but
-should at some future point be considered to maximize performance.
-
-## Alternative implementation (if relevant)
-Several ideas have come up to solve this issue:
-
-### Use AUXfile
-One prevalent idea was to WAL-log the files using our AUXfile mechanism.
-
-Benefits:
-
-+ We already have this storage mechanism
-
-Demerits:
-
- It isn't available on read replicas
- Additional WAL will be consumed during shutdown and after the shutdown
-  checkpoint, which needs PG modifications to work without panics.
- It increases the data we need to manage in our versioned storage, thus
-  causing higher storage costs with higher retention due to duplication at
-  the storage layer.
-
-### Sign URLs for read/write operations, instead of proxying them
-
-Benefits:
-
-+ The service can be implemented with a much reduced IO budget
-
-Demerits:
-
- Users could get access to these signed credentials
- Not all blob storage services may implement URL signing
-
-### Give endpoints each their own directly accessed block volume
-
-Benefits:
-
-+ Easier to integrate for PostgreSQL
-
-Demerits:
-
- Little control on data size and contents
- Potentially problematic as we'd need to store data all across the pgdata
-  directory.
- EBS is not a good candidate
-   - Attaches in 10s of seconds, if not more; i.e. too cold to start
-   - Shared EBS volumes are a no-go, as you'd have to schedule the endpoint
-     with users of the same EBS volumes, which can't work with VM migration
-   - EBS storage costs are very high (>80$/kilotenant when using a
-     volume/tenant)
-   - EBS volumes can't be mounted across AZ boundaries
- Bucket per endpoint is unfeasible
-   - S3 buckets are priced at $20/month per 1k, which we could better spend
-     on developers.
-   - Allocating service accounts takes time (100s of ms), and service accounts
-     are a limited resource, too; so they're not a good candidate to allocate
-     on a per-endpoint basis.
-   - Giving credentials limited to prefix has similar issues as the pre-signed
-     URL approach.
-   - Bucket DNS lookup will fill DNS caches and put pressure on DNS lookup
-     much more than our current systems would.
- Volumes bound by hypervisor are unlikely
-   - This requires significant investment and increased software on the
-     hypervisor.
-   - It is unclear if we can attach volumes after boot, i.e. for pooled
-     instances.
-
-### Put the files into a table
-
-Benefits:
-
- + Mostly already available in PostgreSQL
-
-Demerits:
-
- - Uses WAL
-   - Can't be used after shutdown checkpoint
-   - Needs a RW endpoint, and table & catalog access to write to this data
- - Gets hit with DB size limitations
- - Depending on user acces:
-   - Inaccessible:  
-     The user doesn't have control over database size caused by
-     these systems.
-   - Accessible:  
-     The user can corrupt these files and cause the system to crash while
-     user-corrupted files are present, thus increasing on-call overhead.
-
-## Definition of Done (if relevant)
-
-This project is done if we have:
-
- One S3 bucket equivalent per region, which stores this per-endpoint data.
- A new service endpoint in at least every AZ, which indirectly grants
-  endpoints access to the data stored for these endpoints in these buckets.
- Compute writes & reads temp-data at shutdown and startup, respectively, for
-  at least the pg_prewarm or lfc_prewarm state files.
- Cleanup of endpoint data is triggered when the endpoint is deleted or is
-  detached from its current timeline.
--- a/libs/compute_api/Cargo.toml
+++ b/libs/compute_api/Cargo.toml
@@ -12,7 +12,6 @@ jsonwebtoken.workspace = true
 serde.workspace = true
 serde_json.workspace = true
 regex.workspace = true
-url.workspace = true

 utils = { path = "../utils" }
 remote_storage = { version = "0.1", path = "../remote_storage/" }
--- a/libs/compute_api/src/spec.rs
+++ b/libs/compute_api/src/spec.rs
@@ -4,14 +4,11 @@
 //! provide it by calling the compute_ctl's `/compute_ctl` endpoint, or
 //! compute_ctl can fetch it by calling the control plane's API.
 use std::collections::HashMap;
-use std::fmt::Display;

-use anyhow::anyhow;
 use indexmap::IndexMap;
 use regex::Regex;
 use remote_storage::RemotePath;
 use serde::{Deserialize, Serialize};
-use url::Url;
 use utils::id::{TenantId, TimelineId};
 use utils::lsn::Lsn;

@@ -184,11 +181,6 @@ pub struct ComputeSpec {
    /// Download LFC state from endpoint_storage and pass it to Postgres on startup
    #[serde(default)]
    pub autoprewarm: bool,
-
-    /// Suspend timeout in seconds.
-    ///
-    /// We use this value to derive other values, such as the installed extensions metric.
-    pub suspend_timeout_seconds: i64,
 }

 /// Feature flag to signal `compute_ctl` to enable certain experimental functionality.
@@ -437,47 +429,6 @@ pub struct JwksSettings {
    pub jwt_audience: Option<String>,
 }

-/// Protocol used to connect to a Pageserver. Parsed from the connstring scheme.
-#[derive(Clone, Copy, Debug, Default)]
-pub enum PageserverProtocol {
-    /// The original protocol based on libpq and COPY. Uses postgresql:// or postgres:// scheme.
-    #[default]
-    Libpq,
-    /// A newer, gRPC-based protocol. Uses grpc:// scheme.
-    Grpc,
-}
-
-impl PageserverProtocol {
-    /// Parses the protocol from a connstring scheme. Defaults to Libpq if no scheme is given.
-    /// Errors if the connstring is an invalid URL.
-    pub fn from_connstring(connstring: &str) -> anyhow::Result<Self> {
-        let scheme = match Url::parse(connstring) {
-            Ok(url) => url.scheme().to_lowercase(),
-            Err(url::ParseError::RelativeUrlWithoutBase) => return Ok(Self::default()),
-            Err(err) => return Err(anyhow!("invalid connstring URL: {err}")),
-        };
-        match scheme.as_str() {
-            "postgresql" | "postgres" => Ok(Self::Libpq),
-            "grpc" => Ok(Self::Grpc),
-            scheme => Err(anyhow!("invalid protocol scheme: {scheme}")),
-        }
-    }
-
-    /// Returns the URL scheme for the protocol, for use in connstrings.
-    pub fn scheme(&self) -> &'static str {
-        match self {
-            Self::Libpq => "postgresql",
-            Self::Grpc => "grpc",
-        }
-    }
-}
-
-impl Display for PageserverProtocol {
-    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
-        f.write_str(self.scheme())
-    }
-}
-
 #[cfg(test)]
 mod tests {
    use std::fs::File;
--- a/libs/compute_api/tests/cluster_spec.json
+++ b/libs/compute_api/tests/cluster_spec.json
@@ -3,7 +3,6 @@

    "timestamp": "2021-05-23T18:25:43.511Z",
    "operation_uuid": "0f657b36-4b0f-4a2d-9c2e-1dcd615e7d8b",
-    "suspend_timeout_seconds": 3600,

    "cluster": {
        "cluster_id": "test-cluster-42",
--- a/libs/pageserver_api/Cargo.toml
+++ b/libs/pageserver_api/Cargo.toml
@@ -18,8 +18,6 @@ bytes.workspace = true
 byteorder.workspace = true
 utils.workspace = true
 postgres_ffi_types.workspace = true
-postgres_versioninfo.workspace = true
-posthog_client_lite.workspace = true
 enum-map.workspace = true
 strum.workspace = true
 strum_macros.workspace = true
@@ -30,13 +28,12 @@ humantime-serde.workspace = true
 chrono = { workspace = true, features = ["serde"] }
 itertools.workspace = true
 storage_broker.workspace = true
-camino = { workspace = true, features = ["serde1"] }
+camino = {workspace = true, features = ["serde1"]}
 remote_storage.workspace = true
 postgres_backend.workspace = true
-nix = { workspace = true, optional = true }
+nix = {workspace = true, optional = true}
 reqwest.workspace = true
 rand.workspace = true
-tracing.workspace = true
 tracing-utils.workspace = true
 once_cell.workspace = true

--- a/libs/pageserver_api/src/config.rs
+++ b/libs/pageserver_api/src/config.rs
@@ -4,7 +4,6 @@ use camino::Utf8PathBuf;
 mod tests;

 use const_format::formatcp;
-use posthog_client_lite::PostHogClientConfig;
 pub const DEFAULT_PG_LISTEN_PORT: u16 = 64000;
 pub const DEFAULT_PG_LISTEN_ADDR: &str = formatcp!("127.0.0.1:{DEFAULT_PG_LISTEN_PORT}");
 pub const DEFAULT_HTTP_LISTEN_PORT: u16 = 9898;
@@ -64,66 +63,25 @@ impl Display for NodeMetadata {
    }
 }

-/// PostHog integration config. This is used in pageserver, storcon, and neon_local.
-/// Ensure backward compatibility when adding new fields.
+/// PostHog integration config.
 #[derive(Debug, Clone, PartialEq, Eq, serde::Serialize, serde::Deserialize)]
 pub struct PostHogConfig {
    /// PostHog project ID
-    #[serde(default)]
-    #[serde(skip_serializing_if = "Option::is_none")]
-    pub project_id: Option<String>,
+    pub project_id: String,
    /// Server-side (private) API key
-    #[serde(default)]
-    #[serde(skip_serializing_if = "Option::is_none")]
-    pub server_api_key: Option<String>,
+    pub server_api_key: String,
    /// Client-side (public) API key
-    #[serde(default)]
-    #[serde(skip_serializing_if = "Option::is_none")]
-    pub client_api_key: Option<String>,
+    pub client_api_key: String,
    /// Private API URL
-    #[serde(default)]
-    #[serde(skip_serializing_if = "Option::is_none")]
-    pub private_api_url: Option<String>,
+    pub private_api_url: String,
    /// Public API URL
-    #[serde(default)]
-    #[serde(skip_serializing_if = "Option::is_none")]
-    pub public_api_url: Option<String>,
-    /// Refresh interval for the feature flag spec.
-    /// The storcon will push the feature flag spec to the pageserver. If the pageserver does not receive
-    /// the spec for `refresh_interval`, it will fetch the spec from the PostHog API.
-    #[serde(default)]
+    pub public_api_url: String,
+    /// Refresh interval for the feature flag spec
    #[serde(skip_serializing_if = "Option::is_none")]
    #[serde(with = "humantime_serde")]
    pub refresh_interval: Option<Duration>,
 }

-impl PostHogConfig {
-    pub fn try_into_posthog_config(self) -> Result<PostHogClientConfig, &'static str> {
-        let Some(project_id) = self.project_id else {
-            return Err("project_id is required");
-        };
-        let Some(server_api_key) = self.server_api_key else {
-            return Err("server_api_key is required");
-        };
-        let Some(client_api_key) = self.client_api_key else {
-            return Err("client_api_key is required");
-        };
-        let Some(private_api_url) = self.private_api_url else {
-            return Err("private_api_url is required");
-        };
-        let Some(public_api_url) = self.public_api_url else {
-            return Err("public_api_url is required");
-        };
-        Ok(PostHogClientConfig {
-            project_id,
-            server_api_key,
-            client_api_key,
-            private_api_url,
-            public_api_url,
-        })
-    }
-}
-
 /// `pageserver.toml`
 ///
 /// We use serde derive with `#[serde(default)]` to generate a deserializer
@@ -409,9 +367,6 @@ pub struct BasebackupCacheConfig {
    // TODO(diko): support max_entry_size_bytes.
    // pub max_entry_size_bytes: u64,
    pub max_size_entries: usize,
-    /// Size of the channel used to send prepare requests to the basebackup cache worker.
-    /// If exceeded, new prepare requests will be dropped.
-    pub prepare_channel_size: usize,
 }

 impl Default for BasebackupCacheConfig {
@@ -420,8 +375,7 @@ impl Default for BasebackupCacheConfig {
            cleanup_period: Duration::from_secs(60),
            max_total_size_bytes: 1024 * 1024 * 1024, // 1 GiB
            // max_entry_size_bytes: 16 * 1024 * 1024,   // 16 MiB
-            max_size_entries: 10000,
-            prepare_channel_size: 100,
+            max_size_entries: 1000,
        }
    }
 }
--- a/libs/pageserver_api/src/controller_api.rs
+++ b/libs/pageserver_api/src/controller_api.rs
@@ -546,11 +546,6 @@ pub struct TimelineImportRequest {
    pub sk_set: Vec<NodeId>,
 }

-#[derive(serde::Serialize, serde::Deserialize, Clone)]
-pub struct TimelineSafekeeperMigrateRequest {
-    pub new_sk_set: Vec<NodeId>,
-}
-
 #[cfg(test)]
 mod test {
    use serde_json;
--- a/libs/pageserver_api/src/models.rs
+++ b/libs/pageserver_api/src/models.rs
@@ -11,7 +11,6 @@ use std::time::{Duration, SystemTime};

 #[cfg(feature = "testing")]
 use camino::Utf8PathBuf;
-use postgres_versioninfo::PgMajorVersion;
 use serde::{Deserialize, Deserializer, Serialize, Serializer};
 use serde_with::serde_as;
 pub use utilization::PageserverUtilization;
@@ -21,9 +20,7 @@ use utils::{completion, serde_system_time};

 use crate::config::Ratio;
 use crate::key::{CompactKey, Key};
-use crate::shard::{
-    DEFAULT_STRIPE_SIZE, ShardCount, ShardIdentity, ShardStripeSize, TenantShardId,
-};
+use crate::shard::{DEFAULT_STRIPE_SIZE, ShardCount, ShardStripeSize, TenantShardId};

 /// The state of a tenant in this pageserver.
 ///
@@ -401,7 +398,7 @@ pub enum TimelineCreateRequestMode {
        // inherits the ancestor's pg_version. Earlier code wasn't
        // using a flattened enum, so, it was an accepted field, and
        // we continue to accept it by having it here.
-        pg_version: Option<PgMajorVersion>,
+        pg_version: Option<u32>,
        #[serde(default, skip_serializing_if = "std::ops::Not::not")]
        read_only: bool,
    },
@@ -413,7 +410,7 @@ pub enum TimelineCreateRequestMode {
    Bootstrap {
        #[serde(default)]
        existing_initdb_timeline_id: Option<TimelineId>,
-        pg_version: Option<PgMajorVersion>,
+        pg_version: Option<u32>,
    },
 }

@@ -477,7 +474,7 @@ pub struct TenantShardSplitResponse {
 }

 /// Parameters that apply to all shards in a tenant.  Used during tenant creation.
-#[derive(Clone, Copy, Serialize, Deserialize, Debug)]
+#[derive(Serialize, Deserialize, Debug)]
 #[serde(deny_unknown_fields)]
 pub struct ShardParameters {
    pub count: ShardCount,
@@ -499,15 +496,6 @@ impl Default for ShardParameters {
    }
 }

-impl From<ShardIdentity> for ShardParameters {
-    fn from(identity: ShardIdentity) -> Self {
-        Self {
-            count: identity.count,
-            stripe_size: identity.stripe_size,
-        }
-    }
-}
-
 #[derive(Debug, Default, Clone, Eq, PartialEq)]
 pub enum FieldPatch<T> {
    Upsert(T),
@@ -1585,7 +1573,7 @@ pub struct TimelineInfo {
    pub last_received_msg_lsn: Option<Lsn>,
    /// the timestamp (in microseconds) of the last received message
    pub last_received_msg_ts: Option<u128>,
-    pub pg_version: PgMajorVersion,
+    pub pg_version: u32,

    pub state: TimelineState,

--- a/libs/pageserver_api/src/shard.rs
+++ b/libs/pageserver_api/src/shard.rs
@@ -37,7 +37,6 @@ use std::hash::{Hash, Hasher};
 pub use ::utils::shard::*;
 use postgres_ffi_types::forknum::INIT_FORKNUM;
 use serde::{Deserialize, Serialize};
-use utils::critical;

 use crate::key::Key;
 use crate::models::ShardParameters;
@@ -180,7 +179,7 @@ impl ShardIdentity {

    /// For use when creating ShardIdentity instances for new shards, where a creation request
    /// specifies the ShardParameters that apply to all shards.
-    pub fn from_params(number: ShardNumber, params: ShardParameters) -> Self {
+    pub fn from_params(number: ShardNumber, params: &ShardParameters) -> Self {
        Self {
            number,
            count: params.count,
@@ -189,17 +188,6 @@ impl ShardIdentity {
        }
    }

-    /// Asserts that the given shard identities are equal. Changes to shard parameters will likely
-    /// result in data corruption.
-    pub fn assert_equal(&self, other: ShardIdentity) {
-        if self != &other {
-            // TODO: for now, we're conservative and just log errors in production. Turn this into a
-            // real assertion when we're confident it doesn't misfire, and also reject requests that
-            // attempt to change it with an error response.
-            critical!("shard identity mismatch: {self:?} != {other:?}");
-        }
-    }
-
    fn is_broken(&self) -> bool {
        self.layout == LAYOUT_BROKEN
    }
--- a/libs/postgres_ffi/Cargo.toml
+++ b/libs/postgres_ffi/Cargo.toml
@@ -19,7 +19,6 @@ serde.workspace = true
 postgres_ffi_types.workspace = true
 utils.workspace = true
 tracing.workspace = true
-postgres_versioninfo.workspace = true

 [dev-dependencies]
 env_logger.workspace = true
--- a/libs/postgres_ffi/benches/waldecoder.rs
+++ b/libs/postgres_ffi/benches/waldecoder.rs
@@ -4,7 +4,6 @@ use criterion::{Bencher, Criterion, criterion_group, criterion_main};
 use postgres_ffi::v17::wal_generator::LogicalMessageGenerator;
 use postgres_ffi::v17::waldecoder_handler::WalStreamDecoderHandler;
 use postgres_ffi::waldecoder::WalStreamDecoder;
-use postgres_versioninfo::PgMajorVersion;
 use pprof::criterion::{Output, PProfProfiler};
 use utils::lsn::Lsn;

@@ -33,7 +32,7 @@ fn bench_complete_record(c: &mut Criterion) {
        let value_size = LogicalMessageGenerator::make_value_size(size, PREFIX);
        let value = vec![1; value_size];

-        let mut decoder = WalStreamDecoder::new(Lsn(0), PgMajorVersion::PG17);
+        let mut decoder = WalStreamDecoder::new(Lsn(0), 170000);
        let msg = LogicalMessageGenerator::new(PREFIX, &value)
            .next()
            .unwrap()
--- a/libs/postgres_ffi/src/lib.rs
+++ b/libs/postgres_ffi/src/lib.rs
@@ -14,8 +14,6 @@ use bytes::Bytes;
 use utils::bin_ser::SerializeError;
 use utils::lsn::Lsn;

-pub use postgres_versioninfo::PgMajorVersion;
-
 macro_rules! postgres_ffi {
    ($version:ident) => {
        #[path = "."]
@@ -93,22 +91,21 @@ macro_rules! dispatch_pgversion {
            $version => $code,
            default = $invalid_pgver_handling,
            pgversions = [
-                $crate::PgMajorVersion::PG14 => v14,
-                $crate::PgMajorVersion::PG15 => v15,
-                $crate::PgMajorVersion::PG16 => v16,
-                $crate::PgMajorVersion::PG17 => v17,
+                14 : v14,
+                15 : v15,
+                16 : v16,
+                17 : v17,
            ]
        )
    };
    ($pgversion:expr => $code:expr,
     default = $default:expr,
-     pgversions = [$($sv:pat => $vsv:ident),+ $(,)?]) => {
-        match ($pgversion.clone().into()) {
+     pgversions = [$($sv:literal : $vsv:ident),+ $(,)?]) => {
+        match ($pgversion) {
            $($sv => {
                use $crate::$vsv as pgv;
                $code
            },)+
-            #[allow(unreachable_patterns)]
            _ => {
                $default
            }
@@ -182,9 +179,9 @@ macro_rules! enum_pgversion {
            $($variant ( $crate::$md::$t )),+
        }
        impl self::$name {
-            pub fn pg_version(&self) -> PgMajorVersion {
+            pub fn pg_version(&self) -> u32 {
                enum_pgversion_dispatch!(self, $name, _ign, {
-                    pgv::bindings::MY_PGVERSION
+                    pgv::bindings::PG_MAJORVERSION_NUM
                })
            }
        }
@@ -198,15 +195,15 @@ macro_rules! enum_pgversion {
    };
    {name = $name:ident,
     path = $p:ident,
-     $(typ = $t:ident,)?
+     typ = $t:ident,
     pgversions = [$($variant:ident : $md:ident),+ $(,)?]} => {
        pub enum $name {
-            $($variant $(($crate::$md::$p::$t))?),+
+            $($variant ($crate::$md::$p::$t)),+
        }
        impl $name {
-            pub fn pg_version(&self) -> PgMajorVersion {
+            pub fn pg_version(&self) -> u32 {
                enum_pgversion_dispatch!(self, $name, _ign, {
-                    pgv::bindings::MY_PGVERSION
+                    pgv::bindings::PG_MAJORVERSION_NUM
                })
            }
        }
@@ -252,21 +249,22 @@ pub use v14::xlog_utils::{
    try_from_pg_timestamp,
 };

-pub fn bkpimage_is_compressed(bimg_info: u8, version: PgMajorVersion) -> bool {
+pub fn bkpimage_is_compressed(bimg_info: u8, version: u32) -> bool {
    dispatch_pgversion!(version, pgv::bindings::bkpimg_is_compressed(bimg_info))
 }

 pub fn generate_wal_segment(
    segno: u64,
    system_id: u64,
-    pg_version: PgMajorVersion,
+    pg_version: u32,
    lsn: Lsn,
 ) -> Result<Bytes, SerializeError> {
    assert_eq!(segno, lsn.segment_number(WAL_SEGMENT_SIZE));

    dispatch_pgversion!(
        pg_version,
-        pgv::xlog_utils::generate_wal_segment(segno, system_id, lsn)
+        pgv::xlog_utils::generate_wal_segment(segno, system_id, lsn),
+        Err(SerializeError::BadInput)
    )
 }

@@ -274,7 +272,7 @@ pub fn generate_pg_control(
    pg_control_bytes: &[u8],
    checkpoint_bytes: &[u8],
    lsn: Lsn,
-    pg_version: PgMajorVersion,
+    pg_version: u32,
 ) -> anyhow::Result<(Bytes, u64, bool)> {
    dispatch_pgversion!(
        pg_version,
@@ -354,7 +352,6 @@ pub fn fsm_logical_to_physical(addr: BlockNumber) -> BlockNumber {
 pub mod waldecoder {
    use std::num::NonZeroU32;

-    use crate::PgMajorVersion;
    use bytes::{Buf, Bytes, BytesMut};
    use thiserror::Error;
    use utils::lsn::Lsn;
@@ -372,7 +369,7 @@ pub mod waldecoder {

    pub struct WalStreamDecoder {
        pub lsn: Lsn,
-        pub pg_version: PgMajorVersion,
+        pub pg_version: u32,
        pub inputbuf: BytesMut,
        pub state: State,
    }
@@ -385,7 +382,7 @@ pub mod waldecoder {
    }

    impl WalStreamDecoder {
-        pub fn new(lsn: Lsn, pg_version: PgMajorVersion) -> WalStreamDecoder {
+        pub fn new(lsn: Lsn, pg_version: u32) -> WalStreamDecoder {
            WalStreamDecoder {
                lsn,
                pg_version,
--- a/libs/postgres_ffi/src/pg_constants_v14.rs
+++ b/libs/postgres_ffi/src/pg_constants_v14.rs
@@ -1,7 +1,3 @@
-use crate::PgMajorVersion;
-
-pub const MY_PGVERSION: PgMajorVersion = PgMajorVersion::PG14;
-
 pub const XLOG_DBASE_CREATE: u8 = 0x00;
 pub const XLOG_DBASE_DROP: u8 = 0x10;

--- a/libs/postgres_ffi/src/pg_constants_v15.rs
+++ b/libs/postgres_ffi/src/pg_constants_v15.rs
@@ -1,7 +1,3 @@
-use crate::PgMajorVersion;
-
-pub const MY_PGVERSION: PgMajorVersion = PgMajorVersion::PG15;
-
 pub const XACT_XINFO_HAS_DROPPED_STATS: u32 = 1u32 << 8;

 pub const XLOG_DBASE_CREATE_FILE_COPY: u8 = 0x00;
--- a/libs/postgres_ffi/src/pg_constants_v16.rs
+++ b/libs/postgres_ffi/src/pg_constants_v16.rs
@@ -1,7 +1,3 @@
-use crate::PgMajorVersion;
-
-pub const MY_PGVERSION: PgMajorVersion = PgMajorVersion::PG16;
-
 pub const XACT_XINFO_HAS_DROPPED_STATS: u32 = 1u32 << 8;

 pub const XLOG_DBASE_CREATE_FILE_COPY: u8 = 0x00;
--- a/libs/postgres_ffi/src/pg_constants_v17.rs
+++ b/libs/postgres_ffi/src/pg_constants_v17.rs
@@ -1,7 +1,3 @@
-use crate::PgMajorVersion;
-
-pub const MY_PGVERSION: PgMajorVersion = PgMajorVersion::PG17;
-
 pub const XACT_XINFO_HAS_DROPPED_STATS: u32 = 1u32 << 8;

 pub const XLOG_DBASE_CREATE_FILE_COPY: u8 = 0x00;
--- a/libs/postgres_ffi/src/walrecord.rs
+++ b/libs/postgres_ffi/src/walrecord.rs
@@ -9,8 +9,8 @@ use utils::bin_ser::DeserializeError;
 use utils::lsn::Lsn;

 use crate::{
-    BLCKSZ, BlockNumber, MultiXactId, MultiXactOffset, MultiXactStatus, Oid, PgMajorVersion,
-    RepOriginId, TimestampTz, TransactionId, XLOG_SIZE_OF_XLOG_RECORD, XLogRecord, pg_constants,
+    BLCKSZ, BlockNumber, MultiXactId, MultiXactOffset, MultiXactStatus, Oid, RepOriginId,
+    TimestampTz, TransactionId, XLOG_SIZE_OF_XLOG_RECORD, XLogRecord, pg_constants,
 };

 #[repr(C)]
@@ -199,17 +199,20 @@ impl DecodedWALRecord {
    /// Check if this WAL record represents a legacy "copy" database creation, which populates new relations
    /// by reading other existing relations' data blocks.  This is more complex to apply than new-style database
    /// creations which simply include all the desired blocks in the WAL, so we need a helper function to detect this case.
-    pub fn is_dbase_create_copy(&self, pg_version: PgMajorVersion) -> bool {
+    pub fn is_dbase_create_copy(&self, pg_version: u32) -> bool {
        if self.xl_rmid == pg_constants::RM_DBASE_ID {
            let info = self.xl_info & pg_constants::XLR_RMGR_INFO_MASK;
            match pg_version {
-                PgMajorVersion::PG14 => {
+                14 => {
                    // Postgres 14 database creations are always the legacy kind
                    info == crate::v14::bindings::XLOG_DBASE_CREATE
                }
-                PgMajorVersion::PG15 => info == crate::v15::bindings::XLOG_DBASE_CREATE_FILE_COPY,
-                PgMajorVersion::PG16 => info == crate::v16::bindings::XLOG_DBASE_CREATE_FILE_COPY,
-                PgMajorVersion::PG17 => info == crate::v17::bindings::XLOG_DBASE_CREATE_FILE_COPY,
+                15 => info == crate::v15::bindings::XLOG_DBASE_CREATE_FILE_COPY,
+                16 => info == crate::v16::bindings::XLOG_DBASE_CREATE_FILE_COPY,
+                17 => info == crate::v17::bindings::XLOG_DBASE_CREATE_FILE_COPY,
+                _ => {
+                    panic!("Unsupported postgres version {pg_version}")
+                }
            }
        } else {
            false
@@ -245,7 +248,7 @@ impl DecodedWALRecord {
 pub fn decode_wal_record(
    record: Bytes,
    decoded: &mut DecodedWALRecord,
-    pg_version: PgMajorVersion,
+    pg_version: u32,
 ) -> anyhow::Result<()> {
    let mut rnode_spcnode: u32 = 0;
    let mut rnode_dbnode: u32 = 0;
@@ -1103,9 +1106,9 @@ pub struct XlClogTruncate {
 }

 impl XlClogTruncate {
-    pub fn decode(buf: &mut Bytes, pg_version: PgMajorVersion) -> XlClogTruncate {
+    pub fn decode(buf: &mut Bytes, pg_version: u32) -> XlClogTruncate {
        XlClogTruncate {
-            pageno: if pg_version < PgMajorVersion::PG17 {
+            pageno: if pg_version < 17 {
                buf.get_u32_le()
            } else {
                buf.get_u64_le() as u32
--- a/libs/postgres_ffi/src/xlog_utils.rs
+++ b/libs/postgres_ffi/src/xlog_utils.rs
@@ -11,9 +11,9 @@ use super::super::waldecoder::WalStreamDecoder;
 use super::bindings::{
    CheckPoint, ControlFileData, DBState_DB_SHUTDOWNED, FullTransactionId, TimeLineID, TimestampTz,
    XLogLongPageHeaderData, XLogPageHeaderData, XLogRecPtr, XLogRecord, XLogSegNo, XLOG_PAGE_MAGIC,
-    MY_PGVERSION
 };
 use super::wal_generator::LogicalMessageGenerator;
+use super::PG_MAJORVERSION;
 use crate::pg_constants;
 use crate::PG_TLI;
 use crate::{uint32, uint64, Oid};
@@ -233,7 +233,7 @@ pub fn find_end_of_wal(
    let mut result = start_lsn;
    let mut curr_lsn = start_lsn;
    let mut buf = [0u8; XLOG_BLCKSZ];
-    let pg_version = MY_PGVERSION;
+    let pg_version = PG_MAJORVERSION[1..3].parse::<u32>().unwrap();
    debug!("find_end_of_wal PG_VERSION: {}", pg_version);

    let mut decoder = WalStreamDecoder::new(start_lsn, pg_version);
--- a/libs/postgres_ffi/wal_craft/src/bin/wal_craft.rs
+++ b/libs/postgres_ffi/wal_craft/src/bin/wal_craft.rs
@@ -4,7 +4,6 @@ use std::str::FromStr;
 use anyhow::*;
 use clap::{Arg, ArgMatches, Command, value_parser};
 use postgres::Client;
-use postgres_ffi::PgMajorVersion;
 use wal_craft::*;

 fn main() -> Result<()> {
@@ -49,7 +48,7 @@ fn main() -> Result<()> {
        Some(("with-initdb", arg_matches)) => {
            let cfg = Conf {
                pg_version: *arg_matches
-                    .get_one::<PgMajorVersion>("pg-version")
+                    .get_one::<u32>("pg-version")
                    .context("'pg-version' is required")?,
                pg_distrib_dir: arg_matches
                    .get_one::<PathBuf>("pg-distrib-dir")
--- a/libs/postgres_ffi/wal_craft/src/lib.rs
+++ b/libs/postgres_ffi/wal_craft/src/lib.rs
@@ -9,8 +9,8 @@ use log::*;
 use postgres::Client;
 use postgres::types::PgLsn;
 use postgres_ffi::{
-    PgMajorVersion, WAL_SEGMENT_SIZE, XLOG_BLCKSZ, XLOG_SIZE_OF_XLOG_LONG_PHD,
-    XLOG_SIZE_OF_XLOG_RECORD, XLOG_SIZE_OF_XLOG_SHORT_PHD,
+    WAL_SEGMENT_SIZE, XLOG_BLCKSZ, XLOG_SIZE_OF_XLOG_LONG_PHD, XLOG_SIZE_OF_XLOG_RECORD,
+    XLOG_SIZE_OF_XLOG_SHORT_PHD,
 };

 macro_rules! xlog_utils_test {
@@ -29,7 +29,7 @@ macro_rules! xlog_utils_test {
 postgres_ffi::for_all_postgres_versions! { xlog_utils_test }

 pub struct Conf {
-    pub pg_version: PgMajorVersion,
+    pub pg_version: u32,
    pub pg_distrib_dir: PathBuf,
    pub datadir: PathBuf,
 }
@@ -52,7 +52,11 @@ impl Conf {
    pub fn pg_distrib_dir(&self) -> anyhow::Result<PathBuf> {
        let path = self.pg_distrib_dir.clone();

-        Ok(path.join(self.pg_version.v_str()))
+        #[allow(clippy::manual_range_patterns)]
+        match self.pg_version {
+            14 | 15 | 16 | 17 => Ok(path.join(format!("v{}", self.pg_version))),
+            _ => bail!("Unsupported postgres version: {}", self.pg_version),
+        }
    }

    fn pg_bin_dir(&self) -> anyhow::Result<PathBuf> {
--- a/libs/postgres_ffi/wal_craft/src/xlog_utils_test.rs
+++ b/libs/postgres_ffi/wal_craft/src/xlog_utils_test.rs
@@ -24,7 +24,7 @@ fn init_logging() {
 fn test_end_of_wal<C: crate::Crafter>(test_name: &str) {
    use crate::*;

-    let pg_version = MY_PGVERSION;
+    let pg_version = PG_MAJORVERSION[1..3].parse::<u32>().unwrap();

    // Craft some WAL
    let top_path = PathBuf::from(env!("CARGO_MANIFEST_DIR"))
--- a/libs/postgres_initdb/Cargo.toml
+++ b/libs/postgres_initdb/Cargo.toml
@@ -9,5 +9,4 @@ anyhow.workspace = true
 tokio.workspace = true
 camino.workspace = true
 thiserror.workspace = true
-postgres_versioninfo.workspace = true
 workspace_hack = { version = "0.1", path = "../../workspace_hack" }
--- a/libs/postgres_initdb/src/lib.rs
+++ b/libs/postgres_initdb/src/lib.rs
@@ -7,13 +7,12 @@
 use std::fmt;

 use camino::Utf8Path;
-use postgres_versioninfo::PgMajorVersion;

 pub struct RunInitdbArgs<'a> {
    pub superuser: &'a str,
    pub locale: &'a str,
    pub initdb_bin: &'a Utf8Path,
-    pub pg_version: PgMajorVersion,
+    pub pg_version: u32,
    pub library_search_path: &'a Utf8Path,
    pub pgdata: &'a Utf8Path,
 }
@@ -80,16 +79,12 @@ pub async fn do_run_initdb(args: RunInitdbArgs<'_>) -> Result<(), Error> {
        .stderr(std::process::Stdio::piped());

    // Before version 14, only the libc provide was available.
-    if pg_version > PgMajorVersion::PG14 {
+    if pg_version > 14 {
        // Version 17 brought with it a builtin locale provider which only provides
        // C and C.UTF-8. While being safer for collation purposes since it is
        // guaranteed to be consistent throughout a major release, it is also more
        // performant.
-        let locale_provider = if pg_version >= PgMajorVersion::PG17 {
-            "builtin"
-        } else {
-            "libc"
-        };
+        let locale_provider = if pg_version >= 17 { "builtin" } else { "libc" };

        initdb_command.args(["--locale-provider", locale_provider]);
    }
--- a/libs/postgres_versioninfo/Cargo.toml
+++ b/libs/postgres_versioninfo/Cargo.toml
@@ -1,12 +0,0 @@
-[package]
-name = "postgres_versioninfo"
-version = "0.1.0"
-edition = "2024"
-license.workspace = true
-
-[dependencies]
-anyhow.workspace = true
-thiserror.workspace = true
-serde.workspace = true
-serde_repr.workspace = true
-workspace_hack = { version = "0.1", path = "../../workspace_hack" }
--- a/libs/postgres_versioninfo/src/lib.rs
+++ b/libs/postgres_versioninfo/src/lib.rs
@@ -1,175 +0,0 @@
-use serde::{Deserialize, Deserializer, Serialize, Serializer};
-use serde_repr::{Deserialize_repr, Serialize_repr};
-use std::fmt::{Display, Formatter};
-use std::str::FromStr;
-
-/// An enum with one variant for each major version of PostgreSQL that we support.
-///
-#[derive(Debug, Clone, Copy, Ord, PartialOrd, Eq, PartialEq, Deserialize_repr, Serialize_repr)]
-#[repr(u32)]
-pub enum PgMajorVersion {
-    PG14 = 14,
-    PG15 = 15,
-    PG16 = 16,
-    PG17 = 17,
-    // !!! When you add a new PgMajorVersion, don't forget to update PgMajorVersion::ALL
-}
-
-/// A full PostgreSQL version ID, in MMmmbb numerical format (Major/minor/bugfix)
-#[derive(Debug, Copy, Clone, Ord, PartialOrd, Eq, PartialEq)]
-#[repr(transparent)]
-pub struct PgVersionId(u32);
-
-impl PgVersionId {
-    pub const UNKNOWN: PgVersionId = PgVersionId(0);
-
-    pub fn from_full_pg_version(version: u32) -> PgVersionId {
-        match version {
-            0 => PgVersionId(version), // unknown version
-            140000..180000 => PgVersionId(version),
-            _ => panic!("Invalid full PostgreSQL version ID {version}"),
-        }
-    }
-}
-
-impl Display for PgVersionId {
-    fn fmt(&self, f: &mut Formatter<'_>) -> std::fmt::Result {
-        u32::fmt(&self.0, f)
-    }
-}
-
-impl Serialize for PgVersionId {
-    fn serialize<S>(&self, serializer: S) -> Result<S::Ok, S::Error>
-    where
-        S: Serializer,
-    {
-        u32::serialize(&self.0, serializer)
-    }
-}
-
-impl<'de> Deserialize<'de> for PgVersionId {
-    fn deserialize<D>(deserializer: D) -> Result<Self, D::Error>
-    where
-        D: Deserializer<'de>,
-    {
-        u32::deserialize(deserializer).map(PgVersionId)
-    }
-
-    fn deserialize_in_place<D>(deserializer: D, place: &mut Self) -> Result<(), D::Error>
-    where
-        D: Deserializer<'de>,
-    {
-        u32::deserialize_in_place(deserializer, &mut place.0)
-    }
-}
-
-impl PgMajorVersion {
-    /// Get the numerical representation of the represented Major Version
-    pub const fn major_version_num(&self) -> u32 {
-        match self {
-            PgMajorVersion::PG14 => 14,
-            PgMajorVersion::PG15 => 15,
-            PgMajorVersion::PG16 => 16,
-            PgMajorVersion::PG17 => 17,
-        }
-    }
-
-    /// Get the contents of this version's PG_VERSION file.
-    ///
-    /// The PG_VERSION file is used to determine the PostgreSQL version that currently
-    /// owns the data in a PostgreSQL data directory.
-    pub fn versionfile_string(&self) -> &'static str {
-        match self {
-            PgMajorVersion::PG14 => "14",
-            PgMajorVersion::PG15 => "15",
-            PgMajorVersion::PG16 => "16\x0A",
-            PgMajorVersion::PG17 => "17\x0A",
-        }
-    }
-
-    /// Get the v{version} string of this major PostgreSQL version.
-    ///
-    /// Because this was hand-coded in various places, this was moved into a shared
-    /// implementation.
-    pub fn v_str(&self) -> String {
-        match self {
-            PgMajorVersion::PG14 => "v14",
-            PgMajorVersion::PG15 => "v15",
-            PgMajorVersion::PG16 => "v16",
-            PgMajorVersion::PG17 => "v17",
-        }
-        .to_string()
-    }
-
-    /// All currently supported major versions of PostgreSQL.
-    pub const ALL: &'static [PgMajorVersion] = &[
-        PgMajorVersion::PG14,
-        PgMajorVersion::PG15,
-        PgMajorVersion::PG16,
-        PgMajorVersion::PG17,
-    ];
-}
-
-impl Display for PgMajorVersion {
-    fn fmt(&self, f: &mut Formatter<'_>) -> std::fmt::Result {
-        f.write_str(match self {
-            PgMajorVersion::PG14 => "PgMajorVersion::PG14",
-            PgMajorVersion::PG15 => "PgMajorVersion::PG15",
-            PgMajorVersion::PG16 => "PgMajorVersion::PG16",
-            PgMajorVersion::PG17 => "PgMajorVersion::PG17",
-        })
-    }
-}
-
-#[derive(Debug, thiserror::Error)]
-#[allow(dead_code)]
-pub struct InvalidPgVersion(u32);
-
-impl Display for InvalidPgVersion {
-    fn fmt(&self, f: &mut Formatter<'_>) -> std::fmt::Result {
-        write!(f, "InvalidPgVersion({})", self.0)
-    }
-}
-
-impl TryFrom<PgVersionId> for PgMajorVersion {
-    type Error = InvalidPgVersion;
-
-    fn try_from(value: PgVersionId) -> Result<Self, Self::Error> {
-        Ok(match value.0 / 10000 {
-            14 => PgMajorVersion::PG14,
-            15 => PgMajorVersion::PG15,
-            16 => PgMajorVersion::PG16,
-            17 => PgMajorVersion::PG17,
-            _ => return Err(InvalidPgVersion(value.0)),
-        })
-    }
-}
-
-impl From<PgMajorVersion> for PgVersionId {
-    fn from(value: PgMajorVersion) -> Self {
-        PgVersionId((value as u32) * 10000)
-    }
-}
-
-#[derive(Debug, PartialEq, Eq, thiserror::Error)]
-pub struct PgMajorVersionParseError(String);
-
-impl Display for PgMajorVersionParseError {
-    fn fmt(&self, f: &mut Formatter<'_>) -> std::fmt::Result {
-        write!(f, "PgMajorVersionParseError({})", self.0)
-    }
-}
-
-impl FromStr for PgMajorVersion {
-    type Err = PgMajorVersionParseError;
-
-    fn from_str(s: &str) -> Result<Self, Self::Err> {
-        Ok(match s {
-            "14" => PgMajorVersion::PG14,
-            "15" => PgMajorVersion::PG15,
-            "16" => PgMajorVersion::PG16,
-            "17" => PgMajorVersion::PG17,
-            _ => return Err(PgMajorVersionParseError(s.to_string())),
-        })
-    }
-}
--- a/libs/posthog_client_lite/src/background_loop.rs
+++ b/libs/posthog_client_lite/src/background_loop.rs
@@ -1,22 +1,17 @@
 //! A background loop that fetches feature flags from PostHog and updates the feature store.

-use std::{
-    sync::Arc,
-    time::{Duration, SystemTime},
-};
+use std::{sync::Arc, time::Duration};

 use arc_swap::ArcSwap;
 use tokio_util::sync::CancellationToken;
 use tracing::{Instrument, info_span};

-use crate::{
-    CaptureEvent, FeatureStore, LocalEvaluationResponse, PostHogClient, PostHogClientConfig,
-};
+use crate::{CaptureEvent, FeatureStore, PostHogClient, PostHogClientConfig};

 /// A background loop that fetches feature flags from PostHog and updates the feature store.
 pub struct FeatureResolverBackgroundLoop {
    posthog_client: PostHogClient,
-    feature_store: ArcSwap<(SystemTime, Arc<FeatureStore>)>,
+    feature_store: ArcSwap<FeatureStore>,
    cancel: CancellationToken,
 }

@@ -24,35 +19,11 @@ impl FeatureResolverBackgroundLoop {
    pub fn new(config: PostHogClientConfig, shutdown_pageserver: CancellationToken) -> Self {
        Self {
            posthog_client: PostHogClient::new(config),
-            feature_store: ArcSwap::new(Arc::new((
-                SystemTime::UNIX_EPOCH,
-                Arc::new(FeatureStore::new()),
-            ))),
+            feature_store: ArcSwap::new(Arc::new(FeatureStore::new())),
            cancel: shutdown_pageserver,
        }
    }

-    /// Update the feature store with a new feature flag spec bypassing the normal refresh loop.
-    pub fn update(&self, spec: String) -> anyhow::Result<()> {
-        let resp: LocalEvaluationResponse = serde_json::from_str(&spec)?;
-        self.update_feature_store_nofail(resp, "http_propagate");
-        Ok(())
-    }
-
-    fn update_feature_store_nofail(&self, resp: LocalEvaluationResponse, source: &'static str) {
-        let project_id = self.posthog_client.config.project_id.parse::<u64>().ok();
-        match FeatureStore::new_with_flags(resp.flags, project_id) {
-            Ok(feature_store) => {
-                self.feature_store
-                    .store(Arc::new((SystemTime::now(), Arc::new(feature_store))));
-                tracing::info!("Feature flag updated from {}", source);
-            }
-            Err(e) => {
-                tracing::warn!("Cannot process feature flag spec from {}: {}", source, e);
-            }
-        }
-    }
-
    pub fn spawn(
        self: Arc<Self>,
        handle: &tokio::runtime::Handle,
@@ -76,17 +47,6 @@ impl FeatureResolverBackgroundLoop {
                        _ = ticker.tick() => {}
                        _ = cancel.cancelled() => break
                    }
-                    {
-                        let last_update = this.feature_store.load().0;
-                        if let Ok(elapsed) = last_update.elapsed() {
-                            if elapsed < refresh_period {
-                                tracing::debug!(
-                                    "Skipping feature flag refresh because it's too soon"
-                                );
-                                continue;
-                            }
-                        }
-                    }
                    let resp = match this
                        .posthog_client
                        .get_feature_flags_local_evaluation()
@@ -98,7 +58,16 @@ impl FeatureResolverBackgroundLoop {
                            continue;
                        }
                    };
-                    this.update_feature_store_nofail(resp, "refresh_loop");
+                    let project_id = this.posthog_client.config.project_id.parse::<u64>().ok();
+                    match FeatureStore::new_with_flags(resp.flags, project_id) {
+                        Ok(feature_store) => {
+                            this.feature_store.store(Arc::new(feature_store));
+                            tracing::info!("Feature flag updated");
+                        }
+                        Err(e) => {
+                            tracing::warn!("Cannot process feature flag spec: {}", e);
+                        }
+                    }
                }
                tracing::info!("PostHog feature resolver stopped");
            }
@@ -123,6 +92,6 @@ impl FeatureResolverBackgroundLoop {
    }

    pub fn feature_store(&self) -> Arc<FeatureStore> {
-        self.feature_store.load().1.clone()
+        self.feature_store.load_full()
    }
 }
--- a/libs/posthog_client_lite/src/lib.rs
+++ b/libs/posthog_client_lite/src/lib.rs
@@ -544,8 +544,17 @@ impl PostHogClient {
        self.config.server_api_key.starts_with("phs_")
    }

-    /// Get the raw JSON spec, same as `get_feature_flags_local_evaluation` but without parsing.
-    pub async fn get_feature_flags_local_evaluation_raw(&self) -> anyhow::Result<String> {
+    /// Fetch the feature flag specs from the server.
+    ///
+    /// This is unfortunately an undocumented API at:
+    /// - <https://posthog.com/docs/api/feature-flags#get-api-projects-project_id-feature_flags-local_evaluation>
+    /// - <https://posthog.com/docs/feature-flags/local-evaluation>
+    ///
+    /// The handling logic in [`FeatureStore`] mostly follows the Python API implementation.
+    /// See `_compute_flag_locally` in <https://github.com/PostHog/posthog-python/blob/master/posthog/client.py>
+    pub async fn get_feature_flags_local_evaluation(
+        &self,
+    ) -> anyhow::Result<LocalEvaluationResponse> {
        // BASE_URL/api/projects/:project_id/feature_flags/local_evaluation
        // with bearer token of self.server_api_key
        // OR
@@ -579,22 +588,7 @@ impl PostHogClient {
                body
            ));
        }
-        Ok(body)
-    }
-
-    /// Fetch the feature flag specs from the server.
-    ///
-    /// This is unfortunately an undocumented API at:
-    /// - <https://posthog.com/docs/api/feature-flags#get-api-projects-project_id-feature_flags-local_evaluation>
-    /// - <https://posthog.com/docs/feature-flags/local-evaluation>
-    ///
-    /// The handling logic in [`FeatureStore`] mostly follows the Python API implementation.
-    /// See `_compute_flag_locally` in <https://github.com/PostHog/posthog-python/blob/master/posthog/client.py>
-    pub async fn get_feature_flags_local_evaluation(
-        &self,
-    ) -> Result<LocalEvaluationResponse, anyhow::Error> {
-        let raw = self.get_feature_flags_local_evaluation_raw().await?;
-        Ok(serde_json::from_str(&raw)?)
+        Ok(serde_json::from_str(&body)?)
    }

    /// Capture an event. This will only be used to report the feature flag usage back to PostHog, though
--- a/libs/proxy/tokio-postgres2/src/config.rs
+++ b/libs/proxy/tokio-postgres2/src/config.rs
@@ -12,9 +12,7 @@ use tokio::net::TcpStream;

 use crate::connect::connect;
 use crate::connect_raw::{RawConnection, connect_raw};
-use crate::connect_tls::connect_tls;
-use crate::maybe_tls_stream::MaybeTlsStream;
-use crate::tls::{MakeTlsConnect, TlsConnect, TlsStream};
+use crate::tls::{MakeTlsConnect, TlsConnect};
 use crate::{Client, Connection, Error};

 /// TLS configuration.
@@ -240,7 +238,7 @@ impl Config {
        connect(tls, self).await
    }

-    pub async fn tls_and_authenticate<S, T>(
+    pub async fn connect_raw<S, T>(
        &self,
        stream: S,
        tls: T,
@@ -249,19 +247,7 @@ impl Config {
        S: AsyncRead + AsyncWrite + Unpin,
        T: TlsConnect<S>,
    {
-        let stream = connect_tls(stream, self.ssl_mode, tls).await?;
-        connect_raw(stream, self).await
-    }
-
-    pub async fn authenticate<S, T>(
-        &self,
-        stream: MaybeTlsStream<S, T>,
-    ) -> Result<RawConnection<S, T>, Error>
-    where
-        S: AsyncRead + AsyncWrite + Unpin,
-        T: TlsStream + Unpin,
-    {
-        connect_raw(stream, self).await
+        connect_raw(stream, tls, self).await
    }
 }

--- a/libs/proxy/tokio-postgres2/src/connect.rs
+++ b/libs/proxy/tokio-postgres2/src/connect.rs
@@ -9,7 +9,6 @@ use crate::codec::BackendMessage;
 use crate::config::Host;
 use crate::connect_raw::connect_raw;
 use crate::connect_socket::connect_socket;
-use crate::connect_tls::connect_tls;
 use crate::tls::{MakeTlsConnect, TlsConnect};
 use crate::{Client, Config, Connection, Error, RawConnection};

@@ -45,14 +44,13 @@ where
    T: TlsConnect<TcpStream>,
 {
    let socket = connect_socket(host_addr, host, port, config.connect_timeout).await?;
-    let stream = connect_tls(socket, config.ssl_mode, tls).await?;
    let RawConnection {
        stream,
        parameters,
        delayed_notice,
        process_id,
        secret_key,
-    } = connect_raw(stream, config).await?;
+    } = connect_raw(socket, tls, config).await?;

    let socket_config = SocketConfig {
        host_addr,
--- a/libs/proxy/tokio-postgres2/src/connect_raw.rs
+++ b/libs/proxy/tokio-postgres2/src/connect_raw.rs
@@ -16,8 +16,9 @@ use tokio_util::codec::Framed;
 use crate::Error;
 use crate::codec::{BackendMessage, BackendMessages, FrontendMessage, PostgresCodec};
 use crate::config::{self, AuthKeys, Config};
+use crate::connect_tls::connect_tls;
 use crate::maybe_tls_stream::MaybeTlsStream;
-use crate::tls::TlsStream;
+use crate::tls::{TlsConnect, TlsStream};

 pub struct StartupStream<S, T> {
    inner: Framed<MaybeTlsStream<S, T>, PostgresCodec>,
@@ -86,13 +87,16 @@ pub struct RawConnection<S, T> {
 }

 pub async fn connect_raw<S, T>(
-    stream: MaybeTlsStream<S, T>,
+    stream: S,
+    tls: T,
    config: &Config,
-) -> Result<RawConnection<S, T>, Error>
+) -> Result<RawConnection<S, T::Stream>, Error>
 where
    S: AsyncRead + AsyncWrite + Unpin,
-    T: TlsStream + Unpin,
+    T: TlsConnect<S>,
 {
+    let stream = connect_tls(stream, config.ssl_mode, tls).await?;
+
    let mut stream = StartupStream {
        inner: Framed::new(stream, PostgresCodec),
        buf: BackendMessages::empty(),
--- a/libs/safekeeper_api/Cargo.toml
+++ b/libs/safekeeper_api/Cargo.toml
@@ -10,7 +10,6 @@ const_format.workspace = true
 serde.workspace = true
 serde_json.workspace = true
 postgres_ffi.workspace = true
-postgres_versioninfo.workspace = true
 pq_proto.workspace = true
 tokio.workspace = true
 utils.workspace = true
--- a/libs/safekeeper_api/src/lib.rs
+++ b/libs/safekeeper_api/src/lib.rs
@@ -8,8 +8,6 @@ pub mod membership;
 /// Public API types
 pub mod models;

-pub use postgres_versioninfo::{PgMajorVersion, PgVersionId};
-
 /// Consensus logical timestamp. Note: it is a part of sk control file.
 pub type Term = u64;
 /// With this term timeline is created initially. It
@@ -22,7 +20,7 @@ pub const INITIAL_TERM: Term = 0;
 #[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
 pub struct ServerInfo {
    /// Postgres server version
-    pub pg_version: PgVersionId,
+    pub pg_version: u32,
    pub system_id: SystemId,
    pub wal_seg_size: u32,
 }
--- a/libs/safekeeper_api/src/models.rs
+++ b/libs/safekeeper_api/src/models.rs
@@ -4,7 +4,6 @@ use std::net::SocketAddr;

 use pageserver_api::shard::ShardIdentity;
 use postgres_ffi::TimestampTz;
-use postgres_versioninfo::PgVersionId;
 use serde::{Deserialize, Serialize};
 use tokio::time::Instant;
 use utils::id::{NodeId, TenantId, TenantTimelineId, TimelineId};
@@ -24,7 +23,8 @@ pub struct TimelineCreateRequest {
    pub tenant_id: TenantId,
    pub timeline_id: TimelineId,
    pub mconf: Configuration,
-    pub pg_version: PgVersionId,
+    /// In the PG_VERSION_NUM macro format, like 140017.
+    pub pg_version: u32,
    pub system_id: Option<u64>,
    // By default WAL_SEGMENT_SIZE
    pub wal_seg_size: Option<u32>,
@@ -210,7 +210,7 @@ pub struct TimelineStatus {
 }

 /// Request to switch membership configuration.
-#[derive(Clone, Serialize, Deserialize)]
+#[derive(Serialize, Deserialize)]
 #[serde(transparent)]
 pub struct TimelineMembershipSwitchRequest {
    pub mconf: Configuration,
@@ -221,8 +221,6 @@ pub struct TimelineMembershipSwitchRequest {
 pub struct TimelineMembershipSwitchResponse {
    pub previous_conf: Configuration,
    pub current_conf: Configuration,
-    pub term: Term,
-    pub flush_lsn: Lsn,
 }

 #[derive(Clone, Copy, Serialize, Deserialize)]
--- a/libs/utils/src/sync/gate.rs
+++ b/libs/utils/src/sync/gate.rs
@@ -86,14 +86,6 @@ pub enum GateError {
    GateClosed,
 }

-impl GateError {
-    pub fn is_cancel(&self) -> bool {
-        match self {
-            GateError::GateClosed => true,
-        }
-    }
-}
-
 impl Default for Gate {
    fn default() -> Self {
        Self {
--- a/libs/wal_decoder/benches/bench_interpret_wal.rs
+++ b/libs/wal_decoder/benches/bench_interpret_wal.rs
@@ -10,7 +10,7 @@ use futures::StreamExt;
 use futures::stream::FuturesUnordered;
 use pageserver_api::shard::{ShardIdentity, ShardStripeSize};
 use postgres_ffi::waldecoder::WalStreamDecoder;
-use postgres_ffi::{MAX_SEND_SIZE, PgMajorVersion, WAL_SEGMENT_SIZE};
+use postgres_ffi::{MAX_SEND_SIZE, WAL_SEGMENT_SIZE};
 use pprof::criterion::{Output, PProfProfiler};
 use remote_storage::{
    DownloadOpts, GenericRemoteStorage, ListingMode, RemoteStorageConfig, RemoteStorageKind,
@@ -115,7 +115,7 @@ struct BenchmarkData {

 #[derive(Deserialize)]
 struct BenchmarkMetadata {
-    pg_version: PgMajorVersion,
+    pg_version: u32,
    start_lsn: Lsn,
 }

--- a/libs/wal_decoder/src/decoder.rs
+++ b/libs/wal_decoder/src/decoder.rs
@@ -7,8 +7,8 @@ use bytes::{Buf, Bytes};
 use pageserver_api::key::rel_block_to_key;
 use pageserver_api::reltag::{RelTag, SlruKind};
 use pageserver_api::shard::ShardIdentity;
+use postgres_ffi::pg_constants;
 use postgres_ffi::walrecord::*;
-use postgres_ffi::{PgMajorVersion, pg_constants};
 use postgres_ffi_types::forknum::VISIBILITYMAP_FORKNUM;
 use utils::lsn::Lsn;

@@ -24,7 +24,7 @@ impl InterpretedWalRecord {
        buf: Bytes,
        shards: &[ShardIdentity],
        next_record_lsn: Lsn,
-        pg_version: PgMajorVersion,
+        pg_version: u32,
    ) -> anyhow::Result<HashMap<ShardIdentity, InterpretedWalRecord>> {
        let mut decoded = DecodedWALRecord::default();
        decode_wal_record(buf, &mut decoded, pg_version)?;
@@ -78,7 +78,7 @@ impl MetadataRecord {
        decoded: &DecodedWALRecord,
        shard_records: &mut HashMap<ShardIdentity, InterpretedWalRecord>,
        next_record_lsn: Lsn,
-        pg_version: PgMajorVersion,
+        pg_version: u32,
    ) -> anyhow::Result<()> {
        // Note: this doesn't actually copy the bytes since
        // the [`Bytes`] type implements it via a level of indirection.
@@ -193,7 +193,7 @@ impl MetadataRecord {
    fn decode_heapam_record(
        buf: &mut Bytes,
        decoded: &DecodedWALRecord,
-        pg_version: PgMajorVersion,
+        pg_version: u32,
    ) -> anyhow::Result<Option<MetadataRecord>> {
        // Handle VM bit updates that are implicitly part of heap records.

@@ -205,7 +205,7 @@ impl MetadataRecord {
        let mut flags = pg_constants::VISIBILITYMAP_VALID_BITS;

        match pg_version {
-            PgMajorVersion::PG14 => {
+            14 => {
                if decoded.xl_rmid == pg_constants::RM_HEAP_ID {
                    let info = decoded.xl_info & pg_constants::XLOG_HEAP_OPMASK;

@@ -272,7 +272,7 @@ impl MetadataRecord {
                    anyhow::bail!("Unknown RMGR {} for Heap decoding", decoded.xl_rmid);
                }
            }
-            PgMajorVersion::PG15 => {
+            15 => {
                if decoded.xl_rmid == pg_constants::RM_HEAP_ID {
                    let info = decoded.xl_info & pg_constants::XLOG_HEAP_OPMASK;

@@ -339,7 +339,7 @@ impl MetadataRecord {
                    anyhow::bail!("Unknown RMGR {} for Heap decoding", decoded.xl_rmid);
                }
            }
-            PgMajorVersion::PG16 => {
+            16 => {
                if decoded.xl_rmid == pg_constants::RM_HEAP_ID {
                    let info = decoded.xl_info & pg_constants::XLOG_HEAP_OPMASK;

@@ -406,7 +406,7 @@ impl MetadataRecord {
                    anyhow::bail!("Unknown RMGR {} for Heap decoding", decoded.xl_rmid);
                }
            }
-            PgMajorVersion::PG17 => {
+            17 => {
                if decoded.xl_rmid == pg_constants::RM_HEAP_ID {
                    let info = decoded.xl_info & pg_constants::XLOG_HEAP_OPMASK;

@@ -473,6 +473,7 @@ impl MetadataRecord {
                    anyhow::bail!("Unknown RMGR {} for Heap decoding", decoded.xl_rmid);
                }
            }
+            _ => {}
        }

        if new_heap_blkno.is_some() || old_heap_blkno.is_some() {
@@ -499,7 +500,7 @@ impl MetadataRecord {
    fn decode_neonmgr_record(
        buf: &mut Bytes,
        decoded: &DecodedWALRecord,
-        pg_version: PgMajorVersion,
+        pg_version: u32,
    ) -> anyhow::Result<Option<MetadataRecord>> {
        // Handle VM bit updates that are implicitly part of heap records.

@@ -513,7 +514,7 @@ impl MetadataRecord {
        assert_eq!(decoded.xl_rmid, pg_constants::RM_NEON_ID);

        match pg_version {
-            PgMajorVersion::PG16 | PgMajorVersion::PG17 => {
+            16 | 17 => {
                let info = decoded.xl_info & pg_constants::XLOG_HEAP_OPMASK;

                match info {
@@ -573,7 +574,7 @@ impl MetadataRecord {
                    info => anyhow::bail!("Unknown WAL record type for Neon RMGR: {}", info),
                }
            }
-            PgMajorVersion::PG15 | PgMajorVersion::PG14 => anyhow::bail!(
+            _ => anyhow::bail!(
                "Neon RMGR has no known compatibility with PostgreSQL version {}",
                pg_version
            ),
@@ -628,121 +629,116 @@ impl MetadataRecord {
    fn decode_dbase_record(
        buf: &mut Bytes,
        decoded: &DecodedWALRecord,
-        pg_version: PgMajorVersion,
+        pg_version: u32,
    ) -> anyhow::Result<Option<MetadataRecord>> {
        // TODO: Refactor this to avoid the duplication between postgres versions.

        let info = decoded.xl_info & pg_constants::XLR_RMGR_INFO_MASK;
        tracing::debug!(%info, %pg_version, "handle RM_DBASE_ID");

-        match pg_version {
-            PgMajorVersion::PG14 => {
-                if info == postgres_ffi::v14::bindings::XLOG_DBASE_CREATE {
-                    let createdb = XlCreateDatabase::decode(buf);
-                    tracing::debug!("XLOG_DBASE_CREATE v14");
+        if pg_version == 14 {
+            if info == postgres_ffi::v14::bindings::XLOG_DBASE_CREATE {
+                let createdb = XlCreateDatabase::decode(buf);
+                tracing::debug!("XLOG_DBASE_CREATE v14");

-                    let record = MetadataRecord::Dbase(DbaseRecord::Create(DbaseCreate {
-                        db_id: createdb.db_id,
-                        tablespace_id: createdb.tablespace_id,
-                        src_db_id: createdb.src_db_id,
-                        src_tablespace_id: createdb.src_tablespace_id,
-                    }));
+                let record = MetadataRecord::Dbase(DbaseRecord::Create(DbaseCreate {
+                    db_id: createdb.db_id,
+                    tablespace_id: createdb.tablespace_id,
+                    src_db_id: createdb.src_db_id,
+                    src_tablespace_id: createdb.src_tablespace_id,
+                }));

-                    return Ok(Some(record));
-                } else if info == postgres_ffi::v14::bindings::XLOG_DBASE_DROP {
-                    let dropdb = XlDropDatabase::decode(buf);
+                return Ok(Some(record));
+            } else if info == postgres_ffi::v14::bindings::XLOG_DBASE_DROP {
+                let dropdb = XlDropDatabase::decode(buf);

-                    let record = MetadataRecord::Dbase(DbaseRecord::Drop(DbaseDrop {
-                        db_id: dropdb.db_id,
-                        tablespace_ids: dropdb.tablespace_ids,
-                    }));
+                let record = MetadataRecord::Dbase(DbaseRecord::Drop(DbaseDrop {
+                    db_id: dropdb.db_id,
+                    tablespace_ids: dropdb.tablespace_ids,
+                }));

-                    return Ok(Some(record));
-                }
+                return Ok(Some(record));
            }
-            PgMajorVersion::PG15 => {
-                if info == postgres_ffi::v15::bindings::XLOG_DBASE_CREATE_WAL_LOG {
-                    tracing::debug!("XLOG_DBASE_CREATE_WAL_LOG: noop");
-                } else if info == postgres_ffi::v15::bindings::XLOG_DBASE_CREATE_FILE_COPY {
-                    // The XLOG record was renamed between v14 and v15,
-                    // but the record format is the same.
-                    // So we can reuse XlCreateDatabase here.
-                    tracing::debug!("XLOG_DBASE_CREATE_FILE_COPY");
+        } else if pg_version == 15 {
+            if info == postgres_ffi::v15::bindings::XLOG_DBASE_CREATE_WAL_LOG {
+                tracing::debug!("XLOG_DBASE_CREATE_WAL_LOG: noop");
+            } else if info == postgres_ffi::v15::bindings::XLOG_DBASE_CREATE_FILE_COPY {
+                // The XLOG record was renamed between v14 and v15,
+                // but the record format is the same.
+                // So we can reuse XlCreateDatabase here.
+                tracing::debug!("XLOG_DBASE_CREATE_FILE_COPY");

-                    let createdb = XlCreateDatabase::decode(buf);
-                    let record = MetadataRecord::Dbase(DbaseRecord::Create(DbaseCreate {
-                        db_id: createdb.db_id,
-                        tablespace_id: createdb.tablespace_id,
-                        src_db_id: createdb.src_db_id,
-                        src_tablespace_id: createdb.src_tablespace_id,
-                    }));
+                let createdb = XlCreateDatabase::decode(buf);
+                let record = MetadataRecord::Dbase(DbaseRecord::Create(DbaseCreate {
+                    db_id: createdb.db_id,
+                    tablespace_id: createdb.tablespace_id,
+                    src_db_id: createdb.src_db_id,
+                    src_tablespace_id: createdb.src_tablespace_id,
+                }));

-                    return Ok(Some(record));
-                } else if info == postgres_ffi::v15::bindings::XLOG_DBASE_DROP {
-                    let dropdb = XlDropDatabase::decode(buf);
-                    let record = MetadataRecord::Dbase(DbaseRecord::Drop(DbaseDrop {
-                        db_id: dropdb.db_id,
-                        tablespace_ids: dropdb.tablespace_ids,
-                    }));
+                return Ok(Some(record));
+            } else if info == postgres_ffi::v15::bindings::XLOG_DBASE_DROP {
+                let dropdb = XlDropDatabase::decode(buf);
+                let record = MetadataRecord::Dbase(DbaseRecord::Drop(DbaseDrop {
+                    db_id: dropdb.db_id,
+                    tablespace_ids: dropdb.tablespace_ids,
+                }));

-                    return Ok(Some(record));
-                }
+                return Ok(Some(record));
            }
-            PgMajorVersion::PG16 => {
-                if info == postgres_ffi::v16::bindings::XLOG_DBASE_CREATE_WAL_LOG {
-                    tracing::debug!("XLOG_DBASE_CREATE_WAL_LOG: noop");
-                } else if info == postgres_ffi::v16::bindings::XLOG_DBASE_CREATE_FILE_COPY {
-                    // The XLOG record was renamed between v14 and v15,
-                    // but the record format is the same.
-                    // So we can reuse XlCreateDatabase here.
-                    tracing::debug!("XLOG_DBASE_CREATE_FILE_COPY");
+        } else if pg_version == 16 {
+            if info == postgres_ffi::v16::bindings::XLOG_DBASE_CREATE_WAL_LOG {
+                tracing::debug!("XLOG_DBASE_CREATE_WAL_LOG: noop");
+            } else if info == postgres_ffi::v16::bindings::XLOG_DBASE_CREATE_FILE_COPY {
+                // The XLOG record was renamed between v14 and v15,
+                // but the record format is the same.
+                // So we can reuse XlCreateDatabase here.
+                tracing::debug!("XLOG_DBASE_CREATE_FILE_COPY");

-                    let createdb = XlCreateDatabase::decode(buf);
-                    let record = MetadataRecord::Dbase(DbaseRecord::Create(DbaseCreate {
-                        db_id: createdb.db_id,
-                        tablespace_id: createdb.tablespace_id,
-                        src_db_id: createdb.src_db_id,
-                        src_tablespace_id: createdb.src_tablespace_id,
-                    }));
+                let createdb = XlCreateDatabase::decode(buf);
+                let record = MetadataRecord::Dbase(DbaseRecord::Create(DbaseCreate {
+                    db_id: createdb.db_id,
+                    tablespace_id: createdb.tablespace_id,
+                    src_db_id: createdb.src_db_id,
+                    src_tablespace_id: createdb.src_tablespace_id,
+                }));

-                    return Ok(Some(record));
-                } else if info == postgres_ffi::v16::bindings::XLOG_DBASE_DROP {
-                    let dropdb = XlDropDatabase::decode(buf);
-                    let record = MetadataRecord::Dbase(DbaseRecord::Drop(DbaseDrop {
-                        db_id: dropdb.db_id,
-                        tablespace_ids: dropdb.tablespace_ids,
-                    }));
+                return Ok(Some(record));
+            } else if info == postgres_ffi::v16::bindings::XLOG_DBASE_DROP {
+                let dropdb = XlDropDatabase::decode(buf);
+                let record = MetadataRecord::Dbase(DbaseRecord::Drop(DbaseDrop {
+                    db_id: dropdb.db_id,
+                    tablespace_ids: dropdb.tablespace_ids,
+                }));

-                    return Ok(Some(record));
-                }
+                return Ok(Some(record));
            }
-            PgMajorVersion::PG17 => {
-                if info == postgres_ffi::v17::bindings::XLOG_DBASE_CREATE_WAL_LOG {
-                    tracing::debug!("XLOG_DBASE_CREATE_WAL_LOG: noop");
-                } else if info == postgres_ffi::v17::bindings::XLOG_DBASE_CREATE_FILE_COPY {
-                    // The XLOG record was renamed between v14 and v15,
-                    // but the record format is the same.
-                    // So we can reuse XlCreateDatabase here.
-                    tracing::debug!("XLOG_DBASE_CREATE_FILE_COPY");
+        } else if pg_version == 17 {
+            if info == postgres_ffi::v17::bindings::XLOG_DBASE_CREATE_WAL_LOG {
+                tracing::debug!("XLOG_DBASE_CREATE_WAL_LOG: noop");
+            } else if info == postgres_ffi::v17::bindings::XLOG_DBASE_CREATE_FILE_COPY {
+                // The XLOG record was renamed between v14 and v15,
+                // but the record format is the same.
+                // So we can reuse XlCreateDatabase here.
+                tracing::debug!("XLOG_DBASE_CREATE_FILE_COPY");

-                    let createdb = XlCreateDatabase::decode(buf);
-                    let record = MetadataRecord::Dbase(DbaseRecord::Create(DbaseCreate {
-                        db_id: createdb.db_id,
-                        tablespace_id: createdb.tablespace_id,
-                        src_db_id: createdb.src_db_id,
-                        src_tablespace_id: createdb.src_tablespace_id,
-                    }));
+                let createdb = XlCreateDatabase::decode(buf);
+                let record = MetadataRecord::Dbase(DbaseRecord::Create(DbaseCreate {
+                    db_id: createdb.db_id,
+                    tablespace_id: createdb.tablespace_id,
+                    src_db_id: createdb.src_db_id,
+                    src_tablespace_id: createdb.src_tablespace_id,
+                }));

-                    return Ok(Some(record));
-                } else if info == postgres_ffi::v17::bindings::XLOG_DBASE_DROP {
-                    let dropdb = XlDropDatabase::decode(buf);
-                    let record = MetadataRecord::Dbase(DbaseRecord::Drop(DbaseDrop {
-                        db_id: dropdb.db_id,
-                        tablespace_ids: dropdb.tablespace_ids,
-                    }));
+                return Ok(Some(record));
+            } else if info == postgres_ffi::v17::bindings::XLOG_DBASE_DROP {
+                let dropdb = XlDropDatabase::decode(buf);
+                let record = MetadataRecord::Dbase(DbaseRecord::Drop(DbaseDrop {
+                    db_id: dropdb.db_id,
+                    tablespace_ids: dropdb.tablespace_ids,
+                }));

-                    return Ok(Some(record));
-                }
+                return Ok(Some(record));
            }
        }

@@ -752,12 +748,12 @@ impl MetadataRecord {
    fn decode_clog_record(
        buf: &mut Bytes,
        decoded: &DecodedWALRecord,
-        pg_version: PgMajorVersion,
+        pg_version: u32,
    ) -> anyhow::Result<Option<MetadataRecord>> {
        let info = decoded.xl_info & !pg_constants::XLR_INFO_MASK;

        if info == pg_constants::CLOG_ZEROPAGE {
-            let pageno = if pg_version < PgMajorVersion::PG17 {
+            let pageno = if pg_version < 17 {
                buf.get_u32_le()
            } else {
                buf.get_u64_le() as u32
@@ -769,7 +765,7 @@ impl MetadataRecord {
                ClogZeroPage { segno, rpageno },
            ))))
        } else {
-            assert_eq!(info, pg_constants::CLOG_TRUNCATE);
+            assert!(info == pg_constants::CLOG_TRUNCATE);
            let xlrec = XlClogTruncate::decode(buf, pg_version);

            Ok(Some(MetadataRecord::Clog(ClogRecord::Truncate(
@@ -842,14 +838,14 @@ impl MetadataRecord {
    fn decode_multixact_record(
        buf: &mut Bytes,
        decoded: &DecodedWALRecord,
-        pg_version: PgMajorVersion,
+        pg_version: u32,
    ) -> anyhow::Result<Option<MetadataRecord>> {
        let info = decoded.xl_info & pg_constants::XLR_RMGR_INFO_MASK;

        if info == pg_constants::XLOG_MULTIXACT_ZERO_OFF_PAGE
            || info == pg_constants::XLOG_MULTIXACT_ZERO_MEM_PAGE
        {
-            let pageno = if pg_version < PgMajorVersion::PG17 {
+            let pageno = if pg_version < 17 {
                buf.get_u32_le()
            } else {
                buf.get_u64_le() as u32
--- a/libs/wal_decoder/src/serialized_batch.rs
+++ b/libs/wal_decoder/src/serialized_batch.rs
@@ -13,7 +13,7 @@ use pageserver_api::keyspace::KeySpace;
 use pageserver_api::reltag::RelTag;
 use pageserver_api::shard::ShardIdentity;
 use postgres_ffi::walrecord::{DecodedBkpBlock, DecodedWALRecord};
-use postgres_ffi::{BLCKSZ, PgMajorVersion, page_is_new, page_set_lsn, pg_constants};
+use postgres_ffi::{BLCKSZ, page_is_new, page_set_lsn, pg_constants};
 use serde::{Deserialize, Serialize};
 use utils::bin_ser::BeSer;
 use utils::lsn::Lsn;
@@ -139,7 +139,7 @@ impl SerializedValueBatch {
        decoded: DecodedWALRecord,
        shard_records: &mut HashMap<ShardIdentity, InterpretedWalRecord>,
        next_record_lsn: Lsn,
-        pg_version: PgMajorVersion,
+        pg_version: u32,
    ) -> anyhow::Result<()> {
        // First determine how big the buffers need to be and allocate it up-front.
        // This duplicates some of the work below, but it's empirically much faster.
@@ -267,7 +267,7 @@ impl SerializedValueBatch {
    fn estimate_buffer_size(
        decoded: &DecodedWALRecord,
        shard: &ShardIdentity,
-        pg_version: PgMajorVersion,
+        pg_version: u32,
    ) -> usize {
        let mut estimate: usize = 0;

@@ -303,11 +303,7 @@ impl SerializedValueBatch {
        estimate
    }

-    fn block_is_image(
-        decoded: &DecodedWALRecord,
-        blk: &DecodedBkpBlock,
-        pg_version: PgMajorVersion,
-    ) -> bool {
+    fn block_is_image(decoded: &DecodedWALRecord, blk: &DecodedBkpBlock, pg_version: u32) -> bool {
        blk.apply_image
            && blk.has_image
            && decoded.xl_rmid == pg_constants::RM_XLOG_ID
--- a/pageserver/benches/bench_walredo.rs
+++ b/pageserver/benches/bench_walredo.rs
--- a/pageserver/client/Cargo.toml
+++ b/pageserver/client/Cargo.toml
@@ -18,7 +18,6 @@ workspace_hack = { version = "0.1", path = "../../workspace_hack" }
 tokio-postgres.workspace = true
 tokio-stream.workspace = true
 tokio.workspace = true
-postgres_versioninfo.workspace = true
 futures.workspace = true
 tokio-util.workspace = true
 anyhow.workspace = true
--- a/pageserver/client/src/mgmt_api.rs
+++ b/pageserver/client/src/mgmt_api.rs
@@ -7,7 +7,6 @@ use detach_ancestor::AncestorDetached;
 use http_utils::error::HttpErrorBody;
 use pageserver_api::models::*;
 use pageserver_api::shard::TenantShardId;
-use postgres_versioninfo::PgMajorVersion;
 pub use reqwest::Body as ReqwestBody;
 use reqwest::{IntoUrl, Method, StatusCode, Url};
 use utils::id::{TenantId, TimelineId};
@@ -746,11 +745,9 @@ impl Client {
        timeline_id: TimelineId,
        base_lsn: Lsn,
        end_lsn: Lsn,
-        pg_version: PgMajorVersion,
+        pg_version: u32,
        basebackup_tarball: ReqwestBody,
    ) -> Result<()> {
-        let pg_version = pg_version.major_version_num();
-
        let uri = format!(
            "{}/v1/tenant/{tenant_id}/timeline/{timeline_id}/import_basebackup?base_lsn={base_lsn}&end_lsn={end_lsn}&pg_version={pg_version}",
            self.mgmt_api_endpoint,
@@ -844,13 +841,4 @@ impl Client {
            .await
            .map_err(Error::ReceiveBody)
    }
-
-    pub async fn update_feature_flag_spec(&self, spec: String) -> Result<()> {
-        let uri = format!("{}/v1/feature_flag_spec", self.mgmt_api_endpoint);
-        self.request(Method::POST, uri, spec)
-            .await?
-            .json()
-            .await
-            .map_err(Error::ReceiveBody)
-    }
 }
--- a/pageserver/page_api/Cargo.toml
+++ b/pageserver/page_api/Cargo.toml
@@ -9,14 +9,12 @@ anyhow.workspace = true
 bytes.workspace = true
 futures.workspace = true
 pageserver_api.workspace = true
-postgres_ffi_types.workspace = true
+postgres_ffi.workspace = true
 prost.workspace = true
-prost-types.workspace = true
 strum.workspace = true
 strum_macros.workspace = true
 thiserror.workspace = true
 tokio.workspace = true
-tokio-util.workspace = true
 tonic.workspace = true
 utils.workspace = true
 workspace_hack.workspace = true
--- a/pageserver/page_api/proto/page_service.proto
+++ b/pageserver/page_api/proto/page_service.proto
@@ -35,8 +35,6 @@
 syntax = "proto3";
 package page_api;

-import "google/protobuf/timestamp.proto";
-
 service PageService {
  // Returns whether a relation exists.
  rpc CheckRelExists(CheckRelExistsRequest) returns (CheckRelExistsResponse);
@@ -66,10 +64,6 @@ service PageService {

  // Fetches an SLRU segment.
  rpc GetSlruSegment (GetSlruSegmentRequest) returns (GetSlruSegmentResponse);
-
-  // Acquires or extends a lease on the given LSN. This guarantees that the Pageserver won't garbage
-  // collect the LSN until the lease expires. Must be acquired on all relevant shards.
-  rpc LeaseLsn (LeaseLsnRequest) returns (LeaseLsnResponse);
 }

 // The LSN a request should read at.
@@ -116,19 +110,6 @@ message GetBaseBackupRequest {
  bool replica = 2;
  // If true, include relation files in the base backup. Mainly for debugging and tests.
  bool full = 3;
-  // Compression algorithm to use. Base backups send a compressed payload instead of using gRPC
-  // compression, so that we can cache compressed backups on the server.
-  BaseBackupCompression compression = 4;
-}
-
-// Base backup compression algorithms.
-enum BaseBackupCompression {
-  // Unknown algorithm. Used when clients send an unsupported algorithm.
-  BASE_BACKUP_COMPRESSION_UNKNOWN = 0;
-  // No compression.
-  BASE_BACKUP_COMPRESSION_NONE = 1;
-  // GZIP compression.
-  BASE_BACKUP_COMPRESSION_GZIP = 2;
 }

 // Base backup response chunk, returned as an ordered stream.
@@ -258,17 +239,3 @@ message GetSlruSegmentRequest {
 message GetSlruSegmentResponse {
  bytes segment = 1;
 }
-
-// Acquires or extends a lease on the given LSN. This guarantees that the Pageserver won't garbage
-// collect the LSN until the lease expires. Must be acquired on all relevant shards.
-message LeaseLsnRequest {
-  // The LSN to lease. Can't be 0 or below the current GC cutoff.
-  uint64 lsn = 1;
-}
-
-// Lease acquisition response. If the lease could not be granted because the LSN has already been
-// garbage collected, a FailedPrecondition status will be returned instead.
-message LeaseLsnResponse {
-  // The lease expiration time.
-  google.protobuf.Timestamp expires = 1;
-}
--- a/pageserver/page_api/src/client.rs
+++ b/pageserver/page_api/src/client.rs
@@ -1,7 +1,8 @@
-use anyhow::Result;
-use futures::{Stream, StreamExt as _, TryStreamExt as _};
-use tokio::io::AsyncRead;
-use tokio_util::io::StreamReader;
+use std::convert::TryInto;
+
+use bytes::Bytes;
+use futures::TryStreamExt;
+use futures::{Stream, StreamExt};
 use tonic::metadata::AsciiMetadataValue;
 use tonic::metadata::errors::InvalidMetadataValue;
 use tonic::transport::Channel;
@@ -11,6 +12,8 @@ use utils::id::TenantId;
 use utils::id::TimelineId;
 use utils::shard::ShardIndex;

+use anyhow::Result;
+
 use crate::model;
 use crate::proto;

@@ -66,7 +69,6 @@ impl tonic::service::Interceptor for AuthInterceptor {
        Ok(req)
    }
 }
-
 #[derive(Clone)]
 pub struct Client {
    client: proto::PageServiceClient<
@@ -93,6 +95,7 @@ impl Client {

        if let Some(compression) = compression {
            // TODO: benchmark this (including network latency).
+            // TODO: consider enabling compression by default.
            client = client
                .accept_compressed(compression)
                .send_compressed(compression);
@@ -118,15 +121,22 @@ impl Client {
    pub async fn get_base_backup(
        &mut self,
        req: model::GetBaseBackupRequest,
-    ) -> Result<impl AsyncRead + use<>, tonic::Status> {
-        let req = proto::GetBaseBackupRequest::from(req);
-        let chunks = self.client.get_base_backup(req).await?.into_inner();
-        let reader = StreamReader::new(
-            chunks
-                .map_ok(|resp| resp.chunk)
-                .map_err(std::io::Error::other),
-        );
-        Ok(reader)
+    ) -> Result<impl Stream<Item = Result<Bytes, tonic::Status>> + 'static, tonic::Status> {
+        let proto_req = proto::GetBaseBackupRequest::from(req);
+
+        let response_stream: Streaming<proto::GetBaseBackupResponseChunk> =
+            self.client.get_base_backup(proto_req).await?.into_inner();
+
+        // TODO: Consider dechunking internally
+        let domain_stream = response_stream.map(|chunk_res| {
+            chunk_res.and_then(|proto_chunk| {
+                proto_chunk.try_into().map_err(|e| {
+                    tonic::Status::internal(format!("Failed to convert response chunk: {e}"))
+                })
+            })
+        });
+
+        Ok(domain_stream)
    }

    /// Returns the total size of a database, as # of bytes.
@@ -187,17 +197,4 @@ impl Client {
        let response = self.client.get_slru_segment(proto_req).await?;
        Ok(response.into_inner().try_into()?)
    }
-
-    /// Acquires or extends a lease on the given LSN. This guarantees that the Pageserver won't
-    /// garbage collect the LSN until the lease expires. Must be acquired on all relevant shards.
-    ///
-    /// Returns the lease expiration time, or a FailedPrecondition status if the lease could not be
-    /// acquired because the LSN has already been garbage collected.
-    pub async fn lease_lsn(
-        &mut self,
-        req: model::LeaseLsnRequest,
-    ) -> Result<model::LeaseLsnResponse, tonic::Status> {
-        let req = proto::LeaseLsnRequest::from(req);
-        Ok(self.client.lease_lsn(req).await?.into_inner().try_into()?)
-    }
 }
--- a/pageserver/page_api/src/model.rs
+++ b/pageserver/page_api/src/model.rs
@@ -16,11 +16,10 @@
 //! stream combinators without dealing with errors, and avoids validating the same message twice.

 use std::fmt::Display;
-use std::time::{Duration, SystemTime, UNIX_EPOCH};

 use bytes::Bytes;
-use postgres_ffi_types::Oid;
-// TODO: split out Lsn, RelTag, SlruKind and other basic types to a separate crate, to avoid
+use postgres_ffi::Oid;
+// TODO: split out Lsn, RelTag, SlruKind, Oid and other basic types to a separate crate, to avoid
 // pulling in all of their other crate dependencies when building the client.
 use utils::lsn::Lsn;

@@ -192,21 +191,15 @@ pub struct GetBaseBackupRequest {
    pub replica: bool,
    /// If true, include relation files in the base backup. Mainly for debugging and tests.
    pub full: bool,
-    /// Compression algorithm to use. Base backups send a compressed payload instead of using gRPC
-    /// compression, so that we can cache compressed backups on the server.
-    pub compression: BaseBackupCompression,
 }

-impl TryFrom<proto::GetBaseBackupRequest> for GetBaseBackupRequest {
-    type Error = ProtocolError;
-
-    fn try_from(pb: proto::GetBaseBackupRequest) -> Result<Self, Self::Error> {
-        Ok(Self {
+impl From<proto::GetBaseBackupRequest> for GetBaseBackupRequest {
+    fn from(pb: proto::GetBaseBackupRequest) -> Self {
+        Self {
            lsn: (pb.lsn != 0).then_some(Lsn(pb.lsn)),
            replica: pb.replica,
            full: pb.full,
-            compression: pb.compression.try_into()?,
-        })
+        }
    }
 }

@@ -216,55 +209,10 @@ impl From<GetBaseBackupRequest> for proto::GetBaseBackupRequest {
            lsn: request.lsn.unwrap_or_default().0,
            replica: request.replica,
            full: request.full,
-            compression: request.compression.into(),
        }
    }
 }

-/// Base backup compression algorithm.
-#[derive(Clone, Copy, Debug)]
-pub enum BaseBackupCompression {
-    None,
-    Gzip,
-}
-
-impl TryFrom<proto::BaseBackupCompression> for BaseBackupCompression {
-    type Error = ProtocolError;
-
-    fn try_from(pb: proto::BaseBackupCompression) -> Result<Self, Self::Error> {
-        match pb {
-            proto::BaseBackupCompression::Unknown => Err(ProtocolError::invalid("compression", pb)),
-            proto::BaseBackupCompression::None => Ok(Self::None),
-            proto::BaseBackupCompression::Gzip => Ok(Self::Gzip),
-        }
-    }
-}
-
-impl TryFrom<i32> for BaseBackupCompression {
-    type Error = ProtocolError;
-
-    fn try_from(compression: i32) -> Result<Self, Self::Error> {
-        proto::BaseBackupCompression::try_from(compression)
-            .map_err(|_| ProtocolError::invalid("compression", compression))
-            .and_then(Self::try_from)
-    }
-}
-
-impl From<BaseBackupCompression> for proto::BaseBackupCompression {
-    fn from(compression: BaseBackupCompression) -> Self {
-        match compression {
-            BaseBackupCompression::None => Self::None,
-            BaseBackupCompression::Gzip => Self::Gzip,
-        }
-    }
-}
-
-impl From<BaseBackupCompression> for i32 {
-    fn from(compression: BaseBackupCompression) -> Self {
-        proto::BaseBackupCompression::from(compression).into()
-    }
-}
-
 pub type GetBaseBackupResponseChunk = Bytes;

 impl TryFrom<proto::GetBaseBackupResponseChunk> for GetBaseBackupResponseChunk {
@@ -704,54 +652,3 @@ impl From<GetSlruSegmentResponse> for proto::GetSlruSegmentResponse {

 // SlruKind is defined in pageserver_api::reltag.
 pub type SlruKind = pageserver_api::reltag::SlruKind;
-
-/// Acquires or extends a lease on the given LSN. This guarantees that the Pageserver won't garbage
-/// collect the LSN until the lease expires.
-pub struct LeaseLsnRequest {
-    /// The LSN to lease.
-    pub lsn: Lsn,
-}
-
-impl TryFrom<proto::LeaseLsnRequest> for LeaseLsnRequest {
-    type Error = ProtocolError;
-
-    fn try_from(pb: proto::LeaseLsnRequest) -> Result<Self, Self::Error> {
-        if pb.lsn == 0 {
-            return Err(ProtocolError::Missing("lsn"));
-        }
-        Ok(Self { lsn: Lsn(pb.lsn) })
-    }
-}
-
-impl From<LeaseLsnRequest> for proto::LeaseLsnRequest {
-    fn from(request: LeaseLsnRequest) -> Self {
-        Self { lsn: request.lsn.0 }
-    }
-}
-
-/// Lease expiration time. If the lease could not be granted because the LSN has already been
-/// garbage collected, a FailedPrecondition status will be returned instead.
-pub type LeaseLsnResponse = SystemTime;
-
-impl TryFrom<proto::LeaseLsnResponse> for LeaseLsnResponse {
-    type Error = ProtocolError;
-
-    fn try_from(pb: proto::LeaseLsnResponse) -> Result<Self, Self::Error> {
-        let expires = pb.expires.ok_or(ProtocolError::Missing("expires"))?;
-        UNIX_EPOCH
-            .checked_add(Duration::new(expires.seconds as u64, expires.nanos as u32))
-            .ok_or_else(|| ProtocolError::invalid("expires", expires))
-    }
-}
-
-impl From<LeaseLsnResponse> for proto::LeaseLsnResponse {
-    fn from(response: LeaseLsnResponse) -> Self {
-        let expires = response.duration_since(UNIX_EPOCH).unwrap_or_default();
-        Self {
-            expires: Some(prost_types::Timestamp {
-                seconds: expires.as_secs() as i64,
-                nanos: expires.subsec_nanos() as i32,
-            }),
-        }
-    }
-}
--- a/pageserver/pagebench/src/cmd/basebackup.rs
+++ b/pageserver/pagebench/src/cmd/basebackup.rs
@@ -317,7 +317,6 @@ impl Client for LibpqClient {
 /// A gRPC Pageserver client.
 struct GrpcClient {
    inner: page_api::Client,
-    compression: page_api::BaseBackupCompression,
 }

 impl GrpcClient {
@@ -332,14 +331,10 @@ impl GrpcClient {
            ttid.timeline_id,
            ShardIndex::unsharded(),
            None,
-            None, // NB: uses payload compression
+            compression.then_some(tonic::codec::CompressionEncoding::Zstd),
        )
        .await?;
-        let compression = match compression {
-            true => page_api::BaseBackupCompression::Gzip,
-            false => page_api::BaseBackupCompression::None,
-        };
-        Ok(Self { inner, compression })
+        Ok(Self { inner })
    }
 }

@@ -353,8 +348,10 @@ impl Client for GrpcClient {
            lsn,
            replica: false,
            full: false,
-            compression: self.compression,
        };
-        Ok(Box::pin(self.inner.get_base_backup(req).await?))
+        let stream = self.inner.get_base_backup(req).await?;
+        Ok(Box::pin(StreamReader::new(
+            stream.map_err(std::io::Error::other),
+        )))
    }
 }
--- a/pageserver/src/basebackup.rs
+++ b/pageserver/src/basebackup.rs
@@ -14,19 +14,18 @@ use std::fmt::Write as FmtWrite;
 use std::time::{Instant, SystemTime};

 use anyhow::{Context, anyhow};
-use async_compression::tokio::write::GzipEncoder;
 use bytes::{BufMut, Bytes, BytesMut};
 use fail::fail_point;
 use pageserver_api::key::{Key, rel_block_to_key};
 use pageserver_api::reltag::{RelTag, SlruKind};
 use postgres_ffi::pg_constants::{PG_HBA, PGDATA_SPECIAL_FILES};
 use postgres_ffi::{
-    BLCKSZ, PG_TLI, PgMajorVersion, RELSEG_SIZE, WAL_SEGMENT_SIZE, XLogFileName,
-    dispatch_pgversion, pg_constants,
+    BLCKSZ, PG_TLI, RELSEG_SIZE, WAL_SEGMENT_SIZE, XLogFileName, dispatch_pgversion, pg_constants,
 };
 use postgres_ffi_types::constants::{DEFAULTTABLESPACE_OID, GLOBALTABLESPACE_OID};
 use postgres_ffi_types::forknum::{INIT_FORKNUM, MAIN_FORKNUM};
-use tokio::io::{self, AsyncWrite, AsyncWriteExt as _};
+use tokio::io;
+use tokio::io::AsyncWrite;
 use tokio_tar::{Builder, EntryType, Header};
 use tracing::*;
 use utils::lsn::Lsn;
@@ -97,7 +96,6 @@ impl From<BasebackupError> for tonic::Status {
 ///  * 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,
@@ -105,7 +103,6 @@ pub async fn send_basebackup_tarball<'a, W>(
    prev_lsn: Option<Lsn>,
    full_backup: bool,
    replica: bool,
-    gzip_level: Option<async_compression::Level>,
    ctx: &'a RequestContext,
 ) -> Result<(), BasebackupError>
 where
@@ -124,7 +121,7 @@ where
    // prev_lsn value; that happens if the timeline was just branched from
    // an old LSN and it doesn't have any WAL of its own yet. We will set
    // prev_lsn to Lsn(0) if we cannot provide the correct value.
-    let (backup_prev, lsn) = if let Some(req_lsn) = req_lsn {
+    let (backup_prev, backup_lsn) = if let Some(req_lsn) = req_lsn {
        // Backup was requested at a particular LSN. The caller should've
        // already checked that it's a valid LSN.

@@ -145,7 +142,7 @@ where
    };

    // Consolidate the derived and the provided prev_lsn values
-    let prev_record_lsn = if let Some(provided_prev_lsn) = prev_lsn {
+    let prev_lsn = if let Some(provided_prev_lsn) = prev_lsn {
        if backup_prev != Lsn(0) && backup_prev != provided_prev_lsn {
            return Err(BasebackupError::Server(anyhow!(
                "backup_prev {backup_prev} != provided_prev_lsn {provided_prev_lsn}"
@@ -157,55 +154,30 @@ where
    };

    info!(
-        "taking basebackup lsn={lsn}, prev_lsn={prev_record_lsn} \
-        (full_backup={full_backup}, replica={replica}, gzip={gzip_level:?})",
-    );
-    let span = info_span!("send_tarball", backup_lsn=%lsn);
-
-    let io_concurrency = IoConcurrency::spawn_from_conf(
-        timeline.conf.get_vectored_concurrent_io,
-        timeline
-            .gate
-            .enter()
-            .map_err(|_| BasebackupError::Shutdown)?,
+        "taking basebackup lsn={}, prev_lsn={} (full_backup={}, replica={})",
+        backup_lsn, prev_lsn, full_backup, replica
    );

-    if let Some(gzip_level) = gzip_level {
-        let mut encoder = GzipEncoder::with_quality(write, gzip_level);
-        Basebackup {
-            ar: Builder::new_non_terminated(&mut encoder),
-            timeline,
-            lsn,
-            prev_record_lsn,
-            full_backup,
-            replica,
-            ctx,
-            io_concurrency,
-        }
+    let basebackup = Basebackup {
+        ar: Builder::new_non_terminated(write),
+        timeline,
+        lsn: backup_lsn,
+        prev_record_lsn: prev_lsn,
+        full_backup,
+        replica,
+        ctx,
+        io_concurrency: IoConcurrency::spawn_from_conf(
+            timeline.conf.get_vectored_concurrent_io,
+            timeline
+                .gate
+                .enter()
+                .map_err(|_| BasebackupError::Shutdown)?,
+        ),
+    };
+    basebackup
        .send_tarball()
-        .instrument(span)
-        .await?;
-        encoder
-            .shutdown()
-            .await
-            .map_err(|err| BasebackupError::Client(err, "gzip"))?;
-    } else {
-        Basebackup {
-            ar: Builder::new_non_terminated(write),
-            timeline,
-            lsn,
-            prev_record_lsn,
-            full_backup,
-            replica,
-            ctx,
-            io_concurrency,
-        }
-        .send_tarball()
-        .instrument(span)
-        .await?;
-    }
-
-    Ok(())
+        .instrument(info_span!("send_tarball", backup_lsn=%backup_lsn))
+        .await
 }

 /// This is short-living object only for the time of tarball creation,
@@ -647,7 +619,10 @@ where
        };

        if spcnode == GLOBALTABLESPACE_OID {
-            let pg_version_str = self.timeline.pg_version.versionfile_string();
+            let pg_version_str = match self.timeline.pg_version {
+                14 | 15 => self.timeline.pg_version.to_string(),
+                ver => format!("{ver}\x0A"),
+            };
            let header = new_tar_header("PG_VERSION", pg_version_str.len() as u64)?;
            self.ar
                .append(&header, pg_version_str.as_bytes())
@@ -704,7 +679,10 @@ where
            if let Some(img) = relmap_img {
                let dst_path = format!("base/{dbnode}/PG_VERSION");

-                let pg_version_str = self.timeline.pg_version.versionfile_string();
+                let pg_version_str = match self.timeline.pg_version {
+                    14 | 15 => self.timeline.pg_version.to_string(),
+                    ver => format!("{ver}\x0A"),
+                };
                let header = new_tar_header(&dst_path, pg_version_str.len() as u64)?;
                self.ar
                    .append(&header, pg_version_str.as_bytes())
@@ -735,7 +713,7 @@ where
        buf.extend_from_slice(&img[..]);
        let crc = crc32c::crc32c(&img[..]);
        buf.put_u32_le(crc);
-        let path = if self.timeline.pg_version < PgMajorVersion::PG17 {
+        let path = if self.timeline.pg_version < 17 {
            format!("pg_twophase/{xid:>08X}")
        } else {
            format!("pg_twophase/{xid:>016X}")
--- a/pageserver/src/basebackup_cache.rs
+++ b/pageserver/src/basebackup_cache.rs
@@ -1,12 +1,13 @@
 use std::{collections::HashMap, sync::Arc};

 use anyhow::Context;
+use async_compression::tokio::write::GzipEncoder;
 use camino::{Utf8Path, Utf8PathBuf};
 use metrics::core::{AtomicU64, GenericCounter};
 use pageserver_api::{config::BasebackupCacheConfig, models::TenantState};
 use tokio::{
    io::{AsyncWriteExt, BufWriter},
-    sync::mpsc::{Receiver, Sender, error::TrySendError},
+    sync::mpsc::{UnboundedReceiver, UnboundedSender},
 };
 use tokio_util::sync::CancellationToken;
 use utils::{
@@ -19,8 +20,8 @@ use crate::{
    basebackup::send_basebackup_tarball,
    context::{DownloadBehavior, RequestContext},
    metrics::{
-        BASEBACKUP_CACHE_ENTRIES, BASEBACKUP_CACHE_PREPARE, BASEBACKUP_CACHE_PREPARE_QUEUE_SIZE,
-        BASEBACKUP_CACHE_READ, BASEBACKUP_CACHE_SIZE,
+        BASEBACKUP_CACHE_ENTRIES, BASEBACKUP_CACHE_PREPARE, BASEBACKUP_CACHE_READ,
+        BASEBACKUP_CACHE_SIZE,
    },
    task_mgr::TaskKind,
    tenant::{
@@ -35,8 +36,8 @@ pub struct BasebackupPrepareRequest {
    pub lsn: Lsn,
 }

-pub type BasebackupPrepareSender = Sender<BasebackupPrepareRequest>;
-pub type BasebackupPrepareReceiver = Receiver<BasebackupPrepareRequest>;
+pub type BasebackupPrepareSender = UnboundedSender<BasebackupPrepareRequest>;
+pub type BasebackupPrepareReceiver = UnboundedReceiver<BasebackupPrepareRequest>;

 #[derive(Clone)]
 struct CacheEntry {
@@ -60,65 +61,40 @@ struct CacheEntry {
 /// and ~1 RPS for get requests.
 pub struct BasebackupCache {
    data_dir: Utf8PathBuf,
-    config: Option<BasebackupCacheConfig>,

    entries: std::sync::Mutex<HashMap<TenantTimelineId, CacheEntry>>,

-    prepare_sender: BasebackupPrepareSender,
-
    read_hit_count: GenericCounter<AtomicU64>,
    read_miss_count: GenericCounter<AtomicU64>,
    read_err_count: GenericCounter<AtomicU64>,
-
-    prepare_skip_count: GenericCounter<AtomicU64>,
 }

 impl BasebackupCache {
-    /// Create a new BasebackupCache instance.
-    /// Also returns a BasebackupPrepareReceiver which is needed to start
-    /// the background task.
-    /// The cache is initialized from the data_dir in the background task.
-    /// The cache will return `None` for any get requests until the initialization is complete.
-    /// The background task is spawned separately using [`Self::spawn_background_task`]
-    /// to avoid a circular dependency between the cache and the tenant manager.
-    pub fn new(
+    /// Creates a BasebackupCache and spawns the background task.
+    /// The initialization of the cache is performed in the background and does not
+    /// block the caller. The cache will return `None` for any get requests until
+    /// initialization is complete.
+    pub fn spawn(
+        runtime_handle: &tokio::runtime::Handle,
        data_dir: Utf8PathBuf,
        config: Option<BasebackupCacheConfig>,
-    ) -> (Arc<Self>, BasebackupPrepareReceiver) {
-        let chan_size = config.as_ref().map(|c| c.max_size_entries).unwrap_or(1);
-
-        let (prepare_sender, prepare_receiver) = tokio::sync::mpsc::channel(chan_size);
-
+        prepare_receiver: BasebackupPrepareReceiver,
+        tenant_manager: Arc<TenantManager>,
+        cancel: CancellationToken,
+    ) -> Arc<Self> {
        let cache = Arc::new(BasebackupCache {
            data_dir,
-            config,
+
            entries: std::sync::Mutex::new(HashMap::new()),
-            prepare_sender,

            read_hit_count: BASEBACKUP_CACHE_READ.with_label_values(&["hit"]),
            read_miss_count: BASEBACKUP_CACHE_READ.with_label_values(&["miss"]),
            read_err_count: BASEBACKUP_CACHE_READ.with_label_values(&["error"]),
-
-            prepare_skip_count: BASEBACKUP_CACHE_PREPARE.with_label_values(&["skip"]),
        });

-        (cache, prepare_receiver)
-    }
-
-    /// Spawns the background task.
-    /// The background task initializes the cache from the disk,
-    /// processes prepare requests, and cleans up outdated cache entries.
-    /// Noop if the cache is disabled (config is None).
-    pub fn spawn_background_task(
-        self: Arc<Self>,
-        runtime_handle: &tokio::runtime::Handle,
-        prepare_receiver: BasebackupPrepareReceiver,
-        tenant_manager: Arc<TenantManager>,
-        cancel: CancellationToken,
-    ) {
-        if let Some(config) = self.config.clone() {
+        if let Some(config) = config {
            let background = BackgroundTask {
-                c: self,
+                c: cache.clone(),

                config,
                tenant_manager,
@@ -133,45 +109,8 @@ impl BasebackupCache {
            };
            runtime_handle.spawn(background.run(prepare_receiver));
        }
-    }

-    /// Send a basebackup prepare request to the background task.
-    /// The basebackup will be prepared asynchronously, it does not block the caller.
-    /// The request will be skipped if any cache limits are exceeded.
-    pub fn send_prepare(&self, tenant_shard_id: TenantShardId, timeline_id: TimelineId, lsn: Lsn) {
-        let req = BasebackupPrepareRequest {
-            tenant_shard_id,
-            timeline_id,
-            lsn,
-        };
-
-        BASEBACKUP_CACHE_PREPARE_QUEUE_SIZE.inc();
-        let res = self.prepare_sender.try_send(req);
-
-        if let Err(e) = res {
-            BASEBACKUP_CACHE_PREPARE_QUEUE_SIZE.dec();
-            self.prepare_skip_count.inc();
-            match e {
-                TrySendError::Full(_) => {
-                    // Basebackup prepares are pretty rare, normally we should not hit this.
-                    tracing::info!(
-                        tenant_id = %tenant_shard_id.tenant_id,
-                        %timeline_id,
-                        %lsn,
-                        "Basebackup prepare channel is full, skipping the request"
-                    );
-                }
-                TrySendError::Closed(_) => {
-                    // Normal during shutdown, not critical.
-                    tracing::info!(
-                        tenant_id = %tenant_shard_id.tenant_id,
-                        %timeline_id,
-                        %lsn,
-                        "Basebackup prepare channel is closed, skipping the request"
-                    );
-                }
-            }
-        }
+        cache
    }

    /// Gets a basebackup entry from the cache.
@@ -184,10 +123,6 @@ impl BasebackupCache {
        timeline_id: TimelineId,
        lsn: Lsn,
    ) -> Option<tokio::fs::File> {
-        if !self.is_enabled() {
-            return None;
-        }
-
        // Fast path. Check if the entry exists using the in-memory state.
        let tti = TenantTimelineId::new(tenant_id, timeline_id);
        if self.entries.lock().unwrap().get(&tti).map(|e| e.lsn) != Some(lsn) {
@@ -215,10 +150,6 @@ impl BasebackupCache {
        }
    }

-    pub fn is_enabled(&self) -> bool {
-        self.config.is_some()
-    }
-
    // Private methods.

    fn entry_filename(tenant_id: TenantId, timeline_id: TimelineId, lsn: Lsn) -> String {
@@ -436,7 +367,6 @@ impl BackgroundTask {
        loop {
            tokio::select! {
                Some(req) = prepare_receiver.recv() => {
-                    BASEBACKUP_CACHE_PREPARE_QUEUE_SIZE.dec();
                    if let Err(err) = self.prepare_basebackup(
                        req.tenant_shard_id,
                        req.timeline_id,
@@ -664,6 +594,13 @@ impl BackgroundTask {
        let file = tokio::fs::File::create(entry_tmp_path).await?;
        let mut writer = BufWriter::new(file);

+        let mut encoder = GzipEncoder::with_quality(
+            &mut writer,
+            // Level::Best because compression is not on the hot path of basebackup requests.
+            // The decompression is almost not affected by the compression level.
+            async_compression::Level::Best,
+        );
+
        // We may receive a request before the WAL record is applied to the timeline.
        // Wait for the requested LSN to be applied.
        timeline
@@ -676,19 +613,17 @@ impl BackgroundTask {
            .await?;

        send_basebackup_tarball(
-            &mut writer,
+            &mut encoder,
            timeline,
            Some(req_lsn),
            None,
            false,
            false,
-            // Level::Best because compression is not on the hot path of basebackup requests.
-            // The decompression is almost not affected by the compression level.
-            Some(async_compression::Level::Best),
            &ctx,
        )
        .await?;

+        encoder.shutdown().await?;
        writer.flush().await?;
        writer.into_inner().sync_all().await?;

--- a/pageserver/src/bin/pageserver.rs
+++ b/pageserver/src/bin/pageserver.rs
@@ -569,10 +569,8 @@ fn start_pageserver(
        pageserver::l0_flush::L0FlushGlobalState::new(conf.l0_flush.clone());

    // Scan the local 'tenants/' directory and start loading the tenants
-    let (basebackup_cache, basebackup_prepare_receiver) = BasebackupCache::new(
-        conf.basebackup_cache_dir(),
-        conf.basebackup_cache_config.clone(),
-    );
+    let (basebackup_prepare_sender, basebackup_prepare_receiver) =
+        tokio::sync::mpsc::unbounded_channel();
    let deletion_queue_client = deletion_queue.new_client();
    let background_purges = mgr::BackgroundPurges::default();

@@ -584,7 +582,7 @@ fn start_pageserver(
            remote_storage: remote_storage.clone(),
            deletion_queue_client,
            l0_flush_global_state,
-            basebackup_cache: Arc::clone(&basebackup_cache),
+            basebackup_prepare_sender,
            feature_resolver: feature_resolver.clone(),
        },
        shutdown_pageserver.clone(),
@@ -592,8 +590,10 @@ fn start_pageserver(
    let tenant_manager = Arc::new(tenant_manager);
    BACKGROUND_RUNTIME.block_on(mgr::init_tenant_mgr(tenant_manager.clone(), order))?;

-    basebackup_cache.spawn_background_task(
+    let basebackup_cache = BasebackupCache::spawn(
        BACKGROUND_RUNTIME.handle(),
+        conf.basebackup_cache_dir(),
+        conf.basebackup_cache_config.clone(),
        basebackup_prepare_receiver,
        Arc::clone(&tenant_manager),
        shutdown_pageserver.child_token(),
@@ -806,6 +806,7 @@ fn start_pageserver(
        } else {
            None
        },
+        basebackup_cache,
    );

    // Spawn a Pageserver gRPC server task. It will spawn separate tasks for
--- a/pageserver/src/bin/test_helper_slow_client_reads.rs
+++ b/pageserver/src/bin/test_helper_slow_client_reads.rs
@@ -37,7 +37,7 @@ async fn main() -> anyhow::Result<()> {
                not_modified_since: Lsn(23),
            },
            batch_key: 42,
-            message: format!("message {msg}"),
+            message: format!("message {}", msg),
        }));
        let Ok(res) = tokio::time::timeout(Duration::from_secs(10), fut).await else {
            eprintln!("pipe seems full");
--- a/pageserver/src/config.rs
+++ b/pageserver/src/config.rs
@@ -11,7 +11,7 @@ use std::num::NonZeroUsize;
 use std::sync::Arc;
 use std::time::Duration;

-use anyhow::{Context, ensure};
+use anyhow::{Context, bail, ensure};
 use camino::{Utf8Path, Utf8PathBuf};
 use once_cell::sync::OnceCell;
 use pageserver_api::config::{
@@ -22,7 +22,6 @@ use pageserver_api::models::ImageCompressionAlgorithm;
 use pageserver_api::shard::TenantShardId;
 use pem::Pem;
 use postgres_backend::AuthType;
-use postgres_ffi::PgMajorVersion;
 use remote_storage::{RemotePath, RemoteStorageConfig};
 use reqwest::Url;
 use storage_broker::Uri;
@@ -339,16 +338,20 @@ impl PageServerConf {
    //
    // Postgres distribution paths
    //
-    pub fn pg_distrib_dir(&self, pg_version: PgMajorVersion) -> anyhow::Result<Utf8PathBuf> {
+    pub fn pg_distrib_dir(&self, pg_version: u32) -> anyhow::Result<Utf8PathBuf> {
        let path = self.pg_distrib_dir.clone();

-        Ok(path.join(pg_version.v_str()))
+        #[allow(clippy::manual_range_patterns)]
+        match pg_version {
+            14 | 15 | 16 | 17 => Ok(path.join(format!("v{pg_version}"))),
+            _ => bail!("Unsupported postgres version: {}", pg_version),
+        }
    }

-    pub fn pg_bin_dir(&self, pg_version: PgMajorVersion) -> anyhow::Result<Utf8PathBuf> {
+    pub fn pg_bin_dir(&self, pg_version: u32) -> anyhow::Result<Utf8PathBuf> {
        Ok(self.pg_distrib_dir(pg_version)?.join("bin"))
    }
-    pub fn pg_lib_dir(&self, pg_version: PgMajorVersion) -> anyhow::Result<Utf8PathBuf> {
+    pub fn pg_lib_dir(&self, pg_version: u32) -> anyhow::Result<Utf8PathBuf> {
        Ok(self.pg_distrib_dir(pg_version)?.join("lib"))
    }

@@ -762,40 +765,4 @@ mod tests {
        let result = PageServerConf::parse_and_validate(NodeId(0), config_toml, &workdir);
        assert_eq!(result.is_ok(), is_valid);
    }
-
-    #[test]
-    fn test_config_posthog_config_is_valid() {
-        let input = r#"
-            control_plane_api = "http://localhost:6666"
-
-            [posthog_config]
-            server_api_key = "phs_AAA"
-            client_api_key = "phc_BBB"
-            project_id = "000"
-            private_api_url = "https://us.posthog.com"
-            public_api_url = "https://us.i.posthog.com"
-        "#;
-        let config_toml = toml_edit::de::from_str::<pageserver_api::config::ConfigToml>(input)
-            .expect("posthogconfig is valid");
-        let workdir = Utf8PathBuf::from("/nonexistent");
-        PageServerConf::parse_and_validate(NodeId(0), config_toml, &workdir)
-            .expect("parse_and_validate");
-    }
-
-    #[test]
-    fn test_config_posthog_incomplete_config_is_valid() {
-        let input = r#"
-            control_plane_api = "http://localhost:6666"
-
-            [posthog_config]
-            server_api_key = "phs_AAA"
-            private_api_url = "https://us.posthog.com"
-            public_api_url = "https://us.i.posthog.com"
-        "#;
-        let config_toml = toml_edit::de::from_str::<pageserver_api::config::ConfigToml>(input)
-            .expect("posthogconfig is valid");
-        let workdir = Utf8PathBuf::from("/nonexistent");
-        PageServerConf::parse_and_validate(NodeId(0), config_toml, &workdir)
-            .expect("parse_and_validate");
-    }
 }
--- a/pageserver/src/feature_resolver.rs
+++ b/pageserver/src/feature_resolver.rs
@@ -3,16 +3,15 @@ use std::{collections::HashMap, sync::Arc, time::Duration};
 use arc_swap::ArcSwap;
 use pageserver_api::config::NodeMetadata;
 use posthog_client_lite::{
-    CaptureEvent, FeatureResolverBackgroundLoop, PostHogEvaluationError,
+    CaptureEvent, FeatureResolverBackgroundLoop, PostHogClientConfig, PostHogEvaluationError,
    PostHogFlagFilterPropertyValue,
 };
-use rand::Rng;
 use remote_storage::RemoteStorageKind;
 use serde_json::json;
 use tokio_util::sync::CancellationToken;
 use utils::id::TenantId;

-use crate::{config::PageServerConf, metrics::FEATURE_FLAG_EVALUATION, tenant::TenantShard};
+use crate::{config::PageServerConf, metrics::FEATURE_FLAG_EVALUATION};

 const DEFAULT_POSTHOG_REFRESH_INTERVAL: Duration = Duration::from_secs(600);

@@ -32,13 +31,6 @@ impl FeatureResolver {
        }
    }

-    pub fn update(&self, spec: String) -> anyhow::Result<()> {
-        if let Some(inner) = &self.inner {
-            inner.update(spec)?;
-        }
-        Ok(())
-    }
-
    pub fn spawn(
        conf: &PageServerConf,
        shutdown_pageserver: CancellationToken,
@@ -46,24 +38,16 @@ impl FeatureResolver {
    ) -> anyhow::Result<Self> {
        // DO NOT block in this function: make it return as fast as possible to avoid startup delays.
        if let Some(posthog_config) = &conf.posthog_config {
-            let posthog_client_config = match posthog_config.clone().try_into_posthog_config() {
-                Ok(config) => config,
-                Err(e) => {
-                    tracing::warn!(
-                        "invalid posthog config, skipping posthog integration: {}",
-                        e
-                    );
-                    return Ok(FeatureResolver {
-                        inner: None,
-                        internal_properties: None,
-                        force_overrides_for_testing: Arc::new(ArcSwap::new(Arc::new(
-                            HashMap::new(),
-                        ))),
-                    });
-                }
-            };
-            let inner =
-                FeatureResolverBackgroundLoop::new(posthog_client_config, shutdown_pageserver);
+            let inner = FeatureResolverBackgroundLoop::new(
+                PostHogClientConfig {
+                    server_api_key: posthog_config.server_api_key.clone(),
+                    client_api_key: posthog_config.client_api_key.clone(),
+                    project_id: posthog_config.project_id.clone(),
+                    private_api_url: posthog_config.private_api_url.clone(),
+                    public_api_url: posthog_config.public_api_url.clone(),
+                },
+                shutdown_pageserver,
+            );
            let inner = Arc::new(inner);

            // The properties shared by all tenants on this pageserver.
@@ -139,7 +123,6 @@ impl FeatureResolver {
                }
                Arc::new(properties)
            };
-
            let fake_tenants = {
                let mut tenants = Vec::new();
                for i in 0..10 {
@@ -149,16 +132,9 @@ impl FeatureResolver {
                        conf.id,
                        i
                    );
-
-                    let tenant_properties = PerTenantProperties {
-                        remote_size_mb: Some(rand::thread_rng().gen_range(100.0..1000000.00)),
-                    }
-                    .into_posthog_properties();
-
                    let properties = Self::collect_properties_inner(
                        distinct_id.clone(),
                        Some(&internal_properties),
-                        &tenant_properties,
                    );
                    tenants.push(CaptureEvent {
                        event: "initial_tenant_report".to_string(),
@@ -192,7 +168,6 @@ impl FeatureResolver {
    fn collect_properties_inner(
        tenant_id: String,
        internal_properties: Option<&HashMap<String, PostHogFlagFilterPropertyValue>>,
-        tenant_properties: &HashMap<String, PostHogFlagFilterPropertyValue>,
    ) -> HashMap<String, PostHogFlagFilterPropertyValue> {
        let mut properties = HashMap::new();
        if let Some(internal_properties) = internal_properties {
@@ -204,9 +179,6 @@ impl FeatureResolver {
            "tenant_id".to_string(),
            PostHogFlagFilterPropertyValue::String(tenant_id),
        );
-        for (key, value) in tenant_properties.iter() {
-            properties.insert(key.clone(), value.clone());
-        }
        properties
    }

@@ -214,13 +186,8 @@ impl FeatureResolver {
    pub(crate) fn collect_properties(
        &self,
        tenant_id: TenantId,
-        tenant_properties: &HashMap<String, PostHogFlagFilterPropertyValue>,
    ) -> HashMap<String, PostHogFlagFilterPropertyValue> {
-        Self::collect_properties_inner(
-            tenant_id.to_string(),
-            self.internal_properties.as_deref(),
-            tenant_properties,
-        )
+        Self::collect_properties_inner(tenant_id.to_string(), self.internal_properties.as_deref())
    }

    /// Evaluate a multivariate feature flag. Currently, we do not support any properties.
@@ -232,7 +199,6 @@ impl FeatureResolver {
        &self,
        flag_key: &str,
        tenant_id: TenantId,
-        tenant_properties: &HashMap<String, PostHogFlagFilterPropertyValue>,
    ) -> Result<String, PostHogEvaluationError> {
        let force_overrides = self.force_overrides_for_testing.load();
        if let Some(value) = force_overrides.get(flag_key) {
@@ -243,7 +209,7 @@ impl FeatureResolver {
            let res = inner.feature_store().evaluate_multivariate(
                flag_key,
                &tenant_id.to_string(),
-                &self.collect_properties(tenant_id, tenant_properties),
+                &self.collect_properties(tenant_id),
            );
            match &res {
                Ok(value) => {
@@ -276,7 +242,6 @@ impl FeatureResolver {
        &self,
        flag_key: &str,
        tenant_id: TenantId,
-        tenant_properties: &HashMap<String, PostHogFlagFilterPropertyValue>,
    ) -> Result<(), PostHogEvaluationError> {
        let force_overrides = self.force_overrides_for_testing.load();
        if let Some(value) = force_overrides.get(flag_key) {
@@ -291,7 +256,7 @@ impl FeatureResolver {
            let res = inner.feature_store().evaluate_boolean(
                flag_key,
                &tenant_id.to_string(),
-                &self.collect_properties(tenant_id, tenant_properties),
+                &self.collect_properties(tenant_id),
            );
            match &res {
                Ok(()) => {
@@ -337,78 +302,3 @@ impl FeatureResolver {
            .store(Arc::new(force_overrides));
    }
 }
-
-struct PerTenantProperties {
-    pub remote_size_mb: Option<f64>,
-}
-
-impl PerTenantProperties {
-    pub fn into_posthog_properties(self) -> HashMap<String, PostHogFlagFilterPropertyValue> {
-        let mut properties = HashMap::new();
-        if let Some(remote_size_mb) = self.remote_size_mb {
-            properties.insert(
-                "tenant_remote_size_mb".to_string(),
-                PostHogFlagFilterPropertyValue::Number(remote_size_mb),
-            );
-        }
-        properties
-    }
-}
-
-#[derive(Clone)]
-pub struct TenantFeatureResolver {
-    inner: FeatureResolver,
-    tenant_id: TenantId,
-    cached_tenant_properties: Arc<ArcSwap<HashMap<String, PostHogFlagFilterPropertyValue>>>,
-}
-
-impl TenantFeatureResolver {
-    pub fn new(inner: FeatureResolver, tenant_id: TenantId) -> Self {
-        Self {
-            inner,
-            tenant_id,
-            cached_tenant_properties: Arc::new(ArcSwap::new(Arc::new(HashMap::new()))),
-        }
-    }
-
-    pub fn evaluate_multivariate(&self, flag_key: &str) -> Result<String, PostHogEvaluationError> {
-        self.inner.evaluate_multivariate(
-            flag_key,
-            self.tenant_id,
-            &self.cached_tenant_properties.load(),
-        )
-    }
-
-    pub fn evaluate_boolean(&self, flag_key: &str) -> Result<(), PostHogEvaluationError> {
-        self.inner.evaluate_boolean(
-            flag_key,
-            self.tenant_id,
-            &self.cached_tenant_properties.load(),
-        )
-    }
-
-    pub fn collect_properties(&self) -> HashMap<String, PostHogFlagFilterPropertyValue> {
-        self.inner
-            .collect_properties(self.tenant_id, &self.cached_tenant_properties.load())
-    }
-
-    pub fn is_feature_flag_boolean(&self, flag_key: &str) -> Result<bool, PostHogEvaluationError> {
-        self.inner.is_feature_flag_boolean(flag_key)
-    }
-
-    pub fn update_cached_tenant_properties(&self, tenant_shard: &TenantShard) {
-        let mut remote_size_mb = None;
-        for timeline in tenant_shard.list_timelines() {
-            let size = timeline.metrics.resident_physical_size_get();
-            if size == 0 {
-                remote_size_mb = None;
-            }
-            if let Some(ref mut remote_size_mb) = remote_size_mb {
-                *remote_size_mb += size as f64 / 1024.0 / 1024.0;
-            }
-        }
-        self.cached_tenant_properties.store(Arc::new(
-            PerTenantProperties { remote_size_mb }.into_posthog_properties(),
-        ));
-    }
-}
--- a/pageserver/src/http/routes.rs
+++ b/pageserver/src/http/routes.rs
@@ -41,7 +41,6 @@ use pageserver_api::models::{
    TopTenantShardItem, TopTenantShardsRequest, TopTenantShardsResponse,
 };
 use pageserver_api::shard::{ShardCount, TenantShardId};
-use postgres_ffi::PgMajorVersion;
 use remote_storage::{DownloadError, GenericRemoteStorage, TimeTravelError};
 use scopeguard::defer;
 use serde_json::json;
@@ -1893,13 +1892,9 @@ async fn update_tenant_config_handler(
    let location_conf = LocationConf::attached_single(
        new_tenant_conf.clone(),
        tenant.get_generation(),
-        ShardParameters::from(tenant.get_shard_identity()),
+        &ShardParameters::default(),
    );

-    tenant
-        .get_shard_identity()
-        .assert_equal(location_conf.shard); // not strictly necessary since we construct it above
-
    crate::tenant::TenantShard::persist_tenant_config(state.conf, &tenant_shard_id, &location_conf)
        .await
        .map_err(|e| ApiError::InternalServerError(anyhow::anyhow!(e)))?;
@@ -1941,13 +1936,9 @@ async fn patch_tenant_config_handler(
    let location_conf = LocationConf::attached_single(
        updated,
        tenant.get_generation(),
-        ShardParameters::from(tenant.get_shard_identity()),
+        &ShardParameters::default(),
    );

-    tenant
-        .get_shard_identity()
-        .assert_equal(location_conf.shard); // not strictly necessary since we construct it above
-
    crate::tenant::TenantShard::persist_tenant_config(state.conf, &tenant_shard_id, &location_conf)
        .await
        .map_err(|e| ApiError::InternalServerError(anyhow::anyhow!(e)))?;
@@ -3394,7 +3385,7 @@ async fn put_tenant_timeline_import_basebackup(
    let timeline_id: TimelineId = parse_request_param(&request, "timeline_id")?;
    let base_lsn: Lsn = must_parse_query_param(&request, "base_lsn")?;
    let end_lsn: Lsn = must_parse_query_param(&request, "end_lsn")?;
-    let pg_version: PgMajorVersion = must_parse_query_param(&request, "pg_version")?;
+    let pg_version: u32 = must_parse_query_param(&request, "pg_version")?;

    check_permission(&request, Some(tenant_id))?;

@@ -3697,25 +3688,23 @@ async fn tenant_evaluate_feature_flag(
        let tenant = state
            .tenant_manager
            .get_attached_tenant_shard(tenant_shard_id)?;
-        // TODO: the properties we get here might be stale right after it is collected. But such races are rare (updated every 10s) 
-        // and we don't need to worry about it for now.
-        let properties = tenant.feature_resolver.collect_properties();
+        let properties = tenant.feature_resolver.collect_properties(tenant_shard_id.tenant_id);
        if as_type.as_deref() == Some("boolean") {
-            let result = tenant.feature_resolver.evaluate_boolean(&flag);
+            let result = tenant.feature_resolver.evaluate_boolean(&flag, tenant_shard_id.tenant_id);
            let result = result.map(|_| true).map_err(|e| e.to_string());
            json_response(StatusCode::OK, json!({ "result": result, "properties": properties }))
        } else if as_type.as_deref() == Some("multivariate") {
-            let result = tenant.feature_resolver.evaluate_multivariate(&flag).map_err(|e| e.to_string());
+            let result = tenant.feature_resolver.evaluate_multivariate(&flag, tenant_shard_id.tenant_id).map_err(|e| e.to_string());
            json_response(StatusCode::OK, json!({ "result": result, "properties": properties }))
        } else {
            // Auto infer the type of the feature flag.
            let is_boolean = tenant.feature_resolver.is_feature_flag_boolean(&flag).map_err(|e| ApiError::InternalServerError(anyhow::anyhow!("{e}")))?;
            if is_boolean {
-                let result = tenant.feature_resolver.evaluate_boolean(&flag);
+                let result = tenant.feature_resolver.evaluate_boolean(&flag, tenant_shard_id.tenant_id);
                let result = result.map(|_| true).map_err(|e| e.to_string());
                json_response(StatusCode::OK, json!({ "result": result, "properties": properties }))
            } else {
-                let result = tenant.feature_resolver.evaluate_multivariate(&flag).map_err(|e| e.to_string());
+                let result = tenant.feature_resolver.evaluate_multivariate(&flag, tenant_shard_id.tenant_id).map_err(|e| e.to_string());
                json_response(StatusCode::OK, json!({ "result": result, "properties": properties }))
            }
        }
@@ -3753,20 +3742,6 @@ async fn force_override_feature_flag_for_testing_delete(
    json_response(StatusCode::OK, ())
 }

-async fn update_feature_flag_spec(
-    mut request: Request<Body>,
-    _cancel: CancellationToken,
-) -> Result<Response<Body>, ApiError> {
-    check_permission(&request, None)?;
-    let body = json_request(&mut request).await?;
-    let state = get_state(&request);
-    state
-        .feature_resolver
-        .update(body)
-        .map_err(ApiError::InternalServerError)?;
-    json_response(StatusCode::OK, ())
-}
-
 /// Common functionality of all the HTTP API handlers.
 ///
 /// - Adds a tracing span to each request (by `request_span`)
@@ -4152,8 +4127,5 @@ pub fn make_router(
        .delete("/v1/feature_flag/:flag_key", |r| {
            testing_api_handler("force override feature flag - delete", r, force_override_feature_flag_for_testing_delete)
        })
-        .post("/v1/feature_flag_spec", |r| {
-            api_handler(r, update_feature_flag_spec)
-        })
        .any(handler_404))
 }
--- a/pageserver/src/lib.rs
+++ b/pageserver/src/lib.rs
@@ -38,7 +38,6 @@ pub mod walredo;

 use camino::Utf8Path;
 use deletion_queue::DeletionQueue;
-use postgres_ffi::PgMajorVersion;
 use tenant::mgr::{BackgroundPurges, TenantManager};
 use tenant::secondary;
 use tracing::{info, info_span};
@@ -52,7 +51,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: PgMajorVersion = PgMajorVersion::PG17;
+pub const DEFAULT_PG_VERSION: u32 = 17;

 // Magic constants used to identify different kinds of files
 pub const IMAGE_FILE_MAGIC: u16 = 0x5A60;
--- a/pageserver/src/metrics.rs
+++ b/pageserver/src/metrics.rs
@@ -4439,14 +4439,6 @@ pub(crate) static BASEBACKUP_CACHE_SIZE: Lazy<UIntGauge> = Lazy::new(|| {
    .expect("failed to define a metric")
 });

-pub(crate) static BASEBACKUP_CACHE_PREPARE_QUEUE_SIZE: Lazy<UIntGauge> = Lazy::new(|| {
-    register_uint_gauge!(
-        "pageserver_basebackup_cache_prepare_queue_size",
-        "Number of requests in the basebackup prepare channel"
-    )
-    .expect("failed to define a metric")
-});
-
 static PAGESERVER_CONFIG_IGNORED_ITEMS: Lazy<UIntGaugeVec> = Lazy::new(|| {
    register_uint_gauge_vec!(
        "pageserver_config_ignored_items",
--- a/pageserver/src/page_service.rs
+++ b/pageserver/src/page_service.rs
@@ -12,9 +12,9 @@ use std::task::{Context, Poll};
 use std::time::{Duration, Instant, SystemTime};
 use std::{io, str};

-use anyhow::{Context as _, bail};
+use anyhow::{Context as _, anyhow, bail};
+use async_compression::tokio::write::GzipEncoder;
 use bytes::{Buf as _, BufMut as _, BytesMut};
-use chrono::Utc;
 use futures::future::BoxFuture;
 use futures::{FutureExt, Stream};
 use itertools::Itertools;
@@ -63,6 +63,7 @@ use utils::{failpoint_support, span_record};

 use crate::auth::check_permission;
 use crate::basebackup::{self, BasebackupError};
+use crate::basebackup_cache::BasebackupCache;
 use crate::config::PageServerConf;
 use crate::context::{
    DownloadBehavior, PerfInstrumentFutureExt, RequestContext, RequestContextBuilder,
@@ -137,6 +138,7 @@ pub fn spawn(
    perf_trace_dispatch: Option<Dispatch>,
    tcp_listener: tokio::net::TcpListener,
    tls_config: Option<Arc<rustls::ServerConfig>>,
+    basebackup_cache: Arc<BasebackupCache>,
 ) -> Listener {
    let cancel = CancellationToken::new();
    let libpq_ctx = RequestContext::todo_child(
@@ -158,6 +160,7 @@ pub fn spawn(
            conf.pg_auth_type,
            tls_config,
            conf.page_service_pipelining.clone(),
+            basebackup_cache,
            libpq_ctx,
            cancel.clone(),
        )
@@ -216,6 +219,7 @@ pub async fn libpq_listener_main(
    auth_type: AuthType,
    tls_config: Option<Arc<rustls::ServerConfig>>,
    pipelining_config: PageServicePipeliningConfig,
+    basebackup_cache: Arc<BasebackupCache>,
    listener_ctx: RequestContext,
    listener_cancel: CancellationToken,
 ) -> Connections {
@@ -259,6 +263,7 @@ pub async fn libpq_listener_main(
                    auth_type,
                    tls_config.clone(),
                    pipelining_config.clone(),
+                    Arc::clone(&basebackup_cache),
                    connection_ctx,
                    connections_cancel.child_token(),
                    gate_guard,
@@ -301,6 +306,7 @@ async fn page_service_conn_main(
    auth_type: AuthType,
    tls_config: Option<Arc<rustls::ServerConfig>>,
    pipelining_config: PageServicePipeliningConfig,
+    basebackup_cache: Arc<BasebackupCache>,
    connection_ctx: RequestContext,
    cancel: CancellationToken,
    gate_guard: GateGuard,
@@ -366,6 +372,7 @@ async fn page_service_conn_main(
        pipelining_config,
        conf.get_vectored_concurrent_io,
        perf_span_fields,
+        basebackup_cache,
        connection_ctx,
        cancel.clone(),
        gate_guard,
@@ -419,6 +426,8 @@ struct PageServerHandler {
    pipelining_config: PageServicePipeliningConfig,
    get_vectored_concurrent_io: GetVectoredConcurrentIo,

+    basebackup_cache: Arc<BasebackupCache>,
+
    gate_guard: GateGuard,
 }

@@ -904,6 +913,7 @@ impl PageServerHandler {
        pipelining_config: PageServicePipeliningConfig,
        get_vectored_concurrent_io: GetVectoredConcurrentIo,
        perf_span_fields: ConnectionPerfSpanFields,
+        basebackup_cache: Arc<BasebackupCache>,
        connection_ctx: RequestContext,
        cancel: CancellationToken,
        gate_guard: GateGuard,
@@ -917,6 +927,7 @@ impl PageServerHandler {
            cancel,
            pipelining_config,
            get_vectored_concurrent_io,
+            basebackup_cache,
            gate_guard,
        }
    }
@@ -2602,16 +2613,26 @@ impl PageServerHandler {
                prev_lsn,
                full_backup,
                replica,
-                None,
                &ctx,
            )
            .await?;
        } else {
            let mut writer = BufWriter::new(pgb.copyout_writer());

-            let cached = timeline
-                .get_cached_basebackup_if_enabled(lsn, prev_lsn, full_backup, replica, gzip)
-                .await;
+            let cached = {
+                // Basebackup is cached only for this combination of parameters.
+                if timeline.is_basebackup_cache_enabled()
+                    && gzip
+                    && lsn.is_some()
+                    && prev_lsn.is_none()
+                {
+                    self.basebackup_cache
+                        .get(tenant_id, timeline_id, lsn.unwrap())
+                        .await
+                } else {
+                    None
+                }
+            };

            if let Some(mut cached) = cached {
                from_cache = true;
@@ -2620,6 +2641,31 @@ impl PageServerHandler {
                    .map_err(|err| {
                        BasebackupError::Client(err, "handle_basebackup_request,cached,copy")
                    })?;
+            } else if gzip {
+                let mut encoder = GzipEncoder::with_quality(
+                    &mut writer,
+                    // NOTE using fast compression because it's on the critical path
+                    //      for compute startup. For an empty database, we get
+                    //      <100KB with this method. The Level::Best compression method
+                    //      gives us <20KB, but maybe we should add basebackup caching
+                    //      on compute shutdown first.
+                    async_compression::Level::Fastest,
+                );
+                basebackup::send_basebackup_tarball(
+                    &mut encoder,
+                    &timeline,
+                    lsn,
+                    prev_lsn,
+                    full_backup,
+                    replica,
+                    &ctx,
+                )
+                .await?;
+                // shutdown the encoder to ensure the gzip footer is written
+                encoder
+                    .shutdown()
+                    .await
+                    .map_err(|e| QueryError::Disconnected(ConnectionError::Io(e)))?;
            } else {
                basebackup::send_basebackup_tarball(
                    &mut writer,
@@ -2628,11 +2674,6 @@ impl PageServerHandler {
                    prev_lsn,
                    full_backup,
                    replica,
-                    // NB: using fast compression because it's on the critical path for compute
-                    // startup. For an empty database, we get <100KB with this method. The
-                    // Level::Best compression method gives us <20KB, but maybe we should add
-                    // basebackup caching on compute shutdown first.
-                    gzip.then_some(async_compression::Level::Fastest),
                    &ctx,
                )
                .await?;
@@ -3512,7 +3553,7 @@ impl proto::PageService for GrpcPageServiceHandler {
        if timeline.is_archived() == Some(true) {
            return Err(tonic::Status::failed_precondition("timeline is archived"));
        }
-        let req: page_api::GetBaseBackupRequest = req.into_inner().try_into()?;
+        let req: page_api::GetBaseBackupRequest = req.into_inner().into();

        span_record!(lsn=?req.lsn);

@@ -3538,50 +3579,20 @@ impl proto::PageService for GrpcPageServiceHandler {
        let span = Span::current();
        let (mut simplex_read, mut simplex_write) = tokio::io::simplex(CHUNK_SIZE);
        let jh = tokio::spawn(async move {
-            let gzip_level = match req.compression {
-                page_api::BaseBackupCompression::None => None,
-                // NB: using fast compression because it's on the critical path for compute
-                // startup. For an empty database, we get <100KB with this method. The
-                // Level::Best compression method gives us <20KB, but maybe we should add
-                // basebackup caching on compute shutdown first.
-                page_api::BaseBackupCompression::Gzip => Some(async_compression::Level::Fastest),
-            };
-
-            // Check for a cached basebackup.
-            let cached = timeline
-                .get_cached_basebackup_if_enabled(
-                    req.lsn,
-                    None,
-                    req.full,
-                    req.replica,
-                    gzip_level.is_some(),
-                )
-                .await;
-
-            let result = if let Some(mut cached) = cached {
-                // If we have a cached basebackup, send it.
-                tokio::io::copy(&mut cached, &mut simplex_write)
-                    .await
-                    .map(|_| ())
-                    .map_err(|err| BasebackupError::Client(err, "cached,copy"))
-            } else {
-                basebackup::send_basebackup_tarball(
-                    &mut simplex_write,
-                    &timeline,
-                    req.lsn,
-                    None,
-                    req.full,
-                    req.replica,
-                    gzip_level,
-                    &ctx,
-                )
-                .instrument(span) // propagate request span
-                .await
-            };
-            simplex_write
-                .shutdown()
-                .await
-                .map_err(|err| BasebackupError::Client(err, "simplex_write"))?;
+            let result = basebackup::send_basebackup_tarball(
+                &mut simplex_write,
+                &timeline,
+                req.lsn,
+                None,
+                req.full,
+                req.replica,
+                &ctx,
+            )
+            .instrument(span) // propagate request span
+            .await;
+            simplex_write.shutdown().await.map_err(|err| {
+                BasebackupError::Server(anyhow!("simplex shutdown failed: {err}"))
+            })?;
            result
        });

@@ -3761,36 +3772,6 @@ impl proto::PageService for GrpcPageServiceHandler {
        let resp: page_api::GetSlruSegmentResponse = resp.segment;
        Ok(tonic::Response::new(resp.into()))
    }
-
-    #[instrument(skip_all, fields(lsn))]
-    async fn lease_lsn(
-        &self,
-        req: tonic::Request<proto::LeaseLsnRequest>,
-    ) -> Result<tonic::Response<proto::LeaseLsnResponse>, tonic::Status> {
-        let timeline = self.get_request_timeline(&req).await?;
-        let ctx = self.ctx.with_scope_timeline(&timeline);
-
-        // Validate and convert the request, and decorate the span.
-        let req: page_api::LeaseLsnRequest = req.into_inner().try_into()?;
-
-        span_record!(lsn=%req.lsn);
-
-        // Attempt to acquire a lease. Return FailedPrecondition if the lease could not be granted.
-        let lease_length = timeline.get_lsn_lease_length();
-        let expires = match timeline.renew_lsn_lease(req.lsn, lease_length, &ctx) {
-            Ok(lease) => lease.valid_until,
-            Err(err) => return Err(tonic::Status::failed_precondition(format!("{err}"))),
-        };
-
-        // TODO: is this spammy? Move it compute-side?
-        info!(
-            "acquired lease for {} until {}",
-            req.lsn,
-            chrono::DateTime::<Utc>::from(expires).to_rfc3339()
-        );
-
-        Ok(tonic::Response::new(expires.into()))
-    }
 }

 /// gRPC middleware layer that handles observability concerns:
--- a/pageserver/src/pgdatadir_mapping.rs
+++ b/pageserver/src/pgdatadir_mapping.rs
@@ -25,7 +25,7 @@ use pageserver_api::keyspace::{KeySpaceRandomAccum, SparseKeySpace};
 use pageserver_api::models::RelSizeMigration;
 use pageserver_api::reltag::{BlockNumber, RelTag, SlruKind};
 use pageserver_api::shard::ShardIdentity;
-use postgres_ffi::{BLCKSZ, PgMajorVersion, TimestampTz, TransactionId};
+use postgres_ffi::{BLCKSZ, TimestampTz, TransactionId};
 use postgres_ffi_types::forknum::{FSM_FORKNUM, VISIBILITYMAP_FORKNUM};
 use postgres_ffi_types::{Oid, RepOriginId};
 use serde::{Deserialize, Serialize};
@@ -1081,7 +1081,7 @@ impl Timeline {
        // fetch directory entry
        let buf = self.get(TWOPHASEDIR_KEY, lsn, ctx).await?;

-        if self.pg_version >= PgMajorVersion::PG17 {
+        if self.pg_version >= 17 {
            Ok(TwoPhaseDirectoryV17::des(&buf)?.xids)
        } else {
            Ok(TwoPhaseDirectory::des(&buf)?
@@ -1613,7 +1613,7 @@ impl DatadirModification<'_> {
            .push((DirectoryKind::Db, MetricsUpdate::Set(0)));
        self.put(DBDIR_KEY, Value::Image(buf.into()));

-        let buf = if self.tline.pg_version >= PgMajorVersion::PG17 {
+        let buf = if self.tline.pg_version >= 17 {
            TwoPhaseDirectoryV17::ser(&TwoPhaseDirectoryV17 {
                xids: HashSet::new(),
            })
@@ -1967,7 +1967,7 @@ impl DatadirModification<'_> {
    ) -> Result<(), WalIngestError> {
        // Add it to the directory entry
        let dirbuf = self.get(TWOPHASEDIR_KEY, ctx).await?;
-        let newdirbuf = if self.tline.pg_version >= PgMajorVersion::PG17 {
+        let newdirbuf = if self.tline.pg_version >= 17 {
            let mut dir = TwoPhaseDirectoryV17::des(&dirbuf)?;
            if !dir.xids.insert(xid) {
                Err(WalIngestErrorKind::FileAlreadyExists(xid))?;
@@ -2383,7 +2383,7 @@ impl DatadirModification<'_> {
    ) -> Result<(), WalIngestError> {
        // Remove it from the directory entry
        let buf = self.get(TWOPHASEDIR_KEY, ctx).await?;
-        let newdirbuf = if self.tline.pg_version >= PgMajorVersion::PG17 {
+        let newdirbuf = if self.tline.pg_version >= 17 {
            let mut dir = TwoPhaseDirectoryV17::des(&buf)?;

            if !dir.xids.remove(&xid) {
@@ -3015,7 +3015,7 @@ mod tests {
        // This shard will get the even blocks
        let shard = ShardIdentity::from_params(
            ShardNumber(0),
-            ShardParameters {
+            &ShardParameters {
                count: ShardCount(2),
                stripe_size: ShardStripeSize(1),
            },
--- a/pageserver/src/tenant.rs
+++ b/pageserver/src/tenant.rs
@@ -38,7 +38,6 @@ use pageserver_api::models::{
    WalRedoManagerStatus,
 };
 use pageserver_api::shard::{ShardIdentity, ShardStripeSize, TenantShardId};
-use postgres_ffi::PgMajorVersion;
 use remote_storage::{DownloadError, GenericRemoteStorage, TimeoutOrCancel};
 use remote_timeline_client::index::GcCompactionState;
 use remote_timeline_client::manifest::{
@@ -80,13 +79,13 @@ use self::timeline::uninit::{TimelineCreateGuard, TimelineExclusionError, Uninit
 use self::timeline::{
    EvictionTaskTenantState, GcCutoffs, TimelineDeleteProgress, TimelineResources, WaitLsnError,
 };
-use crate::basebackup_cache::BasebackupCache;
+use crate::basebackup_cache::BasebackupPrepareSender;
 use crate::config::PageServerConf;
 use crate::context;
 use crate::context::RequestContextBuilder;
 use crate::context::{DownloadBehavior, RequestContext};
 use crate::deletion_queue::{DeletionQueueClient, DeletionQueueError};
-use crate::feature_resolver::{FeatureResolver, TenantFeatureResolver};
+use crate::feature_resolver::FeatureResolver;
 use crate::l0_flush::L0FlushGlobalState;
 use crate::metrics::{
    BROKEN_TENANTS_SET, CIRCUIT_BREAKERS_BROKEN, CIRCUIT_BREAKERS_UNBROKEN, CONCURRENT_INITDBS,
@@ -162,7 +161,7 @@ pub struct TenantSharedResources {
    pub remote_storage: GenericRemoteStorage,
    pub deletion_queue_client: DeletionQueueClient,
    pub l0_flush_global_state: L0FlushGlobalState,
-    pub basebackup_cache: Arc<BasebackupCache>,
+    pub basebackup_prepare_sender: BasebackupPrepareSender,
    pub feature_resolver: FeatureResolver,
 }

@@ -331,7 +330,7 @@ pub struct TenantShard {
    deletion_queue_client: DeletionQueueClient,

    /// A channel to send async requests to prepare a basebackup for the basebackup cache.
-    basebackup_cache: Arc<BasebackupCache>,
+    basebackup_prepare_sender: BasebackupPrepareSender,

    /// Cached logical sizes updated updated on each [`TenantShard::gather_size_inputs`].
    cached_logical_sizes: tokio::sync::Mutex<HashMap<(TimelineId, Lsn), u64>>,
@@ -386,7 +385,7 @@ pub struct TenantShard {

    l0_flush_global_state: L0FlushGlobalState,

-    pub(crate) feature_resolver: TenantFeatureResolver,
+    pub(crate) feature_resolver: FeatureResolver,
 }
 impl std::fmt::Debug for TenantShard {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
@@ -498,7 +497,7 @@ impl WalRedoManager {
        lsn: Lsn,
        base_img: Option<(Lsn, bytes::Bytes)>,
        records: Vec<(Lsn, wal_decoder::models::record::NeonWalRecord)>,
-        pg_version: PgMajorVersion,
+        pg_version: u32,
        redo_attempt_type: RedoAttemptType,
    ) -> Result<bytes::Bytes, walredo::Error> {
        match self {
@@ -934,7 +933,7 @@ pub(crate) enum CreateTimelineParams {
 pub(crate) struct CreateTimelineParamsBootstrap {
    pub(crate) new_timeline_id: TimelineId,
    pub(crate) existing_initdb_timeline_id: Option<TimelineId>,
-    pub(crate) pg_version: PgMajorVersion,
+    pub(crate) pg_version: u32,
 }

 /// NB: See comment on [`CreateTimelineIdempotency::Branch`] for why there's no `pg_version` here.
@@ -972,7 +971,7 @@ pub(crate) enum CreateTimelineIdempotency {
    /// NB: special treatment, see comment in [`Self`].
    FailWithConflict,
    Bootstrap {
-        pg_version: PgMajorVersion,
+        pg_version: u32,
    },
    /// NB: branches always have the same `pg_version` as their ancestor.
    /// While [`pageserver_api::models::TimelineCreateRequestMode::Branch::pg_version`]
@@ -1363,7 +1362,7 @@ impl TenantShard {
            remote_storage,
            deletion_queue_client,
            l0_flush_global_state,
-            basebackup_cache,
+            basebackup_prepare_sender,
            feature_resolver,
        } = resources;

@@ -1380,7 +1379,7 @@ impl TenantShard {
            remote_storage.clone(),
            deletion_queue_client,
            l0_flush_global_state,
-            basebackup_cache,
+            basebackup_prepare_sender,
            feature_resolver,
        ));

@@ -2542,7 +2541,7 @@ impl TenantShard {
        self: &Arc<Self>,
        new_timeline_id: TimelineId,
        initdb_lsn: Lsn,
-        pg_version: PgMajorVersion,
+        pg_version: u32,
        ctx: &RequestContext,
    ) -> anyhow::Result<(UninitializedTimeline, RequestContext)> {
        anyhow::ensure!(
@@ -2594,7 +2593,7 @@ impl TenantShard {
        self: &Arc<Self>,
        new_timeline_id: TimelineId,
        initdb_lsn: Lsn,
-        pg_version: PgMajorVersion,
+        pg_version: u32,
        ctx: &RequestContext,
    ) -> anyhow::Result<Arc<Timeline>> {
        let (uninit_tl, ctx) = self
@@ -2633,7 +2632,7 @@ impl TenantShard {
        self: &Arc<Self>,
        new_timeline_id: TimelineId,
        initdb_lsn: Lsn,
-        pg_version: PgMajorVersion,
+        pg_version: u32,
        ctx: &RequestContext,
        in_memory_layer_desc: Vec<timeline::InMemoryLayerTestDesc>,
        delta_layer_desc: Vec<timeline::DeltaLayerTestDesc>,
@@ -2899,7 +2898,7 @@ impl TenantShard {
                    Lsn(0),
                    initdb_lsn,
                    initdb_lsn,
-                    PgMajorVersion::PG15,
+                    15,
                );
                this.prepare_new_timeline(
                    new_timeline_id,
@@ -3263,7 +3262,7 @@ impl TenantShard {
                };
                let gc_compaction_strategy = self
                    .feature_resolver
-                    .evaluate_multivariate("gc-comapction-strategy")
+                    .evaluate_multivariate("gc-comapction-strategy", self.tenant_shard_id.tenant_id)
                    .ok();
                let span = if let Some(gc_compaction_strategy) = gc_compaction_strategy {
                    info_span!("gc_compact_timeline", timeline_id = %timeline.timeline_id, strategy = %gc_compaction_strategy)
@@ -3408,9 +3407,6 @@ impl TenantShard {
        if let Some(ref walredo_mgr) = self.walredo_mgr {
            walredo_mgr.maybe_quiesce(WALREDO_IDLE_TIMEOUT);
        }
-
-        // Update the feature resolver with the latest tenant-spcific data.
-        self.feature_resolver.update_cached_tenant_properties(self);
    }

    pub fn timeline_has_no_attached_children(&self, timeline_id: TimelineId) -> bool {
@@ -3875,10 +3871,6 @@ impl TenantShard {
        &self.tenant_shard_id
    }

-    pub(crate) fn get_shard_identity(&self) -> ShardIdentity {
-        self.shard_identity
-    }
-
    pub(crate) fn get_shard_stripe_size(&self) -> ShardStripeSize {
        self.shard_identity.stripe_size
    }
@@ -4387,7 +4379,7 @@ impl TenantShard {
        remote_storage: GenericRemoteStorage,
        deletion_queue_client: DeletionQueueClient,
        l0_flush_global_state: L0FlushGlobalState,
-        basebackup_cache: Arc<BasebackupCache>,
+        basebackup_prepare_sender: BasebackupPrepareSender,
        feature_resolver: FeatureResolver,
    ) -> TenantShard {
        assert!(!attached_conf.location.generation.is_none());
@@ -4492,11 +4484,8 @@ impl TenantShard {
            ongoing_timeline_detach: std::sync::Mutex::default(),
            gc_block: Default::default(),
            l0_flush_global_state,
-            basebackup_cache,
-            feature_resolver: TenantFeatureResolver::new(
-                feature_resolver,
-                tenant_shard_id.tenant_id,
-            ),
+            basebackup_prepare_sender,
+            feature_resolver,
        }
    }

@@ -4535,10 +4524,6 @@ impl TenantShard {
        Ok(toml_edit::de::from_str::<LocationConf>(&config)?)
    }

-    /// Stores a tenant location config to disk.
-    ///
-    /// NB: make sure to call `ShardIdentity::assert_equal` before persisting a new config, to avoid
-    /// changes to shard parameters that may result in data corruption.
    #[tracing::instrument(skip_all, fields(tenant_id=%tenant_shard_id.tenant_id, shard_id=%tenant_shard_id.shard_slug()))]
    pub(super) async fn persist_tenant_config(
        conf: &'static PageServerConf,
@@ -5105,7 +5090,7 @@ impl TenantShard {
    pub(crate) async fn bootstrap_timeline_test(
        self: &Arc<Self>,
        timeline_id: TimelineId,
-        pg_version: PgMajorVersion,
+        pg_version: u32,
        load_existing_initdb: Option<TimelineId>,
        ctx: &RequestContext,
    ) -> anyhow::Result<Arc<Timeline>> {
@@ -5247,7 +5232,7 @@ impl TenantShard {
    async fn bootstrap_timeline(
        self: &Arc<Self>,
        timeline_id: TimelineId,
-        pg_version: PgMajorVersion,
+        pg_version: u32,
        load_existing_initdb: Option<TimelineId>,
        ctx: &RequestContext,
    ) -> Result<CreateTimelineResult, CreateTimelineError> {
@@ -5428,7 +5413,7 @@ impl TenantShard {
            pagestream_throttle_metrics: self.pagestream_throttle_metrics.clone(),
            l0_compaction_trigger: self.l0_compaction_trigger.clone(),
            l0_flush_global_state: self.l0_flush_global_state.clone(),
-            basebackup_cache: self.basebackup_cache.clone(),
+            basebackup_prepare_sender: self.basebackup_prepare_sender.clone(),
            feature_resolver: self.feature_resolver.clone(),
        }
    }
@@ -5785,7 +5770,7 @@ impl TenantShard {
 async fn run_initdb(
    conf: &'static PageServerConf,
    initdb_target_dir: &Utf8Path,
-    pg_version: PgMajorVersion,
+    pg_version: u32,
    cancel: &CancellationToken,
 ) -> Result<(), InitdbError> {
    let initdb_bin_path = conf
@@ -6014,7 +5999,7 @@ pub(crate) mod harness {
        ) -> anyhow::Result<Arc<TenantShard>> {
            let walredo_mgr = Arc::new(WalRedoManager::from(TestRedoManager));

-            let (basebackup_cache, _) = BasebackupCache::new(Utf8PathBuf::new(), None);
+            let (basebackup_requst_sender, _) = tokio::sync::mpsc::unbounded_channel();

            let tenant = Arc::new(TenantShard::new(
                TenantState::Attaching,
@@ -6022,7 +6007,7 @@ pub(crate) mod harness {
                AttachedTenantConf::try_from(LocationConf::attached_single(
                    self.tenant_conf.clone(),
                    self.generation,
-                    ShardParameters::default(),
+                    &ShardParameters::default(),
                ))
                .unwrap(),
                self.shard_identity,
@@ -6032,7 +6017,7 @@ pub(crate) mod harness {
                self.deletion_queue.new_client(),
                // TODO: ideally we should run all unit tests with both configs
                L0FlushGlobalState::new(L0FlushConfig::default()),
-                basebackup_cache,
+                basebackup_requst_sender,
                FeatureResolver::new_disabled(),
            ));

@@ -6066,7 +6051,7 @@ pub(crate) mod harness {
            lsn: Lsn,
            base_img: Option<(Lsn, Bytes)>,
            records: Vec<(Lsn, NeonWalRecord)>,
-            _pg_version: PgMajorVersion,
+            _pg_version: u32,
            _redo_attempt_type: RedoAttemptType,
        ) -> Result<Bytes, walredo::Error> {
            let records_neon = records.iter().all(|r| apply_neon::can_apply_in_neon(&r.1));
@@ -6238,7 +6223,7 @@ mod tests {
    async fn randomize_timeline(
        tenant: &Arc<TenantShard>,
        new_timeline_id: TimelineId,
-        pg_version: PgMajorVersion,
+        pg_version: u32,
        spec: TestTimelineSpecification,
        random: &mut rand::rngs::StdRng,
        ctx: &RequestContext,
@@ -11443,11 +11428,11 @@ mod tests {
        if left != right {
            eprintln!("---LEFT---");
            for left in left.iter() {
-                eprintln!("{left}");
+                eprintln!("{}", left);
            }
            eprintln!("---RIGHT---");
            for right in right.iter() {
-                eprintln!("{right}");
+                eprintln!("{}", right);
            }
            assert_eq!(left, right);
        }
--- a/pageserver/src/tenant/config.rs
+++ b/pageserver/src/tenant/config.rs
@@ -12,7 +12,6 @@
 use pageserver_api::models;
 use pageserver_api::shard::{ShardCount, ShardIdentity, ShardNumber, ShardStripeSize};
 use serde::{Deserialize, Serialize};
-use utils::critical;
 use utils::generation::Generation;

 #[derive(Debug, Copy, Clone, Serialize, Deserialize, PartialEq, Eq)]
@@ -137,7 +136,7 @@ impl LocationConf {
    pub(crate) fn attached_single(
        tenant_conf: pageserver_api::models::TenantConfig,
        generation: Generation,
-        shard_params: models::ShardParameters,
+        shard_params: &models::ShardParameters,
    ) -> Self {
        Self {
            mode: LocationMode::Attached(AttachedLocationConfig {
@@ -172,16 +171,6 @@ impl LocationConf {
            }
        }

-        // This should never happen.
-        // TODO: turn this into a proper assertion.
-        if stripe_size != self.shard.stripe_size {
-            critical!(
-                "stripe size mismatch: {} != {}",
-                self.shard.stripe_size,
-                stripe_size,
-            );
-        }
-
        self.shard.stripe_size = stripe_size;
    }

--- a/pageserver/src/tenant/metadata.rs
+++ b/pageserver/src/tenant/metadata.rs
@@ -18,7 +18,6 @@
 //! [`IndexPart`]: super::remote_timeline_client::index::IndexPart

 use anyhow::ensure;
-use postgres_ffi::PgMajorVersion;
 use serde::{Deserialize, Serialize};
 use utils::bin_ser::{BeSer, SerializeError};
 use utils::id::TimelineId;
@@ -137,7 +136,7 @@ struct TimelineMetadataBodyV2 {
    latest_gc_cutoff_lsn: Lsn,

    initdb_lsn: Lsn,
-    pg_version: PgMajorVersion,
+    pg_version: u32,
 }

 #[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
@@ -168,7 +167,7 @@ impl TimelineMetadata {
        ancestor_lsn: Lsn,
        latest_gc_cutoff_lsn: Lsn,
        initdb_lsn: Lsn,
-        pg_version: PgMajorVersion,
+        pg_version: u32,
    ) -> Self {
        Self {
            hdr: TimelineMetadataHeader {
@@ -216,7 +215,7 @@ impl TimelineMetadata {
            ancestor_lsn: body.ancestor_lsn,
            latest_gc_cutoff_lsn: body.latest_gc_cutoff_lsn,
            initdb_lsn: body.initdb_lsn,
-            pg_version: PgMajorVersion::PG14, // All timelines created before this version had pg_version 14
+            pg_version: 14, // All timelines created before this version had pg_version 14
        };

        hdr.format_version = METADATA_FORMAT_VERSION;
@@ -318,7 +317,7 @@ impl TimelineMetadata {
        self.body.initdb_lsn
    }

-    pub fn pg_version(&self) -> PgMajorVersion {
+    pub fn pg_version(&self) -> u32 {
        self.body.pg_version
    }

@@ -332,7 +331,7 @@ impl TimelineMetadata {
            Lsn::from_hex("00000000").unwrap(),
            Lsn::from_hex("00000000").unwrap(),
            Lsn::from_hex("00000000").unwrap(),
-            PgMajorVersion::PG14,
+            0,
        );
        let bytes = instance.to_bytes().unwrap();
        Self::from_bytes(&bytes).unwrap()
@@ -546,7 +545,7 @@ mod tests {
            Lsn(0),
            Lsn(0),
            Lsn(0),
-            PgMajorVersion::PG14, // All timelines created before this version had pg_version 14
+            14, // All timelines created before this version had pg_version 14
        );

        assert_eq!(
@@ -566,7 +565,7 @@ mod tests {
            Lsn(0),
            // Updating this version to 17 will cause the test to fail at the
            // next assert_eq!().
-            PgMajorVersion::PG16,
+            16,
        );
        let expected_bytes = vec![
            /* TimelineMetadataHeader */
--- a/pageserver/src/tenant/mgr.rs
+++ b/pageserver/src/tenant/mgr.rs
@@ -880,9 +880,6 @@ impl TenantManager {
        // phase of writing config and/or waiting for flush, before returning.
        match fast_path_taken {
            Some(FastPathModified::Attached(tenant)) => {
-                tenant
-                    .shard_identity
-                    .assert_equal(new_location_config.shard);
                TenantShard::persist_tenant_config(
                    self.conf,
                    &tenant_shard_id,
@@ -917,10 +914,7 @@ impl TenantManager {

                return Ok(Some(tenant));
            }
-            Some(FastPathModified::Secondary(secondary_tenant)) => {
-                secondary_tenant
-                    .shard_identity
-                    .assert_equal(new_location_config.shard);
+            Some(FastPathModified::Secondary(_secondary_tenant)) => {
                TenantShard::persist_tenant_config(
                    self.conf,
                    &tenant_shard_id,
@@ -954,10 +948,6 @@ impl TenantManager {

        match slot_guard.get_old_value() {
            Some(TenantSlot::Attached(tenant)) => {
-                tenant
-                    .shard_identity
-                    .assert_equal(new_location_config.shard);
-
                // The case where we keep a Tenant alive was covered above in the special case
                // for Attached->Attached transitions in the same generation.  By this point,
                // if we see an attached tenant we know it will be discarded and should be
@@ -991,13 +981,9 @@ impl TenantManager {
                // rather than assuming it to be empty.
                spawn_mode = SpawnMode::Eager;
            }
-            Some(TenantSlot::Secondary(secondary_tenant)) => {
-                secondary_tenant
-                    .shard_identity
-                    .assert_equal(new_location_config.shard);
-
+            Some(TenantSlot::Secondary(state)) => {
                info!("Shutting down secondary tenant");
-                secondary_tenant.shutdown().await;
+                state.shutdown().await;
            }
            Some(TenantSlot::InProgress(_)) => {
                // This should never happen: acquire_slot should error out
@@ -2214,7 +2200,7 @@ impl TenantManager {
        selector: ShardSelector,
    ) -> ShardResolveResult {
        let tenants = self.tenants.read().unwrap();
-        let mut want_shard: Option<ShardIndex> = None;
+        let mut want_shard = None;
        let mut any_in_progress = None;

        match &*tenants {
@@ -2239,23 +2225,14 @@ impl TenantManager {
                            return ShardResolveResult::Found(tenant.clone());
                        }
                        ShardSelector::Page(key) => {
-                            // Each time we find an attached slot with a different shard count,
-                            // recompute the expected shard number: during shard splits we might
-                            // have multiple shards with the old shard count.
-                            if want_shard.is_none()
-                                || want_shard.unwrap().shard_count != tenant.shard_identity.count
-                            {
-                                want_shard = Some(ShardIndex {
-                                    shard_number: tenant.shard_identity.get_shard_number(&key),
-                                    shard_count: tenant.shard_identity.count,
-                                });
+                            // First slot we see for this tenant, calculate the expected shard number
+                            // for the key: we will use this for checking if this and subsequent
+                            // slots contain the key, rather than recalculating the hash each time.
+                            if want_shard.is_none() {
+                                want_shard = Some(tenant.shard_identity.get_shard_number(&key));
                            }

-                            if Some(ShardIndex {
-                                shard_number: tenant.shard_identity.number,
-                                shard_count: tenant.shard_identity.count,
-                            }) == want_shard
-                            {
+                            if Some(tenant.shard_identity.number) == want_shard {
                                return ShardResolveResult::Found(tenant.clone());
                            }
                        }
@@ -2914,18 +2891,14 @@ mod tests {
    use std::collections::BTreeMap;
    use std::sync::Arc;

-    use camino::Utf8PathBuf;
    use storage_broker::BrokerClientChannel;
    use tracing::Instrument;

    use super::super::harness::TenantHarness;
    use super::TenantsMap;
-    use crate::{
-        basebackup_cache::BasebackupCache,
-        tenant::{
-            TenantSharedResources,
-            mgr::{BackgroundPurges, TenantManager, TenantSlot},
-        },
+    use crate::tenant::{
+        TenantSharedResources,
+        mgr::{BackgroundPurges, TenantManager, TenantSlot},
    };

    #[tokio::test(start_paused = true)]
@@ -2951,7 +2924,9 @@ mod tests {
        // Invoke remove_tenant_from_memory with a cleanup hook that blocks until we manually
        // permit it to proceed: that will stick the tenant in InProgress

-        let (basebackup_cache, _) = BasebackupCache::new(Utf8PathBuf::new(), None);
+        let (basebackup_prepare_sender, _) = tokio::sync::mpsc::unbounded_channel::<
+            crate::basebackup_cache::BasebackupPrepareRequest,
+        >();

        let tenant_manager = TenantManager {
            tenants: std::sync::RwLock::new(TenantsMap::Open(tenants)),
@@ -2965,7 +2940,7 @@ mod tests {
                l0_flush_global_state: crate::l0_flush::L0FlushGlobalState::new(
                    h.conf.l0_flush.clone(),
                ),
-                basebackup_cache,
+                basebackup_prepare_sender,
                feature_resolver: crate::feature_resolver::FeatureResolver::new_disabled(),
            },
            cancel: tokio_util::sync::CancellationToken::new(),
--- a/pageserver/src/tenant/remote_timeline_client/index.rs
+++ b/pageserver/src/tenant/remote_timeline_client/index.rs
@@ -427,8 +427,8 @@ impl GcBlocking {

 #[cfg(test)]
 mod tests {
-    use postgres_ffi::PgMajorVersion;
    use std::str::FromStr;
+
    use utils::id::TimelineId;

    use super::*;
@@ -831,7 +831,7 @@ mod tests {
                Lsn::INVALID,
                Lsn::from_str("0/1696070").unwrap(),
                Lsn::from_str("0/1696070").unwrap(),
-                PgMajorVersion::PG14,
+                14,
            ).with_recalculated_checksum().unwrap(),
            deleted_at: Some(parse_naive_datetime("2023-07-31T09:00:00.123000000")),
            archived_at: None,
@@ -893,7 +893,7 @@ mod tests {
                Lsn::INVALID,
                Lsn::from_str("0/1696070").unwrap(),
                Lsn::from_str("0/1696070").unwrap(),
-                PgMajorVersion::PG14,
+                14,
            ).with_recalculated_checksum().unwrap(),
            deleted_at: Some(parse_naive_datetime("2023-07-31T09:00:00.123000000")),
            archived_at: Some(parse_naive_datetime("2023-04-29T09:00:00.123000000")),
@@ -957,7 +957,7 @@ mod tests {
                Lsn::INVALID,
                Lsn::from_str("0/1696070").unwrap(),
                Lsn::from_str("0/1696070").unwrap(),
-                PgMajorVersion::PG14,
+                14,
            ).with_recalculated_checksum().unwrap(),
            deleted_at: None,
            lineage: Default::default(),
@@ -1033,7 +1033,7 @@ mod tests {
                Lsn::INVALID,
                Lsn::from_str("0/1696070").unwrap(),
                Lsn::from_str("0/1696070").unwrap(),
-                PgMajorVersion::PG14,
+                14,
            ).with_recalculated_checksum().unwrap(),
            deleted_at: None,
            lineage: Default::default(),
@@ -1114,7 +1114,7 @@ mod tests {
                Lsn::INVALID,
                Lsn::from_str("0/1696070").unwrap(),
                Lsn::from_str("0/1696070").unwrap(),
-                PgMajorVersion::PG14,
+                14,
            ).with_recalculated_checksum().unwrap(),
            deleted_at: None,
            lineage: Default::default(),
@@ -1199,7 +1199,7 @@ mod tests {
                Lsn::INVALID,
                Lsn::from_str("0/1696070").unwrap(),
                Lsn::from_str("0/1696070").unwrap(),
-                PgMajorVersion::PG14,
+                14,
            ).with_recalculated_checksum().unwrap(),
            deleted_at: None,
            lineage: Default::default(),
@@ -1287,7 +1287,7 @@ mod tests {
                Lsn::INVALID,
                Lsn::from_str("0/1696070").unwrap(),
                Lsn::from_str("0/1696070").unwrap(),
-                PgMajorVersion::PG14,
+                14,
            ).with_recalculated_checksum().unwrap(),
            deleted_at: None,
            lineage: Default::default(),
--- a/pageserver/src/tenant/secondary.rs
+++ b/pageserver/src/tenant/secondary.rs
@@ -101,7 +101,7 @@ pub(crate) struct SecondaryTenant {
    // Secondary mode does not need the full shard identity or the pageserver_api::models::TenantConfig.  However,
    // storing these enables us to report our full LocationConf, enabling convenient reconciliation
    // by the control plane (see [`Self::get_location_conf`])
-    pub(crate) shard_identity: ShardIdentity,
+    shard_identity: ShardIdentity,
    tenant_conf: std::sync::Mutex<pageserver_api::models::TenantConfig>,

    // Internal state used by the Downloader.
--- a/Show More
+++ b/Show More
Author	SHA1	Message	Date
Mikhail Kot	8004abbb9c	fix	2025-06-24 12:50:59 +01:00
Mikhail Kot	d817dbab37	style	2025-06-24 12:41:30 +01:00
Mikhail Kot	a85cd674d2	NO_CLOSE_RANGE macro for old glibc/kernel builds	2025-06-24 12:41:30 +01:00