Store read_only flag in pageserver timelines

## Problem
Names are not consistent between safekeeper migration RFC and the actual
implementation.

It's not used anywhere in production yet, so it's safe to rename. We
don't need to worry about backward compatibility.

- Follow up on https://github.com/neondatabase/neon/pull/12432

## Summary of changes
- rename term -> last_log_term in TimelineMembershipSwitchResponse 
- add missing fields to TimelineMembershipSwitchResponse in python

.github/actionlint.yml

@@ -7,6 +7,7 @@ self-hosted-runner:
     - small-metal
     - small-arm64
     - unit-perf
+    - unit-perf-aws-arm
     - us-east-2
 config-variables:
   - AWS_ECR_REGION

.github/workflows/build-macos.yml

@@ -32,162 +32,14 @@ permissions:
   contents: read
 
 jobs:
-  build-pgxn:
-    if: |
-      inputs.pg_versions != '[]' || inputs.rebuild_everything ||
-      contains(github.event.pull_request.labels.*.name, 'run-extra-build-macos')  ||
-      contains(github.event.pull_request.labels.*.name, 'run-extra-build-*') ||
-      github.ref_name == 'main'
-    timeout-minutes: 30
-    runs-on: macos-15
-    strategy:
-      matrix:
-        postgres-version: ${{ inputs.rebuild_everything && fromJSON('["v14", "v15", "v16", "v17"]') || fromJSON(inputs.pg_versions) }}
-    env:
-      # Use release build only, to have less debug info around
-      # Hence keeping target/ (and general cache size) smaller
-      BUILD_TYPE: release
-    steps:
-      - name: Harden the runner (Audit all outbound calls)
-        uses: step-security/harden-runner@4d991eb9b905ef189e4c376166672c3f2f230481 # v2.11.0
-        with:
-          egress-policy: audit
-
-      - name: Checkout main repo
-        uses: actions/checkout@11bd71901bbe5b1630ceea73d27597364c9af683 # v4.2.2
-
-      - name: Set pg ${{ matrix.postgres-version }} for caching
-        id: pg_rev
-        run: echo pg_rev=$(git rev-parse HEAD:vendor/postgres-${{ matrix.postgres-version }}) | tee -a "${GITHUB_OUTPUT}"
-
-      - name: Cache postgres ${{ matrix.postgres-version }} build
-        id: cache_pg
-        uses: actions/cache@5a3ec84eff668545956fd18022155c47e93e2684 # v4.2.3
-        with:
-          path: pg_install/${{ matrix.postgres-version }}
-          key: v1-${{ runner.os }}-${{ runner.arch }}-${{ env.BUILD_TYPE }}-pg-${{ matrix.postgres-version }}-${{ steps.pg_rev.outputs.pg_rev }}-${{ hashFiles('Makefile') }}
-
-      - name: Checkout submodule vendor/postgres-${{ matrix.postgres-version }}
-        if: steps.cache_pg.outputs.cache-hit != 'true'
-        run: |
-          git submodule init vendor/postgres-${{ matrix.postgres-version }}
-          git submodule update --depth 1 --recursive
-
-      - name: Install build dependencies
-        if: steps.cache_pg.outputs.cache-hit != 'true'
-        run: |
-          brew install flex bison openssl protobuf icu4c
-
-      - name: Set extra env for macOS
-        if: steps.cache_pg.outputs.cache-hit != 'true'
-        run: |
-          echo 'LDFLAGS=-L/usr/local/opt/openssl@3/lib' >> $GITHUB_ENV
-          echo 'CPPFLAGS=-I/usr/local/opt/openssl@3/include' >> $GITHUB_ENV
-
-      - name: Build Postgres ${{ matrix.postgres-version }}
-        if: steps.cache_pg.outputs.cache-hit != 'true'
-        run: |
-          make postgres-${{ matrix.postgres-version }} -j$(sysctl -n hw.ncpu)
-
-      - name: Build Neon Pg Ext ${{ matrix.postgres-version }}
-        if: steps.cache_pg.outputs.cache-hit != 'true'
-        run: |
-          make "neon-pg-ext-${{ 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:
-          name: pg_install--${{ matrix.postgres-version }}
-          path: pg_install/${{ matrix.postgres-version }}
-          # The artifact is supposed to be used by the next job in the same workflow,
-          # so there’s no need to store it for too long.
-          retention-days: 1
-
-  build-walproposer-lib:
-    if: |
-      contains(inputs.pg_versions, 'v17') || inputs.rebuild_everything ||
-      contains(github.event.pull_request.labels.*.name, 'run-extra-build-macos')  ||
-      contains(github.event.pull_request.labels.*.name, 'run-extra-build-*') ||
-      github.ref_name == 'main'
-    timeout-minutes: 30
-    runs-on: macos-15
-    needs: [build-pgxn]
-    env:
-      # Use release build only, to have less debug info around
-      # Hence keeping target/ (and general cache size) smaller
-      BUILD_TYPE: release
-    steps:
-      - name: Harden the runner (Audit all outbound calls)
-        uses: step-security/harden-runner@4d991eb9b905ef189e4c376166672c3f2f230481 # v2.11.0
-        with:
-          egress-policy: audit
-
-      - name: Checkout main repo
-        uses: actions/checkout@11bd71901bbe5b1630ceea73d27597364c9af683 # v4.2.2
-
-      - name: Set pg v17 for caching
-        id: pg_rev
-        run: echo pg_rev=$(git rev-parse HEAD:vendor/postgres-v17) | tee -a "${GITHUB_OUTPUT}"
-
-      - name: Download "pg_install/v17" artifact
-        uses: actions/download-artifact@d3f86a106a0bac45b974a628896c90dbdf5c8093 # v4.3.0
-        with:
-          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
-        with:
-          path: build/walproposer-lib
-          key: v1-${{ runner.os }}-${{ runner.arch }}-${{ env.BUILD_TYPE }}-walproposer_lib-v17-${{ steps.pg_rev.outputs.pg_rev }}-${{ hashFiles('Makefile') }}
-
-      - name: Checkout submodule vendor/postgres-v17
-        if: steps.cache_walproposer_lib.outputs.cache-hit != 'true'
-        run: |
-          git submodule init vendor/postgres-v17
-          git submodule update --depth 1 --recursive
-
-      - name: Install build dependencies
-        if: steps.cache_walproposer_lib.outputs.cache-hit != 'true'
-        run: |
-          brew install flex bison openssl protobuf icu4c
-
-      - name: Set extra env for macOS
-        if: steps.cache_walproposer_lib.outputs.cache-hit != 'true'
-        run: |
-          echo 'LDFLAGS=-L/usr/local/opt/openssl@3/lib' >> $GITHUB_ENV
-          echo 'CPPFLAGS=-I/usr/local/opt/openssl@3/include' >> $GITHUB_ENV
-
-      - 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
-
-      - name: Upload "build/walproposer-lib" artifact
-        uses: actions/upload-artifact@ea165f8d65b6e75b540449e92b4886f43607fa02 # v4.6.2
-        with:
-          name: build--walproposer-lib
-          path: build/walproposer-lib
-          # The artifact is supposed to be used by the next job in the same workflow,
-          # so there’s no need to store it for too long.
-          retention-days: 1
-
-  cargo-build:
+  make-all:
     if: |
       inputs.pg_versions != '[]' || inputs.rebuild_rust_code || inputs.rebuild_everything ||
       contains(github.event.pull_request.labels.*.name, 'run-extra-build-macos') ||
       contains(github.event.pull_request.labels.*.name, 'run-extra-build-*') ||
       github.ref_name == 'main'
-    timeout-minutes: 30
+    timeout-minutes: 60
     runs-on: macos-15
-    needs: [build-pgxn, build-walproposer-lib]
     env:
       # Use release build only, to have less debug info around
       # Hence keeping target/ (and general cache size) smaller
@@ -203,41 +55,53 @@ jobs:
         with:
           submodules: true
 
-      - name: Download "pg_install/v14" artifact
-        uses: actions/download-artifact@d3f86a106a0bac45b974a628896c90dbdf5c8093 # v4.3.0
-        with:
-          name: pg_install--v14
-          path: pg_install/v14
-
-      - name: Download "pg_install/v15" artifact
-        uses: actions/download-artifact@d3f86a106a0bac45b974a628896c90dbdf5c8093 # v4.3.0
-        with:
-          name: pg_install--v15
-          path: pg_install/v15
-
-      - name: Download "pg_install/v16" artifact
-        uses: actions/download-artifact@d3f86a106a0bac45b974a628896c90dbdf5c8093 # v4.3.0
-        with:
-          name: pg_install--v16
-          path: pg_install/v16
-
-      - name: Download "pg_install/v17" artifact
-        uses: actions/download-artifact@d3f86a106a0bac45b974a628896c90dbdf5c8093 # v4.3.0
-        with:
-          name: pg_install--v17
-          path: pg_install/v17
-
-      - name: Download "build/walproposer-lib" artifact
-        uses: actions/download-artifact@d3f86a106a0bac45b974a628896c90dbdf5c8093 # v4.3.0
-        with:
-          name: build--walproposer-lib
-          path: build/walproposer-lib
-
-      # `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
+      - name: Install build dependencies
         run: |
-          chmod +x pg_install/v*/bin/*
+          brew install flex bison openssl protobuf icu4c
+
+      - name: Set extra env for macOS
+        run: |
+          echo 'LDFLAGS=-L/usr/local/opt/openssl@3/lib' >> $GITHUB_ENV
+          echo 'CPPFLAGS=-I/usr/local/opt/openssl@3/include' >> $GITHUB_ENV
+
+      - name: Restore "pg_install/" cache
+        id: cache_pg
+        uses: actions/cache@5a3ec84eff668545956fd18022155c47e93e2684 # v4.2.3
+        with:
+          path: pg_install
+          key: v1-${{ runner.os }}-${{ runner.arch }}-${{ env.BUILD_TYPE }}-pg-install-v14-${{ hashFiles('Makefile', 'postgres.mk', 'vendor/revisions.json') }}
+
+      - name: Checkout vendor/postgres submodules
+        if: steps.cache_pg.outputs.cache-hit != 'true'
+        run: |
+          git submodule init
+          git submodule update --depth 1 --recursive
+
+      - name: Build Postgres
+        if: steps.cache_pg.outputs.cache-hit != 'true'
+        run: |
+          make postgres -j$(sysctl -n hw.ncpu)
+
+      # This isn't strictly necessary, but it makes the cached and non-cached builds more similar,
+      # When pg_install is restored from cache, there is no 'build/' directory. By removing it
+      # in a non-cached build too, we enforce that the rest of the steps don't depend on it,
+      # so that we notice any build caching bugs earlier.
+      - name: Remove build artifacts
+        if: steps.cache_pg.outputs.cache-hit != 'true'
+        run: |
+          rm -rf build
+
+      # Explicitly update the rust toolchain before running 'make'. The parallel make build can
+      # invoke 'cargo build' more than once in parallel, for different crates.  That's OK, 'cargo'
+      # does its own locking to prevent concurrent builds from stepping on each other's
+      # toes. However, it will first try to update the toolchain, and that step is not locked the
+      # same way. To avoid two toolchain updates running in parallel and stepping on each other's
+      # toes, ensure that the toolchain is up-to-date beforehand.
+      - name: Update rust toolchain
+        run: |
+          rustup --version &&
+          rustup update &&
+          rustup show
 
       - name: Cache cargo deps
         uses: actions/cache@5a3ec84eff668545956fd18022155c47e93e2684 # v4.2.3
@@ -249,17 +113,12 @@ jobs:
             target
           key: v1-${{ runner.os }}-${{ runner.arch }}-cargo-${{ hashFiles('./Cargo.lock') }}-${{ hashFiles('./rust-toolchain.toml') }}-rust
 
-      - name: Install build dependencies
-        run: |
-          brew install flex bison openssl protobuf icu4c
-
-      - name: Set extra env for macOS
-        run: |
-          echo 'LDFLAGS=-L/usr/local/opt/openssl@3/lib' >> $GITHUB_ENV
-          echo 'CPPFLAGS=-I/usr/local/opt/openssl@3/include' >> $GITHUB_ENV
-
-      - name: Run cargo build
-        run: cargo build --all --release -j$(sysctl -n hw.ncpu)
+      # Build the neon-specific postgres extensions, and all the Rust bits.
+      #
+      # Pass PG_INSTALL_CACHED=1 because PostgreSQL was already built and cached
+      # separately.
+      - name: Build all
+        run: PG_INSTALL_CACHED=1 BUILD_TYPE=release make -j$(sysctl -n hw.ncpu) all
 
       - name: Check that no warnings are produced
         run: ./run_clippy.sh

.github/workflows/build_and_test.yml

@@ -306,14 +306,14 @@ jobs:
       statuses: write
       contents: write
       pull-requests: write
-    runs-on: [ self-hosted, unit-perf ]
+    runs-on: [ self-hosted, unit-perf-aws-arm ]
     container:
       image: ${{ needs.build-build-tools-image.outputs.image }}-bookworm
       credentials:
         username: ${{ github.actor }}
         password: ${{ secrets.GITHUB_TOKEN }}
       # for changed limits, see comments on `options:` earlier in this file
-      options: --init --shm-size=512mb --ulimit memlock=67108864:67108864
+      options: --init --shm-size=512mb --ulimit memlock=67108864:67108864 --ulimit nofile=65536:65536 --security-opt seccomp=unconfined
     strategy:
       fail-fast: false
       matrix:

.github/workflows/periodic_pagebench.yml

@@ -1,4 +1,4 @@
-name: Periodic pagebench performance test on unit-perf hetzner runner
+name: Periodic pagebench performance test on unit-perf-aws-arm runners
 
 on:
   schedule:
@@ -40,7 +40,7 @@ jobs:
       statuses: write
       contents: write
       pull-requests: write
-    runs-on: [ self-hosted, unit-perf ]
+    runs-on: [ self-hosted, unit-perf-aws-arm ]
     container:
       image: ghcr.io/neondatabase/build-tools:pinned-bookworm
       credentials:

.github/workflows/proxy-benchmark.yml

@@ -1,4 +1,4 @@
-name: Periodic proxy performance test on unit-perf hetzner runner
+name: Periodic proxy performance test on unit-perf-aws-arm runners
 
 on:
   push: # TODO: remove after testing
@@ -32,7 +32,7 @@ jobs:
       statuses: write
       contents: write
       pull-requests: write
-    runs-on: [self-hosted, unit-perf]
+    runs-on: [self-hosted, unit-perf-aws-arm]
     timeout-minutes: 60  # 1h timeout
     container:
       image: ghcr.io/neondatabase/build-tools:pinned-bookworm

.gitignore

@@ -25,11 +25,6 @@ compaction-suite-results.*
 *.o
 *.so
 *.Po
-*.pid
 
 # pgindent typedef lists
 *.list
-
-# various files for local testing
-/proxy/.subzero
-local_proxy.json

Cargo.toml

@@ -46,6 +46,7 @@ members = [
     "libs/proxy/postgres-types2",
     "libs/proxy/tokio-postgres2",
     "endpoint_storage",
+    "pgxn/neon/communicator",
 ]
 
 [workspace.package]
@@ -255,6 +256,7 @@ desim = { version = "0.1", path = "./libs/desim" }
 endpoint_storage = { version = "0.0.1", path = "./endpoint_storage/" }
 http-utils = { version = "0.1", path = "./libs/http-utils/" }
 metrics = { version = "0.1", path = "./libs/metrics/" }
+neon-shmem = { version = "0.1", path = "./libs/neon-shmem/" }
 pageserver = { path = "./pageserver" }
 pageserver_api = { version = "0.1", path = "./libs/pageserver_api/" }
 pageserver_client = { path = "./pageserver/client" }
@@ -284,6 +286,7 @@ walproposer = { version = "0.1", path = "./libs/walproposer/" }
 workspace_hack = { version = "0.1", path = "./workspace_hack/" }
 
 ## Build dependencies
+cbindgen = "0.29.0"
 criterion = "0.5.1"
 rcgen = "0.13"
 rstest = "0.18"

Dockerfile

@@ -30,7 +30,18 @@ ARG BASE_IMAGE_SHA=debian:${DEBIAN_FLAVOR}
 ARG BASE_IMAGE_SHA=${BASE_IMAGE_SHA/debian:bookworm-slim/debian@$BOOKWORM_SLIM_SHA}
 ARG BASE_IMAGE_SHA=${BASE_IMAGE_SHA/debian:bullseye-slim/debian@$BULLSEYE_SLIM_SHA}
 
-# Build Postgres
+# Naive way:
+#
+# 1. COPY . .
+# 1. make neon-pg-ext
+# 2. cargo build <storage binaries>
+#
+# But to enable docker to cache intermediate layers, we perform a few preparatory steps:
+#
+# - Build all postgres versions, depending on just the contents of vendor/
+# - Use cargo chef to build all rust dependencies
+
+# 1. Build all postgres versions
 FROM $REPOSITORY/$IMAGE:$TAG AS pg-build
 WORKDIR /home/nonroot
 
@@ -38,17 +49,15 @@ COPY --chown=nonroot vendor/postgres-v14 vendor/postgres-v14
 COPY --chown=nonroot vendor/postgres-v15 vendor/postgres-v15
 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
 RUN set -e \
-    && mold -run make -j $(nproc) -s neon-pg-ext \
-    && tar -C pg_install -czf /home/nonroot/postgres_install.tar.gz .
+    && mold -run make -j $(nproc) -s postgres
 
-# Prepare cargo-chef recipe
+# 2. Prepare cargo-chef recipe
 FROM $REPOSITORY/$IMAGE:$TAG AS plan
 WORKDIR /home/nonroot
 
@@ -56,23 +65,22 @@ COPY --chown=nonroot . .
 
 RUN cargo chef prepare --recipe-path recipe.json
 
-# Build neon binaries
+# Main build image
 FROM $REPOSITORY/$IMAGE:$TAG AS build
 WORKDIR /home/nonroot
 ARG GIT_VERSION=local
 ARG BUILD_TAG
-
-COPY --from=pg-build /home/nonroot/pg_install/v14/include/postgresql/server pg_install/v14/include/postgresql/server
-COPY --from=pg-build /home/nonroot/pg_install/v15/include/postgresql/server pg_install/v15/include/postgresql/server
-COPY --from=pg-build /home/nonroot/pg_install/v16/include/postgresql/server pg_install/v16/include/postgresql/server
-COPY --from=pg-build /home/nonroot/pg_install/v17/include/postgresql/server pg_install/v17/include/postgresql/server
-COPY --from=plan     /home/nonroot/recipe.json                              recipe.json
-
 ARG ADDITIONAL_RUSTFLAGS=""
 
+# 3. Build cargo dependencies. Note that this step doesn't depend on anything else than
+# `recipe.json`, so the layer can be reused as long as none of the dependencies change.
+COPY --from=plan     /home/nonroot/recipe.json                              recipe.json
 RUN set -e \
     && RUSTFLAGS="-Clinker=clang -Clink-arg=-fuse-ld=mold -Clink-arg=-Wl,--no-rosegment -Cforce-frame-pointers=yes ${ADDITIONAL_RUSTFLAGS}" cargo chef cook --locked --release --recipe-path recipe.json
 
+# Perform the main build. We reuse the Postgres build artifacts from the intermediate 'pg-build'
+# layer, and the cargo dependencies built in the previous step.
+COPY --chown=nonroot --from=pg-build /home/nonroot/pg_install/ pg_install
 COPY --chown=nonroot . .
 
 RUN set -e \
@@ -87,10 +95,10 @@ RUN set -e \
       --bin endpoint_storage \
       --bin neon_local \
       --bin storage_scrubber \
-      --locked --release
+      --locked --release \
+    && mold -run make -j $(nproc) -s neon-pg-ext
 
-# Build final image
-#
+# Assemble the final image
 FROM $BASE_IMAGE_SHA
 WORKDIR /data
 
@@ -130,12 +138,15 @@ COPY --from=build --chown=neon:neon /home/nonroot/target/release/proxy
 COPY --from=build --chown=neon:neon /home/nonroot/target/release/endpoint_storage    /usr/local/bin
 COPY --from=build --chown=neon:neon /home/nonroot/target/release/neon_local          /usr/local/bin
 COPY --from=build --chown=neon:neon /home/nonroot/target/release/storage_scrubber    /usr/local/bin
+COPY --from=build /home/nonroot/pg_install/v14 /usr/local/v14/
+COPY --from=build /home/nonroot/pg_install/v15 /usr/local/v15/
+COPY --from=build /home/nonroot/pg_install/v16 /usr/local/v16/
+COPY --from=build /home/nonroot/pg_install/v17 /usr/local/v17/
 
-COPY --from=pg-build /home/nonroot/pg_install/v14 /usr/local/v14/
-COPY --from=pg-build /home/nonroot/pg_install/v15 /usr/local/v15/
-COPY --from=pg-build /home/nonroot/pg_install/v16 /usr/local/v16/
-COPY --from=pg-build /home/nonroot/pg_install/v17 /usr/local/v17/
-COPY --from=pg-build /home/nonroot/postgres_install.tar.gz /data/
+# Deprecated: Old deployment scripts use this tarball which contains all the Postgres binaries.
+# That's obsolete, since all the same files are also present under /usr/local/v*. But to keep the
+# old scripts working for now, create the tarball.
+RUN tar -C /usr/local -cvzf /data/postgres_install.tar.gz v14 v15 v16 v17
 
 # By default, pageserver uses `.neon/` working directory in WORKDIR, so create one and fill it with the dummy config.
 # Now, when `docker run ... pageserver` is run, it can start without errors, yet will have some default dummy values.

Makefile

@@ -30,11 +30,18 @@ ifeq ($(BUILD_TYPE),release)
 	PG_CFLAGS += -O2 -g3 $(CFLAGS)
 	PG_LDFLAGS = $(LDFLAGS)
 	CARGO_PROFILE ?= --profile=release
+	# NEON_CARGO_ARTIFACT_TARGET_DIR is the directory where `cargo build` places
+	# the final build artifacts. There is unfortunately no easy way of changing
+	# it to a fully predictable path, nor to extract the path with a simple
+	# command. See https://github.com/rust-lang/cargo/issues/9661 and
+	# https://github.com/rust-lang/cargo/issues/6790.
+	NEON_CARGO_ARTIFACT_TARGET_DIR = $(ROOT_PROJECT_DIR)/target/release
 else ifeq ($(BUILD_TYPE),debug)
 	PG_CONFIGURE_OPTS = --enable-debug --with-openssl --enable-cassert --enable-depend
 	PG_CFLAGS += -O0 -g3 $(CFLAGS)
 	PG_LDFLAGS = $(LDFLAGS)
 	CARGO_PROFILE ?= --profile=dev
+	NEON_CARGO_ARTIFACT_TARGET_DIR = $(ROOT_PROJECT_DIR)/target/debug
 else
 	$(error Bad build type '$(BUILD_TYPE)', see Makefile for options)
 endif
@@ -102,7 +109,7 @@ all: neon postgres-install neon-pg-ext
 
 ### Neon Rust bits
 #
-# The 'postgres_ffi' depends on the Postgres headers.
+# The 'postgres_ffi' crate depends on the Postgres headers.
 .PHONY: neon
 neon: postgres-headers-install walproposer-lib cargo-target-dir
 	+@echo "Compiling Neon"
@@ -115,10 +122,13 @@ cargo-target-dir:
 	test -e target/CACHEDIR.TAG || echo "$(CACHEDIR_TAG_CONTENTS)" > target/CACHEDIR.TAG
 
 .PHONY: neon-pg-ext-%
-neon-pg-ext-%: postgres-install-%
+neon-pg-ext-%: postgres-install-% cargo-target-dir
 	+@echo "Compiling neon-specific Postgres extensions for $*"
 	mkdir -p $(BUILD_DIR)/pgxn-$*
-	$(MAKE) PG_CONFIG=$(POSTGRES_INSTALL_DIR)/$*/bin/pg_config COPT='$(COPT)' \
+	$(MAKE) PG_CONFIG="$(POSTGRES_INSTALL_DIR)/$*/bin/pg_config" COPT='$(COPT)' \
+		NEON_CARGO_ARTIFACT_TARGET_DIR="$(NEON_CARGO_ARTIFACT_TARGET_DIR)" \
+		CARGO_BUILD_FLAGS="$(CARGO_BUILD_FLAGS)" \
+		CARGO_PROFILE="$(CARGO_PROFILE)" \
 		-C $(BUILD_DIR)/pgxn-$*\
 		-f $(ROOT_PROJECT_DIR)/pgxn/Makefile  install
 

compute/compute-node.Dockerfile

@@ -1636,11 +1636,14 @@ RUN make install USE_PGXS=1 -j $(getconf _NPROCESSORS_ONLN)
 # compile neon extensions
 #
 #########################################################################################
-FROM pg-build AS neon-ext-build
+FROM pg-build-with-cargo AS neon-ext-build
 ARG PG_VERSION
 
-COPY pgxn/ pgxn/
-RUN make -j $(getconf _NPROCESSORS_ONLN) -C pgxn -s install-compute
+USER root
+COPY . .
+
+RUN make -j $(getconf _NPROCESSORS_ONLN) -C pgxn -s install-compute \
+      BUILD_TYPE=release CARGO_BUILD_FLAGS="--locked --release" NEON_CARGO_ARTIFACT_TARGET_DIR="$(pwd)/target/release"
 
 #########################################################################################
 #

compute_tools/src/compute.rs

@@ -29,7 +29,8 @@ use std::sync::atomic::{AtomicU32, AtomicU64, Ordering};
 use std::sync::{Arc, Condvar, Mutex, RwLock};
 use std::time::{Duration, Instant};
 use std::{env, fs};
-use tokio::spawn;
+use tokio::task::JoinHandle;
+use tokio::{spawn, time};
 use tracing::{Instrument, debug, error, info, instrument, warn};
 use url::Url;
 use utils::id::{TenantId, TimelineId};
@@ -107,6 +108,8 @@ pub struct ComputeNodeParams {
     pub installed_extensions_collection_interval: Arc<AtomicU64>,
 }
 
+type TaskHandle = Mutex<Option<JoinHandle<()>>>;
+
 /// Compute node info shared across several `compute_ctl` threads.
 pub struct ComputeNode {
     pub params: ComputeNodeParams,
@@ -129,7 +132,8 @@ pub struct ComputeNode {
     pub compute_ctl_config: ComputeCtlConfig,
 
     /// Handle to the extension stats collection task
-    extension_stats_task: Mutex<Option<tokio::task::JoinHandle<()>>>,
+    extension_stats_task: TaskHandle,
+    lfc_offload_task: TaskHandle,
 }
 
 // store some metrics about download size that might impact startup time
@@ -368,7 +372,7 @@ fn maybe_cgexec(cmd: &str) -> Command {
 
 struct PostgresHandle {
     postgres: std::process::Child,
-    log_collector: tokio::task::JoinHandle<Result<()>>,
+    log_collector: JoinHandle<Result<()>>,
 }
 
 impl PostgresHandle {
@@ -382,7 +386,7 @@ struct StartVmMonitorResult {
     #[cfg(target_os = "linux")]
     token: tokio_util::sync::CancellationToken,
     #[cfg(target_os = "linux")]
-    vm_monitor: Option<tokio::task::JoinHandle<Result<()>>>,
+    vm_monitor: Option<JoinHandle<Result<()>>>,
 }
 
 impl ComputeNode {
@@ -433,6 +437,7 @@ impl ComputeNode {
             ext_download_progress: RwLock::new(HashMap::new()),
             compute_ctl_config: config.compute_ctl_config,
             extension_stats_task: Mutex::new(None),
+            lfc_offload_task: Mutex::new(None),
         })
     }
 
@@ -520,8 +525,8 @@ impl ComputeNode {
             None
         };
 
-        // Terminate the extension stats collection task
         this.terminate_extension_stats_task();
+        this.terminate_lfc_offload_task();
 
         // Terminate the vm_monitor so it releases the file watcher on
         // /sys/fs/cgroup/neon-postgres.
@@ -851,12 +856,15 @@ impl ComputeNode {
         // Log metrics so that we can search for slow operations in logs
         info!(?metrics, postmaster_pid = %postmaster_pid, "compute start finished");
 
-        // Spawn the extension stats background task
         self.spawn_extension_stats_task();
 
         if pspec.spec.autoprewarm {
+            info!("autoprewarming on startup as requested");
             self.prewarm_lfc(None);
         }
+        if let Some(seconds) = pspec.spec.offload_lfc_interval_seconds {
+            self.spawn_lfc_offload_task(Duration::from_secs(seconds.into()));
+        };
         Ok(())
     }
 
@@ -2357,10 +2365,7 @@ 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();
-        }
+        self.terminate_extension_stats_task();
 
         let conf = self.tokio_conn_conf.clone();
         let atomic_interval = self.params.installed_extensions_collection_interval.clone();
@@ -2396,8 +2401,30 @@ LIMIT 100",
     }
 
     fn terminate_extension_stats_task(&self) {
-        if let Some(handle) = self.extension_stats_task.lock().unwrap().take() {
-            handle.abort();
+        if let Some(h) = self.extension_stats_task.lock().unwrap().take() {
+            h.abort()
+        }
+    }
+
+    pub fn spawn_lfc_offload_task(self: &Arc<Self>, interval: Duration) {
+        self.terminate_lfc_offload_task();
+        let secs = interval.as_secs();
+        info!("spawning lfc offload worker with {secs}s interval");
+        let this = self.clone();
+        let handle = spawn(async move {
+            let mut interval = time::interval(interval);
+            interval.tick().await; // returns immediately
+            loop {
+                interval.tick().await;
+                this.offload_lfc_async().await;
+            }
+        });
+        *self.lfc_offload_task.lock().unwrap() = Some(handle);
+    }
+
+    fn terminate_lfc_offload_task(&self) {
+        if let Some(h) = self.lfc_offload_task.lock().unwrap().take() {
+            h.abort()
         }
     }
 

compute_tools/src/compute_prewarm.rs

@@ -5,6 +5,7 @@ use compute_api::responses::LfcOffloadState;
 use compute_api::responses::LfcPrewarmState;
 use http::StatusCode;
 use reqwest::Client;
+use std::mem::replace;
 use std::sync::Arc;
 use tokio::{io::AsyncReadExt, spawn};
 use tracing::{error, info};
@@ -88,17 +89,15 @@ impl ComputeNode {
         self.state.lock().unwrap().lfc_offload_state.clone()
     }
 
-    /// Returns false if there is a prewarm request ongoing, true otherwise
+    /// If there is a prewarm request ongoing, return false, true otherwise
     pub fn prewarm_lfc(self: &Arc<Self>, from_endpoint: Option<String>) -> bool {
-        crate::metrics::LFC_PREWARM_REQUESTS.inc();
         {
             let state = &mut self.state.lock().unwrap().lfc_prewarm_state;
-            if let LfcPrewarmState::Prewarming =
-                std::mem::replace(state, LfcPrewarmState::Prewarming)
-            {
+            if let LfcPrewarmState::Prewarming = replace(state, LfcPrewarmState::Prewarming) {
                 return false;
             }
         }
+        crate::metrics::LFC_PREWARMS.inc();
 
         let cloned = self.clone();
         spawn(async move {
@@ -152,32 +151,41 @@ impl ComputeNode {
             .map(|_| ())
     }
 
-    /// Returns false if there is an offload request ongoing, true otherwise
+    /// If offload request is ongoing, return false, true otherwise
     pub fn offload_lfc(self: &Arc<Self>) -> bool {
-        crate::metrics::LFC_OFFLOAD_REQUESTS.inc();
         {
             let state = &mut self.state.lock().unwrap().lfc_offload_state;
-            if let LfcOffloadState::Offloading =
-                std::mem::replace(state, LfcOffloadState::Offloading)
-            {
+            if replace(state, LfcOffloadState::Offloading) == LfcOffloadState::Offloading {
                 return false;
             }
         }
-
         let cloned = self.clone();
-        spawn(async move {
-            let Err(err) = cloned.offload_lfc_impl().await else {
-                cloned.state.lock().unwrap().lfc_offload_state = LfcOffloadState::Completed;
-                return;
-            };
-            error!(%err);
-            cloned.state.lock().unwrap().lfc_offload_state = LfcOffloadState::Failed {
-                error: err.to_string(),
-            };
-        });
+        spawn(async move { cloned.offload_lfc_with_state_update().await });
         true
     }
 
+    pub async fn offload_lfc_async(self: &Arc<Self>) {
+        {
+            let state = &mut self.state.lock().unwrap().lfc_offload_state;
+            if replace(state, LfcOffloadState::Offloading) == LfcOffloadState::Offloading {
+                return;
+            }
+        }
+        self.offload_lfc_with_state_update().await
+    }
+
+    async fn offload_lfc_with_state_update(&self) {
+        crate::metrics::LFC_OFFLOADS.inc();
+        let Err(err) = self.offload_lfc_impl().await else {
+            self.state.lock().unwrap().lfc_offload_state = LfcOffloadState::Completed;
+            return;
+        };
+        error!(%err);
+        self.state.lock().unwrap().lfc_offload_state = LfcOffloadState::Failed {
+            error: err.to_string(),
+        };
+    }
+
     async fn offload_lfc_impl(&self) -> Result<()> {
         let EndpointStoragePair { url, token } = self.endpoint_storage_pair(None)?;
         info!(%url, "requesting LFC state from postgres");

compute_tools/src/metrics.rs

@@ -97,20 +97,18 @@ pub(crate) static PG_TOTAL_DOWNTIME_MS: Lazy<GenericCounter<AtomicU64>> = Lazy::
     .expect("failed to define a metric")
 });
 
-/// Needed as neon.file_cache_prewarm_batch == 0 doesn't mean we never tried to prewarm.
-/// On the other hand, LFC_PREWARMED_PAGES is excessive as we can GET /lfc/prewarm
-pub(crate) static LFC_PREWARM_REQUESTS: Lazy<IntCounter> = Lazy::new(|| {
+pub(crate) static LFC_PREWARMS: Lazy<IntCounter> = Lazy::new(|| {
     register_int_counter!(
-        "compute_ctl_lfc_prewarm_requests_total",
-        "Total number of LFC prewarm requests made by compute_ctl",
+        "compute_ctl_lfc_prewarms_total",
+        "Total number of LFC prewarms requested by compute_ctl or autoprewarm option",
     )
     .expect("failed to define a metric")
 });
 
-pub(crate) static LFC_OFFLOAD_REQUESTS: Lazy<IntCounter> = Lazy::new(|| {
+pub(crate) static LFC_OFFLOADS: Lazy<IntCounter> = Lazy::new(|| {
     register_int_counter!(
-        "compute_ctl_lfc_offload_requests_total",
-        "Total number of LFC offload requests made by compute_ctl",
+        "compute_ctl_lfc_offloads_total",
+        "Total number of LFC offloads requested by compute_ctl or lfc_offload_period_seconds option",
     )
     .expect("failed to define a metric")
 });
@@ -124,7 +122,7 @@ pub fn collect() -> Vec<MetricFamily> {
     metrics.extend(AUDIT_LOG_DIR_SIZE.collect());
     metrics.extend(PG_CURR_DOWNTIME_MS.collect());
     metrics.extend(PG_TOTAL_DOWNTIME_MS.collect());
-    metrics.extend(LFC_PREWARM_REQUESTS.collect());
-    metrics.extend(LFC_OFFLOAD_REQUESTS.collect());
+    metrics.extend(LFC_PREWARMS.collect());
+    metrics.extend(LFC_OFFLOADS.collect());
     metrics
 }

compute_tools/tests/pg_helpers_tests.rs

@@ -31,6 +31,7 @@ mod pg_helpers_tests {
 wal_level = logical
 hot_standby = on
 autoprewarm = off
+offload_lfc_interval_seconds = 20
 neon.safekeepers = '127.0.0.1:6502,127.0.0.1:6503,127.0.0.1:6501'
 wal_log_hints = on
 log_connections = on

control_plane/src/bin/neon_local.rs

@@ -675,6 +675,16 @@ struct EndpointStartCmdArgs {
     #[arg(default_value = "90s")]
     start_timeout: Duration,
 
+    #[clap(
+        long,
+        help = "Download LFC cache from endpoint storage on endpoint startup",
+        default_value = "false"
+    )]
+    autoprewarm: bool,
+
+    #[clap(long, help = "Upload LFC cache to endpoint storage periodically")]
+    offload_lfc_interval_seconds: Option<std::num::NonZeroU64>,
+
     #[clap(
         long,
         help = "Run in development mode, skipping VM-specific operations like process termination",
@@ -1585,22 +1595,24 @@ async fn handle_endpoint(subcmd: &EndpointCmd, env: &local_env::LocalEnv) -> Res
             let endpoint_storage_token = env.generate_auth_token(&claims)?;
             let endpoint_storage_addr = env.endpoint_storage.listen_addr.to_string();
 
+            let args = control_plane::endpoint::EndpointStartArgs {
+                auth_token,
+                endpoint_storage_token,
+                endpoint_storage_addr,
+                safekeepers_generation,
+                safekeepers,
+                pageservers,
+                remote_ext_base_url: remote_ext_base_url.clone(),
+                shard_stripe_size: stripe_size.0 as usize,
+                create_test_user: args.create_test_user,
+                start_timeout: args.start_timeout,
+                autoprewarm: args.autoprewarm,
+                offload_lfc_interval_seconds: args.offload_lfc_interval_seconds,
+                dev: args.dev,
+            };
+
             println!("Starting existing endpoint {endpoint_id}...");
-            endpoint
-                .start(
-                    &auth_token,
-                    endpoint_storage_token,
-                    endpoint_storage_addr,
-                    safekeepers_generation,
-                    safekeepers,
-                    pageservers,
-                    remote_ext_base_url.as_ref(),
-                    stripe_size.0 as usize,
-                    args.create_test_user,
-                    args.start_timeout,
-                    args.dev,
-                )
-                .await?;
+            endpoint.start(args).await?;
         }
         EndpointCmd::Reconfigure(args) => {
             let endpoint_id = &args.endpoint_id;

control_plane/src/endpoint.rs

@@ -373,6 +373,22 @@ impl std::fmt::Display for EndpointTerminateMode {
     }
 }
 
+pub struct EndpointStartArgs {
+    pub auth_token: Option<String>,
+    pub endpoint_storage_token: String,
+    pub endpoint_storage_addr: String,
+    pub safekeepers_generation: Option<SafekeeperGeneration>,
+    pub safekeepers: Vec<NodeId>,
+    pub pageservers: Vec<(PageserverProtocol, Host, u16)>,
+    pub remote_ext_base_url: Option<String>,
+    pub shard_stripe_size: usize,
+    pub create_test_user: bool,
+    pub start_timeout: Duration,
+    pub autoprewarm: bool,
+    pub offload_lfc_interval_seconds: Option<std::num::NonZeroU64>,
+    pub dev: bool,
+}
+
 impl Endpoint {
     fn from_dir_entry(entry: std::fs::DirEntry, env: &LocalEnv) -> Result<Endpoint> {
         if !entry.file_type()?.is_dir() {
@@ -677,21 +693,7 @@ impl Endpoint {
         })
     }
 
-    #[allow(clippy::too_many_arguments)]
-    pub async fn start(
-        &self,
-        auth_token: &Option<String>,
-        endpoint_storage_token: String,
-        endpoint_storage_addr: String,
-        safekeepers_generation: Option<SafekeeperGeneration>,
-        safekeepers: Vec<NodeId>,
-        pageservers: Vec<(PageserverProtocol, Host, u16)>,
-        remote_ext_base_url: Option<&String>,
-        shard_stripe_size: usize,
-        create_test_user: bool,
-        start_timeout: Duration,
-        dev: bool,
-    ) -> Result<()> {
+    pub async fn start(&self, args: EndpointStartArgs) -> Result<()> {
         if self.status() == EndpointStatus::Running {
             anyhow::bail!("The endpoint is already running");
         }
@@ -704,10 +706,10 @@ impl Endpoint {
             std::fs::remove_dir_all(self.pgdata())?;
         }
 
-        let pageserver_connstring = Self::build_pageserver_connstr(&pageservers);
+        let pageserver_connstring = Self::build_pageserver_connstr(&args.pageservers);
         assert!(!pageserver_connstring.is_empty());
 
-        let safekeeper_connstrings = self.build_safekeepers_connstrs(safekeepers)?;
+        let safekeeper_connstrings = self.build_safekeepers_connstrs(args.safekeepers)?;
 
         // check for file remote_extensions_spec.json
         // if it is present, read it and pass to compute_ctl
@@ -735,7 +737,7 @@ impl Endpoint {
                     cluster_id: None, // project ID: not used
                     name: None,       // project name: not used
                     state: None,
-                    roles: if create_test_user {
+                    roles: if args.create_test_user {
                         vec![Role {
                             name: PgIdent::from_str("test").unwrap(),
                             encrypted_password: None,
@@ -744,7 +746,7 @@ impl Endpoint {
                     } else {
                         Vec::new()
                     },
-                    databases: if create_test_user {
+                    databases: if args.create_test_user {
                         vec![Database {
                             name: PgIdent::from_str("neondb").unwrap(),
                             owner: PgIdent::from_str("test").unwrap(),
@@ -766,20 +768,21 @@ impl Endpoint {
                 endpoint_id: Some(self.endpoint_id.clone()),
                 mode: self.mode,
                 pageserver_connstring: Some(pageserver_connstring),
-                safekeepers_generation: safekeepers_generation.map(|g| g.into_inner()),
+                safekeepers_generation: args.safekeepers_generation.map(|g| g.into_inner()),
                 safekeeper_connstrings,
-                storage_auth_token: auth_token.clone(),
+                storage_auth_token: args.auth_token.clone(),
                 remote_extensions,
                 pgbouncer_settings: None,
-                shard_stripe_size: Some(shard_stripe_size),
+                shard_stripe_size: Some(args.shard_stripe_size),
                 local_proxy_config: None,
                 reconfigure_concurrency: self.reconfigure_concurrency,
                 drop_subscriptions_before_start: self.drop_subscriptions_before_start,
                 audit_log_level: ComputeAudit::Disabled,
                 logs_export_host: None::<String>,
-                endpoint_storage_addr: Some(endpoint_storage_addr),
-                endpoint_storage_token: Some(endpoint_storage_token),
-                autoprewarm: false,
+                endpoint_storage_addr: Some(args.endpoint_storage_addr),
+                endpoint_storage_token: Some(args.endpoint_storage_token),
+                autoprewarm: args.autoprewarm,
+                offload_lfc_interval_seconds: args.offload_lfc_interval_seconds,
                 suspend_timeout_seconds: -1, // Only used in neon_local.
             };
 
@@ -791,7 +794,7 @@ impl Endpoint {
                 debug!("spec.cluster {:?}", spec.cluster);
 
                 // fill missing fields again
-                if create_test_user {
+                if args.create_test_user {
                     spec.cluster.roles.push(Role {
                         name: PgIdent::from_str("test").unwrap(),
                         encrypted_password: None,
@@ -826,7 +829,7 @@ impl Endpoint {
         // Launch compute_ctl
         let conn_str = self.connstr("cloud_admin", "postgres");
         println!("Starting postgres node at '{conn_str}'");
-        if create_test_user {
+        if args.create_test_user {
             let conn_str = self.connstr("test", "neondb");
             println!("Also at '{conn_str}'");
         }
@@ -858,11 +861,11 @@ impl Endpoint {
         .stderr(logfile.try_clone()?)
         .stdout(logfile);
 
-        if let Some(remote_ext_base_url) = remote_ext_base_url {
-            cmd.args(["--remote-ext-base-url", remote_ext_base_url]);
+        if let Some(remote_ext_base_url) = args.remote_ext_base_url {
+            cmd.args(["--remote-ext-base-url", &remote_ext_base_url]);
         }
 
-        if dev {
+        if args.dev {
             cmd.arg("--dev");
         }
 
@@ -894,10 +897,11 @@ impl Endpoint {
                 Ok(state) => {
                     match state.status {
                         ComputeStatus::Init => {
-                            if Instant::now().duration_since(start_at) > start_timeout {
+                            let timeout = args.start_timeout;
+                            if Instant::now().duration_since(start_at) > timeout {
                                 bail!(
                                     "compute startup timed out {:?}; still in Init state",
-                                    start_timeout
+                                    timeout
                                 );
                             }
                             // keep retrying
@@ -925,9 +929,10 @@ impl Endpoint {
                     }
                 }
                 Err(e) => {
-                    if Instant::now().duration_since(start_at) > start_timeout {
+                    if Instant::now().duration_since(start_at) > args.start_timeout {
                         return Err(e).context(format!(
-                            "timed out {start_timeout:?} waiting to connect to compute_ctl HTTP",
+                            "timed out {:?} waiting to connect to compute_ctl HTTP",
+                            args.start_timeout
                         ));
                     }
                 }

control_plane/storcon_cli/src/main.rs

@@ -65,12 +65,27 @@ enum Command {
         #[arg(long)]
         scheduling: Option<NodeSchedulingPolicy>,
     },
-    // Set a node status as deleted.
+    /// Exists for backup usage and will be removed in future.
+    /// Use [`Command::NodeStartDelete`] instead, if possible.
     NodeDelete {
         #[arg(long)]
         node_id: NodeId,
     },
+    /// Start deletion of the specified pageserver.
+    NodeStartDelete {
+        #[arg(long)]
+        node_id: NodeId,
+    },
+    /// Cancel deletion of the specified pageserver and wait for `timeout`
+    /// for the operation to be canceled. May be retried.
+    NodeCancelDelete {
+        #[arg(long)]
+        node_id: NodeId,
+        #[arg(long)]
+        timeout: humantime::Duration,
+    },
     /// Delete a tombstone of node from the storage controller.
+    /// This is used when we want to allow the node to be re-registered.
     NodeDeleteTombstone {
         #[arg(long)]
         node_id: NodeId,
@@ -912,10 +927,43 @@ async fn main() -> anyhow::Result<()> {
                 .await?;
         }
         Command::NodeDelete { node_id } => {
+            eprintln!("Warning: This command is obsolete and will be removed in a future version");
+            eprintln!("Use `NodeStartDelete` instead, if possible");
             storcon_client
                 .dispatch::<(), ()>(Method::DELETE, format!("control/v1/node/{node_id}"), None)
                 .await?;
         }
+        Command::NodeStartDelete { node_id } => {
+            storcon_client
+                .dispatch::<(), ()>(
+                    Method::PUT,
+                    format!("control/v1/node/{node_id}/delete"),
+                    None,
+                )
+                .await?;
+            println!("Delete started for {node_id}");
+        }
+        Command::NodeCancelDelete { node_id, timeout } => {
+            storcon_client
+                .dispatch::<(), ()>(
+                    Method::DELETE,
+                    format!("control/v1/node/{node_id}/delete"),
+                    None,
+                )
+                .await?;
+
+            println!("Waiting for node {node_id} to quiesce on scheduling policy ...");
+
+            let final_policy =
+                wait_for_scheduling_policy(storcon_client, node_id, *timeout, |sched| {
+                    !matches!(sched, NodeSchedulingPolicy::Deleting)
+                })
+                .await?;
+
+            println!(
+                "Delete was cancelled for node {node_id}. Schedulling policy is now {final_policy:?}"
+            );
+        }
         Command::NodeDeleteTombstone { node_id } => {
             storcon_client
                 .dispatch::<(), ()>(

docs/rfcs/035-safekeeper-dynamic-membership-change.md

@@ -20,7 +20,7 @@ In our case consensus leader is compute (walproposer), and we don't want to wake
 up all computes for the change. Neither we want to fully reimplement the leader
 logic second time outside compute. Because of that the proposed algorithm relies
 for issuing configurations on the external fault tolerant (distributed) strongly
-consisent storage with simple API: CAS (compare-and-swap) on the single key.
+consistent storage with simple API: CAS (compare-and-swap) on the single key.
 Properly configured postgres suits this.
 
 In the system consensus is implemented at the timeline level, so algorithm below
@@ -34,7 +34,7 @@ A configuration is
 
 ```
 struct Configuration {
-    generation: Generation, // a number uniquely identifying configuration
+    generation: SafekeeperGeneration, // a number uniquely identifying configuration
     sk_set: Vec<NodeId>, // current safekeeper set
     new_sk_set: Optional<Vec<NodeId>>,
 }
@@ -81,11 +81,11 @@ configuration generation in them is less than its current one. Namely, it
 refuses to vote, to truncate WAL in `handle_elected` and to accept WAL. In
 response it sends its current configuration generation to let walproposer know.
 
-Safekeeper gets `PUT /v1/tenants/{tenant_id}/timelines/{timeline_id}/configuration`
-accepting `Configuration`. Safekeeper switches to the given conf it is higher than its
+Safekeeper gets `PUT /v1/tenants/{tenant_id}/timelines/{timeline_id}/membership`
+accepting `Configuration`. Safekeeper switches to the given conf if it is higher than its
 current one and ignores it otherwise. In any case it replies with
 ```
-struct ConfigurationSwitchResponse {
+struct TimelineMembershipSwitchResponse {
     conf: Configuration,
     term: Term,
     last_log_term: Term,
@@ -108,7 +108,7 @@ establishes this configuration as its own and moves to voting.
 It should stop talking to safekeepers not listed in the configuration at this
 point, though it is not unsafe to continue doing so.
 
-To be elected it must receive votes from both majorites if `new_sk_set` is present.
+To be elected it must receive votes from both majorities if `new_sk_set` is present.
 Similarly, to commit WAL it must receive flush acknowledge from both majorities.
 
 If walproposer hears from safekeeper configuration higher than his own (i.e.
@@ -130,7 +130,7 @@ storage are reachable.
 1) Fetch current timeline configuration from the configuration storage.
 2) If it is already joint one and `new_set` is different from `desired_set`
    refuse to change. However, assign join conf to (in memory) var
-   `join_conf` and proceed to step 4 to finish the ongoing change.
+   `joint_conf` and proceed to step 4 to finish the ongoing change.
 3) Else, create joint `joint_conf: Configuration`: increment current conf number
    `n` and put `desired_set` to `new_sk_set`. Persist it in the configuration
    storage by doing CAS on the current generation: change happens only if
@@ -161,11 +161,11 @@ storage are reachable.
    because `pull_timeline` already includes it and plus additionally would be
    broadcast by compute. More importantly, we may proceed to the next step
    only when `<last_log_term, flush_lsn>` on the majority of the new set reached
-   `sync_position`. Similarly, on the happy path no waiting is not needed because
+   `sync_position`. Similarly, on the happy path no waiting is needed because
    `pull_timeline` already includes it. However, we should double
     check to be safe. For example, timeline could have been created earlier e.g.
     manually or after try-to-migrate, abort, try-to-migrate-again sequence.
-7) Create `new_conf: Configuration` incrementing `join_conf` generation and having new
+7) Create `new_conf: Configuration` incrementing `joint_conf` generation and having new
    safekeeper set as `sk_set` and None `new_sk_set`. Write it to configuration
    storage under one more CAS.
 8) Call `PUT` `configuration` on safekeepers from the new set,
@@ -178,12 +178,12 @@ spec of it.
 
 Description above focuses on safety. To make the flow practical and live, here a few more
 considerations.
-1) It makes sense to ping new set to ensure it we are migrating to live node(s) before
+1) It makes sense to ping new set to ensure we are migrating to live node(s) before
   step 3.
 2) If e.g. accidentally wrong new sk set has been specified, before CAS in step `6` is completed
    it is safe to rollback to the old conf with one more CAS.
 3) On step 4 timeline might be already created on members of the new set for various reasons;
-   the simplest is the procedure restart. There are more complicated scenarious like mentioned
+   the simplest is the procedure restart. There are more complicated scenarios like mentioned
    in step 5. Deleting and re-doing `pull_timeline` is generally unsafe without involving
    generations, so seems simpler to treat existing timeline as success. However, this also
    has a disadvantage: you might imagine an surpassingly unlikely schedule where condition in
@@ -192,7 +192,7 @@ considerations.
 4) In the end timeline should be locally deleted on the safekeeper(s) which are
    in the old set but not in the new one, unless they are unreachable. To be
    safe this also should be done under generation number (deletion proceeds only if
-   current configuration is <= than one in request and safekeeper is not memeber of it).
+   current configuration is <= than one in request and safekeeper is not member of it).
 5) If current conf fetched on step 1 is already not joint and members equal to `desired_set`,
    jump to step 7, using it as `new_conf`.
 
@@ -261,14 +261,14 @@ Timeline (branch) creation in cplane should call storage_controller POST
 Response should be augmented with `safekeepers_generation` and `safekeepers`
 fields like described in `/notify-safekeepers` above. Initially (currently)
 these fields may be absent; in this case cplane chooses safekeepers on its own
-like it currently does. The call should be retried until succeeds.
+like it currently does. The call should be retried until it succeeds.
 
 Timeline deletion and tenant deletion in cplane should call appropriate
 storage_controller endpoints like it currently does for sharded tenants. The
 calls should be retried until they succeed.
 
-When compute receives safekeepers list from control plane it needs to know the
-generation to checked whether it should be updated (note that compute may get
+When compute receives safekeeper list from control plane it needs to know the
+generation to check whether it should be updated (note that compute may get
 safekeeper list from either cplane or safekeepers). Currently `neon.safekeepers`
 GUC is just a comma separates list of `host:port`. Let's prefix it with
 `g#<generation>:` to this end, so it will look like
@@ -305,8 +305,8 @@ enum MigrationRequest {
 ```
 
 `FinishPending` requests to run the procedure to ensure state is clean: current
-configuration is not joint and majority of safekeepers are aware of it, but do
-not attempt to migrate anywhere. If current configuration fetched on step 1 is
+configuration is not joint and the majority of safekeepers are aware of it, but do
+not attempt to migrate anywhere. If the current configuration fetched on step 1 is
 not joint it jumps to step 7. It should be run at startup for all timelines (but
 similarly, in the first version it is ok to trigger it manually).
 
@@ -315,7 +315,7 @@ similarly, in the first version it is ok to trigger it manually).
 `safekeepers` table mirroring current `nodes` should be added, except that for
 `scheduling_policy`: it is enough to have at least in the beginning only 3
 fields: 1) `active` 2) `paused` (initially means only not assign new tlis there
-3) `decomissioned` (node is removed).
+3) `decommissioned` (node is removed).
 
 `timelines` table:
 ```
@@ -326,9 +326,10 @@ table! {
         tenant_id -> Varchar,
         start_lsn -> pg_lsn,
         generation -> Int4,
-        sk_set -> Array<Int4>, // list of safekeeper ids
+        sk_set -> Array<Int8>, // list of safekeeper ids
         new_sk_set -> Nullable<Array<Int8>>, // list of safekeeper ids, null if not joint conf
         cplane_notified_generation -> Int4,
+        sk_set_notified_generation -> Int4, // the generation a quorum of sk_set knows about
         deleted_at -> Nullable<Timestamptz>,
     }
 }
@@ -338,13 +339,23 @@ table! {
 might also want to add ancestor_timeline_id to preserve the hierarchy, but for
 this RFC it is not needed.
 
+`cplane_notified_generation` and `sk_set_notified_generation` fields are used to
+track the last stage of the algorithm, when we need to notify safekeeper set and cplane
+with the final configuration after it's already committed to DB.
+
+The timeline is up-to-date (no migration in progress) if `new_sk_set` is null and
+`*_notified_generation` fields are up to date with `generation`. 
+
+It's possible to replace `*_notified_generation` with one boolean field `migration_completed`,
+but for better observability it's nice to have them separately.
+
 #### API
 
 Node management is similar to pageserver:
-1) POST `/control/v1/safekeepers` inserts safekeeper.
-2) GET `/control/v1/safekeepers` lists safekeepers.
-3) GET `/control/v1/safekeepers/:node_id` gets safekeeper.
-4) PUT `/control/v1/safekepers/:node_id/status` changes status to e.g.
+1) POST `/control/v1/safekeeper` inserts safekeeper.
+2) GET `/control/v1/safekeeper` lists safekeepers.
+3) GET `/control/v1/safekeeper/:node_id` gets safekeeper.
+4) PUT `/control/v1/safekeper/:node_id/scheduling_policy` changes status to e.g.
    `offline` or `decomissioned`. Initially it is simpler not to schedule any
     migrations here.
 
@@ -368,8 +379,8 @@ Migration API: the first version is the simplest and the most imperative:
 all timelines from one safekeeper to another. It accepts json
 ```
 {
-    "src_sk": u32,
-    "dst_sk": u32,
+    "src_sk": NodeId,
+    "dst_sk": NodeId,
     "limit": Optional<u32>,
 }
 ```
@@ -379,12 +390,15 @@ Returns list of scheduled requests.
 2) PUT `/control/v1/tenant/:tenant_id/timeline/:timeline_id/safekeeper_migrate` schedules `MigrationRequest`
    to move single timeline to given set of safekeepers:
 ```
-{
-    "desired_set": Vec<u32>,
+struct TimelineSafekeeperMigrateRequest {
+    "new_sk_set": Vec<NodeId>,
 }
 ```
 
-Returns scheduled request.
+In the first version the handler migrates the timeline to `new_sk_set` synchronously.
+Should be retried until success.
+
+In the future we might change it to asynchronous API and return scheduled request.
 
 Similar call should be added for the tenant.
 
@@ -434,6 +448,9 @@ table! {
 }
 ```
 
+We load all pending ops from the table on startup into the memory.
+The table is needed only to preserve the state between restarts.
+
 `op_type` can be `include` (seed from peers and ensure generation is up to
 date), `exclude` (remove locally) and `delete`. Field is actually not strictly
 needed as it can be computed from current configuration, but gives more explicit
@@ -474,7 +491,7 @@ actions must be idempotent. Now, a tricky point here is timeline start LSN. For
 the initial (tenant creation) call cplane doesn't know it. However, setting
 start_lsn on safekeepers during creation is a good thing -- it provides a
 guarantee that walproposer can always find a common point in WAL histories of
-safekeeper and its own, and so absense of it would be a clear sign of
+safekeeper and its own, and so absence of it would be a clear sign of
 corruption. The following sequence works:
 1) Create timeline (or observe that it exists) on pageserver,
    figuring out last_record_lsn in response.
@@ -497,11 +514,9 @@ corruption. The following sequence works:
    retries the call until 200 response.
 
    There is a small question how request handler (timeline creation in this
-   case) would interact with per sk reconciler. As always I prefer to do the
-   simplest possible thing and here it seems to be just waking it up so it
-   re-reads the db for work to do. Passing work in memory is faster, but
-   that shouldn't matter, and path to scan db for work will exist anyway, 
-   simpler to reuse it.
+   case) would interact with per sk reconciler. In the current implementation
+   we first persist the request in the DB, and then send an in-memory request
+   to each safekeeper reconciler to process it.
 
 For pg version / wal segment size: while we may persist them in `timelines`
 table, it is not necessary as initial creation at step 3 can take them from
@@ -509,30 +524,40 @@ pageserver or cplane creation call and later pull_timeline will carry them
 around.
 
 Timeline migration.
-1) CAS to the db to create joint conf, and in the same transaction create
-   `safekeeper_timeline_pending_ops` `include` entries to initialize new members
-   as well as deliver this conf to current ones; poke per sk reconcilers to work
-   on it. Also any conf change should also poke cplane notifier task(s).
-2) Once it becomes possible per alg description above, get out of joint conf
-   with another CAS. Task should get wakeups from per sk reconcilers because 
-   conf switch is required for advancement; however retries should be sleep
-   based as well as LSN advancement might be needed, though in happy path 
-   it isn't. To see whether further transition is possible on wakup migration
-   executor polls safekeepers per the algorithm. CAS creating new conf with only
-   new members should again insert entries to `safekeeper_timeline_pending_ops`
-   to switch them there, as well as `exclude` rows to remove timeline from 
-   old members.
+1) CAS to the db to create joint conf. Since this moment the migration is considered to be 
+   "in progress". We can detect all "in-progress" migrations looking into the database.
+2) Do steps 4-6 from the algorithm, including `pull_timeline` onto `new_sk_set`, update membership
+   configuration on all safekeepers, notify cplane, etc. All operations are idempotent,
+   so we don't need to persist anything in the database at this stage. If any errors occur,
+   it's safe to retry or abort the migration.
+3) Once it becomes possible per alg description above, get out of joint conf
+   with another CAS. Also should insert `exclude` entries into `safekeeper_timeline_pending_ops`
+   in the same DB transaction. Adding `exclude` entries atomically is nesessary because after
+   CAS we don't have the list of excluded safekeepers in the `timelines` table anymore, but we
+   need to have them persisted somewhere in case the migration is interrupted right after the CAS.
+4) Finish the migration. The final membership configuration is committed to the DB at this stage.
+   So, the migration can not be aborted anymore. But it can still be retried if the migration fails
+   past stage 3. To finish the migration we need to send the new membership configuration to
+   a new quorum of safekeepers, notify cplane with the new safekeeper list and schedule the `exclude`
+   requests to in-memory queue for safekeeper reconciler. If the algrorithm is retried, it's
+   possible that we have already committed `exclude` requests to DB, but didn't send them to
+   the in-memory queue. In this case we need to read them from `safekeeper_timeline_pending_ops`
+   because it's the only place where they are persistent. The fields `sk_set_notified_generation`
+   and `cplane_notified_generation` are updated after each step. The migration is considered
+   fully completed when they match the `generation` field.
+
+In practice, we can report "success" after stage 3 and do the "finish" step in per-timeline
+reconciler (if we implement it). But it's wise to at least try to finish them synchronously,
+so the timeline is always in a "good state" and doesn't require an old quorum to commit
+WAL after the migration reported "success".
 
 Timeline deletion: just set `deleted_at` on the timeline row and insert
 `safekeeper_timeline_pending_ops` entries in the same xact, the rest is done by
 per sk reconcilers.
 
-When node is removed (set to `decomissioned`), `safekeeper_timeline_pending_ops`
+When node is removed (set to `decommissioned`), `safekeeper_timeline_pending_ops`
 for it must be cleared in the same transaction.
 
-One more task pool should infinitely retry notifying control plane about changed
-safekeeper sets (trying making `cplane_notified_generation` equal `generation`).
-
 #### Dealing with multiple instances of storage_controller
 
 Operations described above executed concurrently might create some errors but do
@@ -541,7 +566,7 @@ of storage_controller it is fine to have it temporarily, e.g. during redeploy.
 
 To harden against some controller instance creating some work in
 `safekeeper_timeline_pending_ops` and then disappearing without anyone pickup up
-the job per sk reconcilers apart from explicit wakups should scan for work
+the job per sk reconcilers apart from explicit wakeups should scan for work
 periodically. It is possible to remove that though if all db updates are
 protected with leadership token/term -- then such scans are needed only after
 leadership is acquired.
@@ -563,7 +588,7 @@ There should be following layers of tests:
    safekeeper communication and pull_timeline need to be mocked and main switch
    procedure wrapped to as a node (thread) in simulation tests, using these
    mocks. Test would inject migrations like it currently injects
-   safekeeper/walproposer restars. Main assert is the same -- committed WAL must
+   safekeeper/walproposer restarts. Main assert is the same -- committed WAL must
    not be lost.
 
 3) Since simulation testing injects at relatively high level points (not
@@ -613,7 +638,7 @@ Let's have the following implementation bits for gradual rollout:
   `notify-safekeepers`.
 
 Then the rollout for a region would be:
-- Current situation: safekeepers are choosen by control_plane.
+- Current situation: safekeepers are chosen by control_plane.
 - We manually migrate some timelines, test moving them around.
 - Then we enable `--set-safekeepers` so that all new timelines
   are on storage controller.

libs/compute_api/src/responses.rs

@@ -58,7 +58,7 @@ pub enum LfcPrewarmState {
     },
 }
 
-#[derive(Serialize, Default, Debug, Clone)]
+#[derive(Serialize, Default, Debug, Clone, PartialEq)]
 #[serde(tag = "status", rename_all = "snake_case")]
 pub enum LfcOffloadState {
     #[default]

libs/compute_api/src/spec.rs

@@ -181,10 +181,14 @@ pub struct ComputeSpec {
     /// JWT for authorizing requests to endpoint storage service
     pub endpoint_storage_token: Option<String>,
 
-    /// Download LFC state from endpoint_storage and pass it to Postgres on startup
     #[serde(default)]
+    /// Download LFC state from endpoint storage and pass it to Postgres on compute startup
     pub autoprewarm: bool,
 
+    #[serde(default)]
+    /// Upload LFC state to endpoint storage periodically. Default value (None) means "don't upload"
+    pub offload_lfc_interval_seconds: Option<std::num::NonZeroU64>,
+
     /// Suspend timeout in seconds.
     ///
     /// We use this value to derive other values, such as the installed extensions metric.

libs/compute_api/tests/cluster_spec.json

@@ -90,6 +90,11 @@
                 "value": "off",
                 "vartype": "bool"
             },
+            {
+                "name": "offload_lfc_interval_seconds",
+                "value": "20",
+                "vartype": "integer"
+            },
             {
                 "name": "neon.safekeepers",
                 "value": "127.0.0.1:6502,127.0.0.1:6503,127.0.0.1:6501",

libs/pageserver_api/src/controller_api.rs

@@ -386,6 +386,7 @@ pub enum NodeSchedulingPolicy {
     Pause,
     PauseForRestart,
     Draining,
+    Deleting,
 }
 
 impl FromStr for NodeSchedulingPolicy {
@@ -398,6 +399,7 @@ impl FromStr for NodeSchedulingPolicy {
             "pause" => Ok(Self::Pause),
             "pause_for_restart" => Ok(Self::PauseForRestart),
             "draining" => Ok(Self::Draining),
+            "deleting" => Ok(Self::Deleting),
             _ => Err(anyhow::anyhow!("Unknown scheduling state '{s}'")),
         }
     }
@@ -412,6 +414,7 @@ impl From<NodeSchedulingPolicy> for String {
             Pause => "pause",
             PauseForRestart => "pause_for_restart",
             Draining => "draining",
+            Deleting => "deleting",
         }
         .to_string()
     }

libs/pageserver_api/src/models.rs

@@ -1602,6 +1602,10 @@ pub struct TimelineInfo {
     /// The status of the rel_size migration.
     pub rel_size_migration: Option<RelSizeMigration>,
 
+    /// Whether the timeline is read-only (doesn't support WAL advancing).
+    /// The property is not recursive; child timelines can be non-read-only.
+    pub read_only: bool,
+
     /// Whether the timeline is invisible in synthetic size calculations.
     pub is_invisible: Option<bool>,
 }

libs/safekeeper_api/src/models.rs

@@ -221,7 +221,7 @@ pub struct TimelineMembershipSwitchRequest {
 pub struct TimelineMembershipSwitchResponse {
     pub previous_conf: Configuration,
     pub current_conf: Configuration,
-    pub term: Term,
+    pub last_log_term: Term,
     pub flush_lsn: Lsn,
 }
 

pageserver/src/config.rs

@@ -28,6 +28,7 @@ use reqwest::Url;
 use storage_broker::Uri;
 use utils::id::{NodeId, TimelineId};
 use utils::logging::{LogFormat, SecretString};
+use utils::serde_percent::Percent;
 
 use crate::tenant::storage_layer::inmemory_layer::IndexEntry;
 use crate::tenant::{TENANTS_SEGMENT_NAME, TIMELINES_SEGMENT_NAME};
@@ -459,7 +460,16 @@ impl PageServerConf {
             metric_collection_endpoint,
             metric_collection_bucket,
             synthetic_size_calculation_interval,
-            disk_usage_based_eviction,
+            disk_usage_based_eviction: Some(disk_usage_based_eviction.unwrap_or(
+                DiskUsageEvictionTaskConfig {
+                    max_usage_pct: Percent::new(80).unwrap(),
+                    min_avail_bytes: 2_000_000_000,
+                    period: Duration::from_secs(60),
+                    #[cfg(feature = "testing")]
+                    mock_statvfs: None,
+                    eviction_order: Default::default(),
+                },
+            )),
             test_remote_failures,
             ondemand_download_behavior_treat_error_as_warn,
             background_task_maximum_delay,
@@ -697,6 +707,8 @@ impl ConfigurableSemaphore {
 #[cfg(test)]
 mod tests {
 
+    use std::time::Duration;
+
     use camino::Utf8PathBuf;
     use rstest::rstest;
     use utils::id::NodeId;
@@ -798,4 +810,20 @@ mod tests {
         PageServerConf::parse_and_validate(NodeId(0), config_toml, &workdir)
             .expect("parse_and_validate");
     }
+
+    #[test]
+    fn test_config_disk_usage_based_eviction_is_valid() {
+        let input = r#"
+            control_plane_api = "http://localhost:6666"
+        "#;
+        let config_toml = toml_edit::de::from_str::<pageserver_api::config::ConfigToml>(input)
+            .expect("disk_usage_based_eviction is valid");
+        let workdir = Utf8PathBuf::from("/nonexistent");
+        let config = PageServerConf::parse_and_validate(NodeId(0), config_toml, &workdir).unwrap();
+        let disk_usage_based_eviction = config.disk_usage_based_eviction.unwrap();
+        assert_eq!(disk_usage_based_eviction.max_usage_pct.get(), 80);
+        assert_eq!(disk_usage_based_eviction.min_avail_bytes, 2_000_000_000);
+        assert_eq!(disk_usage_based_eviction.period, Duration::from_secs(60));
+        assert_eq!(disk_usage_based_eviction.eviction_order, Default::default());
+    }
 }

pageserver/src/http/routes.rs

@@ -452,6 +452,7 @@ async fn build_timeline_info_common(
     let state = timeline.current_state();
     // Report is_archived = false if the timeline is still loading
     let is_archived = timeline.is_archived().unwrap_or(false);
+    let read_only = timeline.is_read_only().unwrap_or(false);
     let remote_consistent_lsn_projected = timeline
         .get_remote_consistent_lsn_projected()
         .unwrap_or(Lsn(0));
@@ -503,6 +504,7 @@ async fn build_timeline_info_common(
 
         state,
         is_archived: Some(is_archived),
+        read_only,
         rel_size_migration: Some(timeline.get_rel_size_v2_status()),
         is_invisible: Some(is_invisible),
 
@@ -591,12 +593,13 @@ async fn timeline_create_handler(
         TimelineCreateRequestMode::Branch {
             ancestor_timeline_id,
             ancestor_start_lsn,
-            read_only: _,
+            read_only,
             pg_version: _,
         } => tenant::CreateTimelineParams::Branch(tenant::CreateTimelineParamsBranch {
             new_timeline_id,
             ancestor_timeline_id,
             ancestor_start_lsn,
+            read_only,
         }),
         TimelineCreateRequestMode::ImportPgdata {
             import_pgdata:
@@ -3698,7 +3701,7 @@ 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) 
+        // 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();
         if as_type.as_deref() == Some("boolean") {

pageserver/src/tenant.rs

@@ -943,6 +943,7 @@ pub(crate) struct CreateTimelineParamsBranch {
     pub(crate) new_timeline_id: TimelineId,
     pub(crate) ancestor_timeline_id: TimelineId,
     pub(crate) ancestor_start_lsn: Option<Lsn>,
+    pub(crate) read_only: bool,
 }
 
 #[derive(Debug)]
@@ -2572,6 +2573,7 @@ impl TenantShard {
             initdb_lsn,
             pg_version,
         );
+        let read_only = false;
         self.prepare_new_timeline(
             new_timeline_id,
             &new_metadata,
@@ -2579,6 +2581,7 @@ impl TenantShard {
             initdb_lsn,
             None,
             None,
+            read_only,
             ctx,
         )
         .await
@@ -2723,6 +2726,7 @@ impl TenantShard {
                 new_timeline_id,
                 ancestor_timeline_id,
                 mut ancestor_start_lsn,
+                read_only,
             }) => {
                 let ancestor_timeline = self
                     .get_timeline(ancestor_timeline_id, false)
@@ -2775,8 +2779,14 @@ impl TenantShard {
                         })?;
                 }
 
-                self.branch_timeline(&ancestor_timeline, new_timeline_id, ancestor_start_lsn, ctx)
-                    .await?
+                self.branch_timeline(
+                    &ancestor_timeline,
+                    new_timeline_id,
+                    ancestor_start_lsn,
+                    read_only,
+                    ctx,
+                )
+                .await?
             }
             CreateTimelineParams::ImportPgdata(params) => {
                 self.create_timeline_import_pgdata(params, ctx).await?
@@ -2901,6 +2911,7 @@ impl TenantShard {
                     initdb_lsn,
                     PgMajorVersion::PG15,
                 );
+                let read_only = false;
                 this.prepare_new_timeline(
                     new_timeline_id,
                     &new_metadata,
@@ -2908,6 +2919,7 @@ impl TenantShard {
                     initdb_lsn,
                     None,
                     None,
+                    read_only,
                     ctx,
                 )
                 .await
@@ -4957,9 +4969,10 @@ impl TenantShard {
         src_timeline: &Arc<Timeline>,
         dst_id: TimelineId,
         start_lsn: Option<Lsn>,
+        read_only: bool,
         ctx: &RequestContext,
     ) -> Result<CreateTimelineResult, CreateTimelineError> {
-        self.branch_timeline_impl(src_timeline, dst_id, start_lsn, ctx)
+        self.branch_timeline_impl(src_timeline, dst_id, start_lsn, read_only, ctx)
             .await
     }
 
@@ -4968,6 +4981,7 @@ impl TenantShard {
         src_timeline: &Arc<Timeline>,
         dst_id: TimelineId,
         start_lsn: Option<Lsn>,
+        read_only: bool,
         ctx: &RequestContext,
     ) -> Result<CreateTimelineResult, CreateTimelineError> {
         let src_id = src_timeline.timeline_id;
@@ -5079,6 +5093,7 @@ impl TenantShard {
                 start_lsn + 1,
                 Some(Arc::clone(src_timeline)),
                 Some(src_timeline.get_rel_size_v2_status()),
+                read_only,
                 ctx,
             )
             .await?;
@@ -5343,6 +5358,7 @@ impl TenantShard {
             }
         }
         let pgdata_lsn = import_datadir::get_lsn_from_controlfile(&pgdata_path)?.align();
+        let read_only = false;
 
         // Import the contents of the data directory at the initial checkpoint
         // LSN, and any WAL after that.
@@ -5365,6 +5381,7 @@ impl TenantShard {
                 pgdata_lsn,
                 None,
                 None,
+                read_only,
                 ctx,
             )
             .await?;
@@ -5448,14 +5465,17 @@ impl TenantShard {
         start_lsn: Lsn,
         ancestor: Option<Arc<Timeline>>,
         rel_size_v2_status: Option<RelSizeMigration>,
+        read_only: bool,
         ctx: &RequestContext,
     ) -> anyhow::Result<(UninitializedTimeline<'a>, RequestContext)> {
         let tenant_shard_id = self.tenant_shard_id;
 
         let resources = self.build_timeline_resources(new_timeline_id);
-        resources
-            .remote_client
-            .init_upload_queue_for_empty_remote(new_metadata, rel_size_v2_status.clone())?;
+        resources.remote_client.init_upload_queue_for_empty_remote(
+            new_metadata,
+            rel_size_v2_status.clone(),
+            read_only,
+        )?;
 
         let (timeline_struct, timeline_ctx) = self
             .create_timeline_struct(

pageserver/src/tenant/remote_timeline_client.rs

@@ -444,6 +444,7 @@ impl RemoteTimelineClient {
         &self,
         local_metadata: &TimelineMetadata,
         rel_size_v2_status: Option<RelSizeMigration>,
+        read_only: bool,
     ) -> anyhow::Result<()> {
         // Set the maximum number of inprogress tasks to the remote storage concurrency. There's
         // certainly no point in starting more upload tasks than this.
@@ -456,6 +457,9 @@ impl RemoteTimelineClient {
         let initialized_queue =
             upload_queue.initialize_empty_remote(local_metadata, inprogress_limit)?;
         initialized_queue.dirty.rel_size_migration = rel_size_v2_status;
+        if read_only {
+            initialized_queue.dirty.read_only = Some(read_only);
+        }
         self.update_remote_physical_size_gauge(None);
         info!("initialized upload queue as empty");
         Ok(())
@@ -583,6 +587,17 @@ impl RemoteTimelineClient {
             .ok()
     }
 
+    /// Returns whether the timeline is archived.
+    /// Return None if the remote index_part hasn't been downloaded yet.
+    pub(crate) fn is_read_only(&self) -> Option<bool> {
+        self.upload_queue
+            .lock()
+            .unwrap()
+            .initialized_mut()
+            .map(|q| q.clean.0.read_only.unwrap_or_default())
+            .ok()
+    }
+
     /// Returns `Ok(Some(timestamp))` if the timeline has been archived, `Ok(None)` if the timeline hasn't been archived.
     ///
     /// Return Err(_) if the remote index_part hasn't been downloaded yet, or the timeline hasn't been stopped yet.

pageserver/src/tenant/remote_timeline_client/index.rs

@@ -114,6 +114,12 @@ pub struct IndexPart {
     /// The timestamp when the timeline was marked invisible in synthetic size calculations.
     #[serde(skip_serializing_if = "Option::is_none", default)]
     pub(crate) marked_invisible_at: Option<NaiveDateTime>,
+
+    /// Whether the timeline is read only or not.
+    ///
+    /// The property is non-recursive, so child timelines can be non-read-only,
+    /// but it can't be changed for an individual timeline once the timeline has been created.
+    pub(crate) read_only: Option<bool>,
 }
 
 #[derive(Debug, PartialEq, Eq, Clone, Serialize, Deserialize)]
@@ -142,6 +148,7 @@ impl IndexPart {
     /// - 12: +l2_lsn
     /// - 13: +gc_compaction
     /// - 14: +marked_invisible_at
+    /// - 15: +read_only
     const LATEST_VERSION: usize = 14;
 
     // Versions we may see when reading from a bucket.
@@ -165,6 +172,7 @@ impl IndexPart {
             l2_lsn: None,
             gc_compaction: None,
             marked_invisible_at: None,
+            read_only: None,
         }
     }
 
@@ -475,6 +483,7 @@ mod tests {
             l2_lsn: None,
             gc_compaction: None,
             marked_invisible_at: None,
+            read_only: None,
         };
 
         let part = IndexPart::from_json_bytes(example.as_bytes()).unwrap();
@@ -524,6 +533,7 @@ mod tests {
             l2_lsn: None,
             gc_compaction: None,
             marked_invisible_at: None,
+            read_only: None,
         };
 
         let part = IndexPart::from_json_bytes(example.as_bytes()).unwrap();
@@ -574,6 +584,7 @@ mod tests {
             l2_lsn: None,
             gc_compaction: None,
             marked_invisible_at: None,
+            read_only: None,
         };
 
         let part = IndexPart::from_json_bytes(example.as_bytes()).unwrap();
@@ -675,6 +686,7 @@ mod tests {
             l2_lsn: None,
             gc_compaction: None,
             marked_invisible_at: None,
+            read_only: None,
         };
 
         let part = IndexPart::from_json_bytes(example.as_bytes()).unwrap();
@@ -726,6 +738,7 @@ mod tests {
             l2_lsn: None,
             gc_compaction: None,
             marked_invisible_at: None,
+            read_only: None,
         };
 
         let part = IndexPart::from_json_bytes(example.as_bytes()).unwrap();
@@ -782,6 +795,7 @@ mod tests {
             l2_lsn: None,
             gc_compaction: None,
             marked_invisible_at: None,
+            read_only: None,
         };
 
         let part = IndexPart::from_json_bytes(example.as_bytes()).unwrap();
@@ -843,6 +857,7 @@ mod tests {
             l2_lsn: None,
             gc_compaction: None,
             marked_invisible_at: None,
+            read_only: None,
         };
 
         let part = IndexPart::from_json_bytes(example.as_bytes()).unwrap();
@@ -905,6 +920,7 @@ mod tests {
             l2_lsn: None,
             gc_compaction: None,
             marked_invisible_at: None,
+            read_only: None,
         };
 
         let part = IndexPart::from_json_bytes(example.as_bytes()).unwrap();
@@ -972,6 +988,7 @@ mod tests {
             l2_lsn: None,
             gc_compaction: None,
             marked_invisible_at: None,
+            read_only: None,
         };
 
         let part = IndexPart::from_json_bytes(example.as_bytes()).unwrap();
@@ -1052,6 +1069,7 @@ mod tests {
             l2_lsn: None,
             gc_compaction: None,
             marked_invisible_at: None,
+            read_only: None,
         };
 
         let part = IndexPart::from_json_bytes(example.as_bytes()).unwrap();
@@ -1133,6 +1151,7 @@ mod tests {
             l2_lsn: None,
             gc_compaction: None,
             marked_invisible_at: None,
+            read_only: None,
         };
 
         let part = IndexPart::from_json_bytes(example.as_bytes()).unwrap();
@@ -1220,6 +1239,7 @@ mod tests {
                 last_completed_lsn: "0/16960E8".parse::<Lsn>().unwrap(),
             }),
             marked_invisible_at: None,
+            read_only: None,
         };
 
         let part = IndexPart::from_json_bytes(example.as_bytes()).unwrap();
@@ -1308,6 +1328,97 @@ mod tests {
                 last_completed_lsn: "0/16960E8".parse::<Lsn>().unwrap(),
             }),
             marked_invisible_at: Some(parse_naive_datetime("2023-07-31T09:00:00.123000000")),
+            read_only: None,
+        };
+
+        let part = IndexPart::from_json_bytes(example.as_bytes()).unwrap();
+        assert_eq!(part, expected);
+    }
+
+    #[test]
+    fn v15_read_only_is_parsed() {
+        let example = r#"{
+            "version": 14,
+            "layer_metadata":{
+                "000000000000000000000000000000000000-FFFFFFFFFFFFFFFFFFFFFFFFFFFFFFFFFFFF__0000000001696070-00000000016960E9": { "file_size": 25600000 },
+                "000000000000000000000000000000000000-FFFFFFFFFFFFFFFFFFFFFFFFFFFFFFFFFFFF__00000000016B59D8-00000000016B5A51": { "file_size": 9007199254741001 }
+            },
+            "disk_consistent_lsn":"0/16960E8",
+            "metadata": {
+                "disk_consistent_lsn": "0/16960E8",
+                "prev_record_lsn": "0/1696070",
+                "ancestor_timeline": "e45a7f37d3ee2ff17dc14bf4f4e3f52e",
+                "ancestor_lsn": "0/0",
+                "latest_gc_cutoff_lsn": "0/1696070",
+                "initdb_lsn": "0/1696070",
+                "pg_version": 14
+            },
+            "gc_blocking": {
+                "started_at": "2024-07-19T09:00:00.123",
+                "reasons": ["DetachAncestor"]
+            },
+            "import_pgdata": {
+                "V1": {
+                    "Done": {
+                        "idempotency_key": "specified-by-client-218a5213-5044-4562-a28d-d024c5f057f5",
+                        "started_at": "2024-11-13T09:23:42.123",
+                        "finished_at": "2024-11-13T09:42:23.123"
+                    }
+                }
+            },
+            "rel_size_migration": "legacy",
+            "l2_lsn": "0/16960E8",
+            "gc_compaction": {
+                "last_completed_lsn": "0/16960E8"
+            },
+            "marked_invisible_at": "2023-07-31T09:00:00.123",
+            "read_only": true,
+        }"#;
+
+        let expected = IndexPart {
+            version: 14,
+            layer_metadata: HashMap::from([
+                ("000000000000000000000000000000000000-FFFFFFFFFFFFFFFFFFFFFFFFFFFFFFFFFFFF__0000000001696070-00000000016960E9".parse().unwrap(), LayerFileMetadata {
+                    file_size: 25600000,
+                    generation: Generation::none(),
+                    shard: ShardIndex::unsharded()
+                }),
+                ("000000000000000000000000000000000000-FFFFFFFFFFFFFFFFFFFFFFFFFFFFFFFFFFFF__00000000016B59D8-00000000016B5A51".parse().unwrap(), LayerFileMetadata {
+                    file_size: 9007199254741001,
+                    generation: Generation::none(),
+                    shard: ShardIndex::unsharded()
+                })
+            ]),
+            disk_consistent_lsn: "0/16960E8".parse::<Lsn>().unwrap(),
+            metadata: TimelineMetadata::new(
+                Lsn::from_str("0/16960E8").unwrap(),
+                Some(Lsn::from_str("0/1696070").unwrap()),
+                Some(TimelineId::from_str("e45a7f37d3ee2ff17dc14bf4f4e3f52e").unwrap()),
+                Lsn::INVALID,
+                Lsn::from_str("0/1696070").unwrap(),
+                Lsn::from_str("0/1696070").unwrap(),
+                PgMajorVersion::PG14,
+            ).with_recalculated_checksum().unwrap(),
+            deleted_at: None,
+            lineage: Default::default(),
+            gc_blocking: Some(GcBlocking {
+                started_at: parse_naive_datetime("2024-07-19T09:00:00.123000000"),
+                reasons: enumset::EnumSet::from_iter([GcBlockingReason::DetachAncestor]),
+            }),
+            last_aux_file_policy: Default::default(),
+            archived_at: None,
+            import_pgdata: Some(import_pgdata::index_part_format::Root::V1(import_pgdata::index_part_format::V1::Done(import_pgdata::index_part_format::Done{
+                started_at: parse_naive_datetime("2024-11-13T09:23:42.123000000"),
+                finished_at: parse_naive_datetime("2024-11-13T09:42:23.123000000"),
+                idempotency_key: import_pgdata::index_part_format::IdempotencyKey::new("specified-by-client-218a5213-5044-4562-a28d-d024c5f057f5".to_string()),
+            }))),
+            rel_size_migration: Some(RelSizeMigration::Legacy),
+            l2_lsn: Some("0/16960E8".parse::<Lsn>().unwrap()),
+            gc_compaction: Some(GcCompactionState {
+                last_completed_lsn: "0/16960E8".parse::<Lsn>().unwrap(),
+            }),
+            marked_invisible_at: Some(parse_naive_datetime("2023-07-31T09:00:00.123000000")),
+            read_only: Some(true),
         };
 
         let part = IndexPart::from_json_bytes(example.as_bytes()).unwrap();

pageserver/src/tenant/timeline.rs

@@ -2144,14 +2144,31 @@ impl Timeline {
         debug_assert_current_span_has_tenant_and_timeline_id();
 
         // Regardless of whether we're going to try_freeze_and_flush
-        // or not, stop ingesting any more data.
+        // cancel walreceiver to stop ingesting more data asap.
+        //
+        // Note that we're accepting a race condition here where we may
+        // do the final flush below, before walreceiver observes the
+        // cancellation and exits.
+        // This means we may open a new InMemoryLayer after the final flush below.
+        // Flush loop is also still running for a short while, so, in theory, it
+        // could also make its way into the upload queue.
+        //
+        // If we wait for the shutdown of the walreceiver before moving on to the
+        // flush, then that would be avoided. But we don't do it because the
+        // walreceiver entertains reads internally, which means that it possibly
+        // depends on the download of layers. Layer download is only sensitive to
+        // the cancellation of the entire timeline, so cancelling the walreceiver
+        // will have no effect on the individual get requests.
+        // This would cause problems when there is a lot of ongoing downloads or
+        // there is S3 unavailabilities, i.e. detach, deletion, etc would hang,
+        // and we can't deallocate resources of the timeline, etc.
         let walreceiver = self.walreceiver.lock().unwrap().take();
         tracing::debug!(
             is_some = walreceiver.is_some(),
             "Waiting for WalReceiverManager..."
         );
         if let Some(walreceiver) = walreceiver {
-            walreceiver.shutdown().await;
+            walreceiver.cancel().await;
         }
         // ... and inform any waiters for newer LSNs that there won't be any.
         self.last_record_lsn.shutdown();
@@ -2332,6 +2349,10 @@ impl Timeline {
         self.remote_client.is_invisible()
     }
 
+    pub(crate) fn is_read_only(&self) -> Option<bool> {
+        self.remote_client.is_read_only()
+    }
+
     pub(crate) fn is_stopping(&self) -> bool {
         self.current_state() == TimelineState::Stopping
     }

pageserver/src/tenant/timeline/walreceiver.rs

@@ -63,7 +63,6 @@ pub struct WalReceiver {
     /// All task spawned by [`WalReceiver::start`] and its children are sensitive to this token.
     /// It's a child token of [`Timeline`] so that timeline shutdown can cancel WalReceiver tasks early for `freeze_and_flush=true`.
     cancel: CancellationToken,
-    task: tokio::task::JoinHandle<()>,
 }
 
 impl WalReceiver {
@@ -80,7 +79,7 @@ impl WalReceiver {
         let loop_status = Arc::new(std::sync::RwLock::new(None));
         let manager_status = Arc::clone(&loop_status);
         let cancel = timeline.cancel.child_token();
-        let task = WALRECEIVER_RUNTIME.spawn({
+        let _task = WALRECEIVER_RUNTIME.spawn({
             let cancel = cancel.clone();
             async move {
                 debug_assert_current_span_has_tenant_and_timeline_id();
@@ -121,25 +120,14 @@ impl WalReceiver {
         Self {
             manager_status,
             cancel,
-            task,
         }
     }
 
     #[instrument(skip_all, level = tracing::Level::DEBUG)]
-    pub async fn shutdown(self) {
+    pub async fn cancel(self) {
         debug_assert_current_span_has_tenant_and_timeline_id();
         debug!("cancelling walreceiver tasks");
         self.cancel.cancel();
-        match self.task.await {
-            Ok(()) => debug!("Shutdown success"),
-            Err(je) if je.is_cancelled() => unreachable!("not used"),
-            Err(je) if je.is_panic() => {
-                // already logged by panic hook
-            }
-            Err(je) => {
-                error!("shutdown walreceiver task join error: {je}")
-            }
-        }
     }
 
     pub(crate) fn status(&self) -> Option<ConnectionManagerStatus> {

pageserver/src/tenant/timeline/walreceiver/connection_manager.rs

@@ -100,6 +100,7 @@ pub(super) async fn connection_manager_loop_step(
     // with other streams on this client (other connection managers). When
     // object goes out of scope, stream finishes in drop() automatically.
     let mut broker_subscription = subscribe_for_timeline_updates(broker_client, id, cancel).await?;
+    let mut broker_reset_interval = tokio::time::interval(tokio::time::Duration::from_secs(30));
     debug!("Subscribed for broker timeline updates");
 
     loop {
@@ -156,7 +157,10 @@ pub(super) async fn connection_manager_loop_step(
             // Got a new update from the broker
             broker_update = broker_subscription.message() /* TODO: review cancellation-safety */ => {
                 match broker_update {
-                    Ok(Some(broker_update)) => connection_manager_state.register_timeline_update(broker_update),
+                    Ok(Some(broker_update)) => {
+                        broker_reset_interval.reset();
+                        connection_manager_state.register_timeline_update(broker_update);
+                    },
                     Err(status) => {
                         match status.code() {
                             Code::Unknown if status.message().contains("stream closed because of a broken pipe") || status.message().contains("connection reset") || status.message().contains("error reading a body from connection") => {
@@ -178,6 +182,14 @@ pub(super) async fn connection_manager_loop_step(
                 }
             },
 
+            _ = broker_reset_interval.tick() => {
+                if wait_lsn_status.borrow().is_some() {
+                    tracing::warn!("No broker updates received for a while, but waiting for WAL. Re-setting stream ...")
+                }
+
+                broker_subscription = subscribe_for_timeline_updates(broker_client, id, cancel).await?;
+            },
+
             new_event = async {
                 // Reminder: this match arm needs to be cancellation-safe.
                 loop {

pageserver/src/tenant/timeline/walreceiver/walreceiver_connection.rs

@@ -275,20 +275,12 @@ pub(super) async fn handle_walreceiver_connection(
     let copy_stream = replication_client.copy_both_simple(&query).await?;
     let mut physical_stream = pin!(ReplicationStream::new(copy_stream));
 
-    let walingest_future = WalIngest::new(timeline.as_ref(), startpoint, &ctx);
-    let walingest_res = select! {
-        walingest_res = walingest_future => walingest_res,
-        _ = cancellation.cancelled() => {
-            // We are doing reads in WalIngest::new, and those can hang as they come from the network.
-            // Timeline cancellation hits the walreceiver cancellation token before it hits the timeline global one.
-            debug!("Connection cancelled");
-            return Err(WalReceiverError::Cancelled);
-        },
-    };
-    let mut walingest = walingest_res.map_err(|e| match e.kind {
-        crate::walingest::WalIngestErrorKind::Cancelled => WalReceiverError::Cancelled,
-        _ => WalReceiverError::Other(e.into()),
-    })?;
+    let mut walingest = WalIngest::new(timeline.as_ref(), startpoint, &ctx)
+        .await
+        .map_err(|e| match e.kind {
+            crate::walingest::WalIngestErrorKind::Cancelled => WalReceiverError::Cancelled,
+            _ => WalReceiverError::Other(e.into()),
+        })?;
 
     let (format, compression) = match protocol {
         PostgresClientProtocol::Interpreted {

pgxn/neon/Makefile

@@ -22,7 +22,8 @@ OBJS = \
 	walproposer.o \
 	walproposer_pg.o \
 	neon_ddl_handler.o \
-	walsender_hooks.o
+	walsender_hooks.o \
+	$(NEON_CARGO_ARTIFACT_TARGET_DIR)/libcommunicator.a
 
 PG_CPPFLAGS = -I$(libpq_srcdir)
 SHLIB_LINK_INTERNAL = $(libpq)
@@ -54,6 +55,17 @@ WALPROP_OBJS = \
 	neon_utils.o \
 	walproposer_compat.o
 
+# libcommunicator.a is built by cargo from the Rust sources under communicator/
+# subdirectory. `cargo build` also generates communicator_bindings.h.
+neon.o: communicator/communicator_bindings.h
+
+$(NEON_CARGO_ARTIFACT_TARGET_DIR)/libcommunicator.a communicator/communicator_bindings.h &:
+	(cd $(srcdir)/communicator && cargo build $(CARGO_BUILD_FLAGS) $(CARGO_PROFILE))
+
+# Force `cargo build` every time. Some of the Rust sources might have
+# changed.
+.PHONY: $(NEON_CARGO_ARTIFACT_TARGET_DIR)/libcommunicator.a communicator/communicator_bindings.h
+
 .PHONY: walproposer-lib
 walproposer-lib: CPPFLAGS += -DWALPROPOSER_LIB
 walproposer-lib: libwalproposer.a;

pgxn/neon/communicator/.gitignore

@@ -0,0 +1,2 @@
+# generated file (with cbindgen, see build.rs)
+communicator_bindings.h

pgxn/neon/communicator/Cargo.toml

@@ -0,0 +1,20 @@
+[package]
+name = "communicator"
+version = "0.1.0"
+license.workspace = true
+edition.workspace = true
+
+[lib]
+crate-type = ["staticlib"]
+
+[features]
+# 'testing' feature is currently unused in the communicator, but we accept it for convenience of
+# calling build scripts, so that you can pass the same feature to all packages.
+testing = []
+
+[dependencies]
+neon-shmem.workspace = true
+workspace_hack = { version = "0.1", path = "../../../workspace_hack" }
+
+[build-dependencies]
+cbindgen.workspace = true

pgxn/neon/communicator/README.md

@@ -0,0 +1,8 @@
+This package will evolve into a "compute-pageserver communicator"
+process and machinery. For now, it's just a dummy that doesn't do
+anything interesting, but it allows us to test the compilation and
+linking of Rust code into the Postgres extensions.
+
+At compilation time, pgxn/neon/communicator/ produces a static
+library, libcommunicator.a. It is linked to the neon.so extension
+library.

pgxn/neon/communicator/build.rs

@@ -0,0 +1,20 @@
+use std::env;
+
+fn main() -> Result<(), Box<dyn std::error::Error>> {
+    let crate_dir = env::var("CARGO_MANIFEST_DIR").unwrap();
+
+    match cbindgen::generate(crate_dir) {
+        Ok(bindings) => {
+            bindings.write_to_file("communicator_bindings.h");
+        }
+        Err(cbindgen::Error::ParseSyntaxError { .. }) => {
+            // This means there was a syntax error in the Rust sources. Don't panic, because
+            // we want the build to continue and the Rust compiler to hit the error. The
+            // Rust compiler produces a better error message than cbindgen.
+            eprintln!("Generating C bindings failed because of a Rust syntax error");
+        }
+        Err(err) => panic!("Unable to generate C bindings: {err:?}"),
+    };
+
+    Ok(())
+}

pgxn/neon/communicator/cbindgen.toml

@@ -0,0 +1,4 @@
+language = "C"
+
+[enum]
+prefix_with_name = true

pgxn/neon/communicator/src/lib.rs

@@ -0,0 +1,6 @@
+/// dummy function, just to test linking Rust functions into the C
+/// extension
+#[unsafe(no_mangle)]
+pub extern "C" fn communicator_dummy(arg: u32) -> u32 {
+    arg + 1
+}

pgxn/neon/neon.c

@@ -43,6 +43,9 @@
 #include "storage/ipc.h"
 #endif
 
+/* the rust bindings, generated by cbindgen */
+#include "communicator/communicator_bindings.h"
+
 PG_MODULE_MAGIC;
 void		_PG_init(void);
 
@@ -452,6 +455,9 @@ _PG_init(void)
 	shmem_startup_hook = neon_shmem_startup_hook;
 #endif
 
+	/* dummy call to a Rust function in the communicator library, to check that it works */
+	(void) communicator_dummy(123);
+
 	pg_init_libpagestore();
 	lfc_init();
 	pg_init_walproposer();

pgxn/neon/neon_ddl_handler.c

@@ -98,12 +98,14 @@ typedef struct
 typedef struct DdlHashTable
 {
 	struct DdlHashTable *prev_table;
+	size_t		subtrans_level;
 	HTAB	   *db_table;
 	HTAB	   *role_table;
 } DdlHashTable;
 
 static DdlHashTable RootTable;
 static DdlHashTable *CurrentDdlTable = &RootTable;
+static int SubtransLevel; /* current nesting level of subtransactions */
 
 static void
 PushKeyValue(JsonbParseState **state, char *key, char *value)
@@ -332,9 +334,25 @@ SendDeltasToControlPlane()
 	}
 }
 
+static void
+InitCurrentDdlTableIfNeeded()
+{
+	/* Lazy construction of DllHashTable chain */
+	if (SubtransLevel > CurrentDdlTable->subtrans_level)
+	{
+		DdlHashTable *new_table = MemoryContextAlloc(CurTransactionContext, sizeof(DdlHashTable));
+		new_table->prev_table = CurrentDdlTable;
+		new_table->subtrans_level = SubtransLevel;
+		new_table->role_table = NULL;
+		new_table->db_table = NULL;
+		CurrentDdlTable = new_table;
+	}
+}
+
 static void
 InitDbTableIfNeeded()
 {
+	InitCurrentDdlTableIfNeeded();
 	if (!CurrentDdlTable->db_table)
 	{
 		HASHCTL		db_ctl = {};
@@ -353,6 +371,7 @@ InitDbTableIfNeeded()
 static void
 InitRoleTableIfNeeded()
 {
+	InitCurrentDdlTableIfNeeded();
 	if (!CurrentDdlTable->role_table)
 	{
 		HASHCTL		role_ctl = {};
@@ -371,19 +390,21 @@ InitRoleTableIfNeeded()
 static void
 PushTable()
 {
-	DdlHashTable *new_table = MemoryContextAlloc(CurTransactionContext, sizeof(DdlHashTable));
-
-	new_table->prev_table = CurrentDdlTable;
-	new_table->role_table = NULL;
-	new_table->db_table = NULL;
-	CurrentDdlTable = new_table;
+	SubtransLevel += 1;
 }
 
 static void
 MergeTable()
 {
-	DdlHashTable *old_table = CurrentDdlTable;
+	DdlHashTable *old_table;
 
+	Assert(SubtransLevel >= CurrentDdlTable->subtrans_level);
+	if (--SubtransLevel >= CurrentDdlTable->subtrans_level)
+	{
+		return;
+	}
+
+	old_table = CurrentDdlTable;
 	CurrentDdlTable = old_table->prev_table;
 
 	if (old_table->db_table)
@@ -476,11 +497,15 @@ MergeTable()
 static void
 PopTable()
 {
-	/*
-	 * Current table gets freed because it is allocated in aborted
-	 * subtransaction's memory context.
-	 */
-	CurrentDdlTable = CurrentDdlTable->prev_table;
+	Assert(SubtransLevel >= CurrentDdlTable->subtrans_level);
+	if (--SubtransLevel < CurrentDdlTable->subtrans_level)
+	{
+		/*
+		 * Current table gets freed because it is allocated in aborted
+		 * subtransaction's memory context.
+		 */
+		CurrentDdlTable = CurrentDdlTable->prev_table;
+	}
 }
 
 static void

proxy/Cargo.toml

@@ -5,9 +5,8 @@ edition = "2024"
 license.workspace = true
 
 [features]
-default = ["rest_broker"]
+default = []
 testing = ["dep:tokio-postgres"]
-rest_broker = ["subzero-core", "jsonpath_lib", "ouroboros"]
 
 [dependencies]
 ahash.workspace = true
@@ -104,9 +103,6 @@ uuid.workspace = true
 x509-cert.workspace = true
 redis.workspace = true
 zerocopy.workspace = true
-subzero-core = { git = "https://github.com/neondatabase-labs/subzero", rev = "0b3d3278f5f9ac9311a7280cb1676de80e021f06", features = ["postgresql"], optional = true }
-jsonpath_lib = { version = "0.3.0", optional = true }
-ouroboros = { version = "0.18", optional = true }
 
 # jwt stuff
 jose-jwa = "0.1.2"

proxy/README.md

@@ -170,8 +170,8 @@ Create a configuration file called `local_proxy.json` in the root of the repo (u
 
 Start the local proxy:
 ```sh
-cargo run --bin local_proxy --features testing -- \
-  --disable-pg-session-jwt \
+cargo run --bin local_proxy -- \
+  --disable_pg_session_jwt true \
   --http 0.0.0.0:7432
 ```
 
@@ -180,7 +180,6 @@ Start the auth broker:
 LOGFMT=text OTEL_SDK_DISABLED=true cargo run --bin proxy --features testing -- \
   -c server.crt -k server.key \
   --is-auth-broker true \
-  --is-rest-broker true \
   --wss 0.0.0.0:8080 \
   --http 0.0.0.0:7002 \
   --auth-backend local
@@ -198,9 +197,3 @@ curl -k "https://foo.local.neon.build:8080/sql" \
   -H "neon-connection-string: postgresql://authenticator@foo.local.neon.build/database" \
   -d '{"query":"select 1","params":[]}'
 ```
-
-Make a rest request against the auth broker (rest broker):
-```sh
-curl -k "https://foo.local.neon.build:8080/database/rest/v1/items?select=id,name&id=eq.1" \
--H "Authorization: Bearer $NEON_JWT"
-```

proxy/src/binary/local_proxy.rs

@@ -20,9 +20,6 @@ use crate::auth::backend::jwt::JwkCache;
 use crate::auth::backend::local::LocalBackend;
 use crate::auth::{self};
 use crate::cancellation::CancellationHandler;
-
-#[cfg(feature = "rest_broker")]
-use crate::config::RestConfig;
 use crate::config::{
     self, AuthenticationConfig, ComputeConfig, HttpConfig, ProxyConfig, RetryConfig,
     refresh_config_loop,
@@ -279,11 +276,6 @@ fn build_config(args: &LocalProxyCliArgs) -> anyhow::Result<&'static ProxyConfig
             accept_jwts: true,
             console_redirect_confirmation_timeout: Duration::ZERO,
         },
-        #[cfg(feature = "rest_broker")]
-        rest_config: RestConfig {
-            is_rest_broker: false,
-            db_schema_cache: None,
-        },
         proxy_protocol_v2: config::ProxyProtocolV2::Rejected,
         handshake_timeout: Duration::from_secs(10),
         wake_compute_retry_config: RetryConfig::parse(RetryConfig::WAKE_COMPUTE_DEFAULT_VALUES)?,

proxy/src/binary/proxy.rs

@@ -31,8 +31,6 @@ use crate::auth::backend::local::LocalBackend;
 use crate::auth::backend::{ConsoleRedirectBackend, MaybeOwned};
 use crate::batch::BatchQueue;
 use crate::cancellation::{CancellationHandler, CancellationProcessor};
-#[cfg(feature = "rest_broker")]
-use crate::config::RestConfig;
 #[cfg(any(test, feature = "testing"))]
 use crate::config::refresh_config_loop;
 use crate::config::{
@@ -49,8 +47,6 @@ use crate::redis::{elasticache, notifications};
 use crate::scram::threadpool::ThreadPool;
 use crate::serverless::GlobalConnPoolOptions;
 use crate::serverless::cancel_set::CancelSet;
-#[cfg(feature = "rest_broker")]
-use crate::serverless::rest::DbSchemaCache;
 use crate::tls::client_config::compute_client_config_with_root_certs;
 #[cfg(any(test, feature = "testing"))]
 use crate::url::ApiUrl;
@@ -248,12 +244,10 @@ struct ProxyCliArgs {
 
     /// if this is not local proxy, this toggles whether we accept Postgres REST requests
     #[clap(long, default_value_t = false, value_parser = clap::builder::BoolishValueParser::new(), action = clap::ArgAction::Set)]
-    #[cfg(feature = "rest_broker")]
     is_rest_broker: bool,
 
     /// cache for `db_schema_cache` introspection (use `size=0` to disable)
     #[clap(long, default_value = "size=1000,ttl=1h")]
-    #[cfg(feature = "rest_broker")]
     db_schema_cache: String,
 }
 
@@ -521,17 +515,6 @@ pub async fn run() -> anyhow::Result<()> {
     ));
     maintenance_tasks.spawn(control_plane::mgmt::task_main(mgmt_listener));
 
-    // add a task to flush the db_schema cache every 10 minutes
-    #[cfg(feature = "rest_broker")]
-    if let Some(db_schema_cache) = &config.rest_config.db_schema_cache {
-        maintenance_tasks.spawn(async move {
-            loop {
-                tokio::time::sleep(Duration::from_secs(600)).await;
-                db_schema_cache.flush();
-            }
-        });
-    }
-
     if let Some(metrics_config) = &config.metric_collection {
         // TODO: Add gc regardles of the metric collection being enabled.
         maintenance_tasks.spawn(usage_metrics::task_main(metrics_config));
@@ -699,28 +682,6 @@ fn build_config(args: &ProxyCliArgs) -> anyhow::Result<&'static ProxyConfig> {
         timeout: Duration::from_secs(2),
     };
 
-    #[cfg(feature = "rest_broker")]
-    let rest_config = {
-        let db_schema_cache_config: CacheOptions = args.db_schema_cache.parse()?;
-        info!("Using DbSchemaCache with options={db_schema_cache_config:?}");
-
-        let db_schema_cache = if args.is_rest_broker {
-            Some(DbSchemaCache::new(
-                "db_schema_cache",
-                db_schema_cache_config.size,
-                db_schema_cache_config.ttl,
-                true,
-            ))
-        } else {
-            None
-        };
-
-        RestConfig {
-            is_rest_broker: args.is_rest_broker,
-            db_schema_cache,
-        }
-    };
-
     let config = ProxyConfig {
         tls_config,
         metric_collection,
@@ -733,8 +694,6 @@ fn build_config(args: &ProxyCliArgs) -> anyhow::Result<&'static ProxyConfig> {
         connect_to_compute: compute_config,
         #[cfg(feature = "testing")]
         disable_pg_session_jwt: false,
-        #[cfg(feature = "rest_broker")]
-        rest_config,
     };
 
     let config = Box::leak(Box::new(config));

proxy/src/cache/timed_lru.rs

@@ -204,10 +204,6 @@ impl<K: Hash + Eq + Clone, V: Clone> TimedLru<K, V> {
         self.insert_raw_ttl(key, value, ttl, false);
     }
 
-    pub(crate) fn insert(&self, key: K, value: V) {
-        self.insert_raw_ttl(key, value, self.ttl, self.update_ttl_on_retrieval);
-    }
-
     pub(crate) fn insert_unit(&self, key: K, value: V) -> (Option<V>, Cached<&Self, ()>) {
         let (_, old) = self.insert_raw(key.clone(), value);
 
@@ -218,28 +214,6 @@ impl<K: Hash + Eq + Clone, V: Clone> TimedLru<K, V> {
 
         (old, cached)
     }
-
-    pub(crate) fn flush(&self) {
-        let now = Instant::now();
-        let mut cache = self.cache.lock();
-
-        // Collect keys of expired entries first
-        let expired_keys: Vec<_> = cache
-            .iter()
-            .filter_map(|(key, entry)| {
-                if entry.expires_at <= now {
-                    Some(key.clone())
-                } else {
-                    None
-                }
-            })
-            .collect();
-
-        // Remove expired entries
-        for key in expired_keys {
-            cache.remove(&key);
-        }
-    }
 }
 
 impl<K: Hash + Eq, V: Clone> TimedLru<K, V> {

proxy/src/config.rs

@@ -22,8 +22,6 @@ use crate::rate_limiter::{RateBucketInfo, RateLimitAlgorithm, RateLimiterConfig}
 use crate::scram::threadpool::ThreadPool;
 use crate::serverless::GlobalConnPoolOptions;
 use crate::serverless::cancel_set::CancelSet;
-#[cfg(feature = "rest_broker")]
-use crate::serverless::rest::DbSchemaCache;
 pub use crate::tls::server_config::{TlsConfig, configure_tls};
 use crate::types::{Host, RoleName};
 
@@ -32,8 +30,6 @@ pub struct ProxyConfig {
     pub metric_collection: Option<MetricCollectionConfig>,
     pub http_config: HttpConfig,
     pub authentication_config: AuthenticationConfig,
-    #[cfg(feature = "rest_broker")]
-    pub rest_config: RestConfig,
     pub proxy_protocol_v2: ProxyProtocolV2,
     pub handshake_timeout: Duration,
     pub wake_compute_retry_config: RetryConfig,
@@ -84,12 +80,6 @@ pub struct AuthenticationConfig {
     pub console_redirect_confirmation_timeout: tokio::time::Duration,
 }
 
-#[cfg(feature = "rest_broker")]
-pub struct RestConfig {
-    pub is_rest_broker: bool,
-    pub db_schema_cache: Option<DbSchemaCache>,
-}
-
 #[derive(Debug)]
 pub struct EndpointCacheConfig {
     /// Batch size to receive all endpoints on the startup.

proxy/src/control_plane/client/mod.rs

@@ -8,7 +8,7 @@ use std::time::Duration;
 
 use clashmap::ClashMap;
 use tokio::time::Instant;
-use tracing::debug;
+use tracing::{debug, info};
 
 use super::{EndpointAccessControl, RoleAccessControl};
 use crate::auth::backend::ComputeUserInfo;
@@ -234,8 +234,7 @@ impl<K: Hash + Eq + Clone> ApiLocks<K> {
                 // temporary lock a single shard and then clear any semaphores that aren't currently checked out
                 // race conditions: if strong_count == 1, there's no way that it can increase while the shard is locked
                 // therefore releasing it is safe from race conditions
-                debug!(
-                    //FIXME: is anything depending on this being info?
+                info!(
                     name = self.name,
                     shard = i,
                     "performing epoch reclamation on api lock"

proxy/src/serverless/mod.rs

@@ -11,8 +11,6 @@ mod http_conn_pool;
 mod http_util;
 mod json;
 mod local_conn_pool;
-#[cfg(feature = "rest_broker")]
-pub mod rest;
 mod sql_over_http;
 mod websocket;
 
@@ -489,37 +487,6 @@ async fn request_handler(
             .body(Empty::new().map_err(|x| match x {}).boxed())
             .map_err(|e| ApiError::InternalServerError(e.into()))
     } else {
-        #[cfg(feature = "rest_broker")]
-        {
-            if config.rest_config.is_rest_broker && {
-                let path_parts: Vec<&str> = request.uri().path().split('/').collect();
-                path_parts.len() >= 3 && path_parts[2].starts_with("rest")
-            } {
-                let ctx =
-                    RequestContext::new(session_id, conn_info, crate::metrics::Protocol::Http);
-                let span = ctx.span();
-
-                let testodrome_id = request
-                    .headers()
-                    .get("X-Neon-Query-ID")
-                    .and_then(|value| value.to_str().ok())
-                    .map(|s| s.to_string());
-
-                if let Some(query_id) = testodrome_id {
-                    info!(parent: &ctx.span(), "testodrome query ID: {query_id}");
-                    ctx.set_testodrome_id(query_id.into());
-                }
-
-                rest::handle(config, ctx, request, backend, http_cancellation_token)
-                    .instrument(span)
-                    .await
-            } else {
-                json_response(StatusCode::BAD_REQUEST, "query is not supported")
-            }
-        }
-        #[cfg(not(feature = "rest_broker"))]
-        {
-            json_response(StatusCode::BAD_REQUEST, "query is not supported")
-        }
+        json_response(StatusCode::BAD_REQUEST, "query is not supported")
     }
 }

safekeeper/src/timeline.rs

@@ -197,7 +197,7 @@ impl StateSK {
         Ok(TimelineMembershipSwitchResponse {
             previous_conf: result.previous_conf,
             current_conf: result.current_conf,
-            term: self.state().acceptor_state.term,
+            last_log_term: self.state().acceptor_state.term,
             flush_lsn: self.flush_lsn(),
         })
     }

storage_controller/src/background_node_operations.rs

@@ -6,6 +6,11 @@ use utils::id::NodeId;
 
 pub(crate) const MAX_RECONCILES_PER_OPERATION: usize = 64;
 
+#[derive(Copy, Clone)]
+pub(crate) struct Delete {
+    pub(crate) node_id: NodeId,
+}
+
 #[derive(Copy, Clone)]
 pub(crate) struct Drain {
     pub(crate) node_id: NodeId,
@@ -18,6 +23,7 @@ pub(crate) struct Fill {
 
 #[derive(Copy, Clone)]
 pub(crate) enum Operation {
+    Delete(Delete),
     Drain(Drain),
     Fill(Fill),
 }
@@ -30,6 +36,8 @@ pub(crate) enum OperationError {
     FinalizeError(Cow<'static, str>),
     #[error("Operation cancelled")]
     Cancelled,
+    #[error("Impossible constraint error: {0}")]
+    ImpossibleConstraint(Cow<'static, str>),
 }
 
 pub(crate) struct OperationHandler {
@@ -38,6 +46,12 @@ pub(crate) struct OperationHandler {
     pub(crate) cancel: CancellationToken,
 }
 
+impl Display for Delete {
+    fn fmt(&self, f: &mut std::fmt::Formatter) -> std::fmt::Result {
+        write!(f, "delete {}", self.node_id)
+    }
+}
+
 impl Display for Drain {
     fn fmt(&self, f: &mut std::fmt::Formatter) -> std::fmt::Result {
         write!(f, "drain {}", self.node_id)
@@ -53,6 +67,7 @@ impl Display for Fill {
 impl Display for Operation {
     fn fmt(&self, f: &mut std::fmt::Formatter) -> std::fmt::Result {
         match self {
+            Operation::Delete(op) => write!(f, "{op}"),
             Operation::Drain(op) => write!(f, "{op}"),
             Operation::Fill(op) => write!(f, "{op}"),
         }

storage_controller/src/http.rs

@@ -919,7 +919,7 @@ async fn handle_node_drop(req: Request<Body>) -> Result<Response<Body>, ApiError
     json_response(StatusCode::OK, state.service.node_drop(node_id).await?)
 }
 
-async fn handle_node_delete(req: Request<Body>) -> Result<Response<Body>, ApiError> {
+async fn handle_node_delete_old(req: Request<Body>) -> Result<Response<Body>, ApiError> {
     check_permissions(&req, Scope::Admin)?;
 
     let req = match maybe_forward(req).await {
@@ -931,7 +931,10 @@ async fn handle_node_delete(req: Request<Body>) -> Result<Response<Body>, ApiErr
 
     let state = get_state(&req);
     let node_id: NodeId = parse_request_param(&req, "node_id")?;
-    json_response(StatusCode::OK, state.service.node_delete(node_id).await?)
+    json_response(
+        StatusCode::OK,
+        state.service.node_delete_old(node_id).await?,
+    )
 }
 
 async fn handle_tombstone_list(req: Request<Body>) -> Result<Response<Body>, ApiError> {
@@ -1051,6 +1054,42 @@ async fn handle_get_leader(req: Request<Body>) -> Result<Response<Body>, ApiErro
     json_response(StatusCode::OK, leader)
 }
 
+async fn handle_node_delete(req: Request<Body>) -> Result<Response<Body>, ApiError> {
+    check_permissions(&req, Scope::Admin)?;
+
+    let req = match maybe_forward(req).await {
+        ForwardOutcome::Forwarded(res) => {
+            return res;
+        }
+        ForwardOutcome::NotForwarded(req) => req,
+    };
+
+    let state = get_state(&req);
+    let node_id: NodeId = parse_request_param(&req, "node_id")?;
+    json_response(
+        StatusCode::OK,
+        state.service.start_node_delete(node_id).await?,
+    )
+}
+
+async fn handle_cancel_node_delete(req: Request<Body>) -> Result<Response<Body>, ApiError> {
+    check_permissions(&req, Scope::Infra)?;
+
+    let req = match maybe_forward(req).await {
+        ForwardOutcome::Forwarded(res) => {
+            return res;
+        }
+        ForwardOutcome::NotForwarded(req) => req,
+    };
+
+    let state = get_state(&req);
+    let node_id: NodeId = parse_request_param(&req, "node_id")?;
+    json_response(
+        StatusCode::ACCEPTED,
+        state.service.cancel_node_delete(node_id).await?,
+    )
+}
+
 async fn handle_node_drain(req: Request<Body>) -> Result<Response<Body>, ApiError> {
     check_permissions(&req, Scope::Infra)?;
 
@@ -2221,8 +2260,14 @@ pub fn make_router(
         .post("/control/v1/node", |r| {
             named_request_span(r, handle_node_register, RequestName("control_v1_node"))
         })
+        // This endpoint is deprecated and will be removed in a future version.
+        // Use PUT /control/v1/node/:node_id/delete instead.
         .delete("/control/v1/node/:node_id", |r| {
-            named_request_span(r, handle_node_delete, RequestName("control_v1_node_delete"))
+            named_request_span(
+                r,
+                handle_node_delete_old,
+                RequestName("control_v1_node_delete"),
+            )
         })
         .get("/control/v1/node", |r| {
             named_request_span(r, handle_node_list, RequestName("control_v1_node"))
@@ -2247,6 +2292,20 @@ pub fn make_router(
         .get("/control/v1/leader", |r| {
             named_request_span(r, handle_get_leader, RequestName("control_v1_get_leader"))
         })
+        .put("/control/v1/node/:node_id/delete", |r| {
+            named_request_span(
+                r,
+                handle_node_delete,
+                RequestName("control_v1_start_node_delete"),
+            )
+        })
+        .delete("/control/v1/node/:node_id/delete", |r| {
+            named_request_span(
+                r,
+                handle_cancel_node_delete,
+                RequestName("control_v1_cancel_node_delete"),
+            )
+        })
         .put("/control/v1/node/:node_id/drain", |r| {
             named_request_span(r, handle_node_drain, RequestName("control_v1_node_drain"))
         })
@@ -2312,7 +2371,7 @@ pub fn make_router(
             named_request_span(
                 r,
                 handle_safekeeper_scheduling_policy,
-                RequestName("v1_safekeeper_status"),
+                RequestName("v1_safekeeper_scheduling_policy"),
             )
         })
         // Tenant Shard operations

storage_controller/src/lib.rs

@@ -6,13 +6,13 @@ extern crate hyper0 as hyper;
 mod auth;
 mod background_node_operations;
 mod compute_hook;
-mod drain_utils;
 mod heartbeater;
 pub mod http;
 mod id_lock_map;
 mod leadership;
 pub mod metrics;
 mod node;
+mod operation_utils;
 mod pageserver_client;
 mod peer_client;
 pub mod persistence;

storage_controller/src/node.rs

@@ -201,6 +201,7 @@ impl Node {
 
         match self.scheduling {
             NodeSchedulingPolicy::Active => MaySchedule::Yes(utilization),
+            NodeSchedulingPolicy::Deleting => MaySchedule::No,
             NodeSchedulingPolicy::Draining => MaySchedule::No,
             NodeSchedulingPolicy::Filling => MaySchedule::Yes(utilization),
             NodeSchedulingPolicy::Pause => MaySchedule::No,

storage_controller/src/operation_utils.rs

@@ -10,63 +10,19 @@ use crate::node::Node;
 use crate::scheduler::Scheduler;
 use crate::tenant_shard::TenantShard;
 
-pub(crate) struct TenantShardIterator<F> {
-    tenants_accessor: F,
-    inspected_all_shards: bool,
-    last_inspected_shard: Option<TenantShardId>,
-}
-
-/// A simple iterator which can be used in tandem with [`crate::service::Service`]
-/// to iterate over all known tenant shard ids without holding the lock on the
-/// service state at all times.
-impl<F> TenantShardIterator<F>
-where
-    F: Fn(Option<TenantShardId>) -> Option<TenantShardId>,
-{
-    pub(crate) fn new(tenants_accessor: F) -> Self {
-        Self {
-            tenants_accessor,
-            inspected_all_shards: false,
-            last_inspected_shard: None,
-        }
-    }
-
-    /// Returns the next tenant shard id if one exists
-    pub(crate) fn next(&mut self) -> Option<TenantShardId> {
-        if self.inspected_all_shards {
-            return None;
-        }
-
-        match (self.tenants_accessor)(self.last_inspected_shard) {
-            Some(tid) => {
-                self.last_inspected_shard = Some(tid);
-                Some(tid)
-            }
-            None => {
-                self.inspected_all_shards = true;
-                None
-            }
-        }
-    }
-
-    /// Returns true when the end of the iterator is reached and false otherwise
-    pub(crate) fn finished(&self) -> bool {
-        self.inspected_all_shards
-    }
-}
-
 /// Check that the state of the node being drained is as expected:
-/// node is present in memory and scheduling policy is set to [`NodeSchedulingPolicy::Draining`]
+/// node is present in memory and scheduling policy is set to expected_policy
 pub(crate) fn validate_node_state(
     node_id: &NodeId,
     nodes: Arc<HashMap<NodeId, Node>>,
+    expected_policy: NodeSchedulingPolicy,
 ) -> Result<(), OperationError> {
     let node = nodes.get(node_id).ok_or(OperationError::NodeStateChanged(
         format!("node {node_id} was removed").into(),
     ))?;
 
     let current_policy = node.get_scheduling();
-    if !matches!(current_policy, NodeSchedulingPolicy::Draining) {
+    if current_policy != expected_policy {
         // TODO(vlad): maybe cancel pending reconciles before erroring out. need to think
         // about it
         return Err(OperationError::NodeStateChanged(
@@ -182,55 +138,3 @@ impl TenantShardDrain {
         }
     }
 }
-
-#[cfg(test)]
-mod tests {
-    use std::sync::Arc;
-
-    use utils::id::TenantId;
-    use utils::shard::{ShardCount, ShardNumber, TenantShardId};
-
-    use super::TenantShardIterator;
-
-    #[test]
-    fn test_tenant_shard_iterator() {
-        let tenant_id = TenantId::generate();
-        let shard_count = ShardCount(8);
-
-        let mut tenant_shards = Vec::default();
-        for i in 0..shard_count.0 {
-            tenant_shards.push((
-                TenantShardId {
-                    tenant_id,
-                    shard_number: ShardNumber(i),
-                    shard_count,
-                },
-                (),
-            ))
-        }
-
-        let tenant_shards = Arc::new(tenant_shards);
-
-        let mut tid_iter = TenantShardIterator::new({
-            let tenants = tenant_shards.clone();
-            move |last_inspected_shard: Option<TenantShardId>| {
-                let entry = match last_inspected_shard {
-                    Some(skip_past) => {
-                        let mut cursor = tenants.iter().skip_while(|(tid, _)| *tid != skip_past);
-                        cursor.nth(1)
-                    }
-                    None => tenants.first(),
-                };
-
-                entry.map(|(tid, _)| tid).copied()
-            }
-        });
-
-        let mut iterated_over = Vec::default();
-        while let Some(tid) = tid_iter.next() {
-            iterated_over.push((tid, ()));
-        }
-
-        assert_eq!(iterated_over, *tenant_shards);
-    }
-}

storage_controller/src/persistence.rs

@@ -635,18 +635,23 @@ impl Persistence {
         let updated = self
             .with_measured_conn(DatabaseOperation::ReAttach, move |conn| {
                 Box::pin(async move {
-                    // Check if the node is not marked as deleted
-                    let deleted_node: i64 = nodes
+                    let node: Option<NodePersistence> = nodes
                         .filter(node_id.eq(input_node_id.0 as i64))
-                        .filter(lifecycle.eq(String::from(NodeLifecycle::Deleted)))
-                        .count()
-                        .get_result(conn)
-                        .await?;
-                    if deleted_node > 0 {
-                        return Err(DatabaseError::Logical(format!(
-                            "Node {input_node_id} is marked as deleted, re-attach is not allowed"
-                        )));
-                    }
+                        .first::<NodePersistence>(conn)
+                        .await
+                        .optional()?;
+
+                    // Check if the node is not marked as deleted
+                    match node {
+                        Some(node) if matches!(NodeLifecycle::from_str(&node.lifecycle), Ok(NodeLifecycle::Deleted)) => {
+                            return Err(DatabaseError::Logical(format!(
+                                "Node {input_node_id} is marked as deleted, re-attach is not allowed"
+                            )));
+                        }
+                        _ => {
+                            // go through
+                        }
+                    };
 
                     let rows_updated = diesel::update(tenant_shards)
                         .filter(generation_pageserver.eq(input_node_id.0 as i64))
@@ -664,21 +669,23 @@ impl Persistence {
                         .load(conn)
                         .await?;
 
-                    // If the node went through a drain and restart phase before re-attaching,
-                    // then reset it's node scheduling policy to active.
-                    diesel::update(nodes)
-                        .filter(node_id.eq(input_node_id.0 as i64))
-                        .filter(
-                            scheduling_policy
-                                .eq(String::from(NodeSchedulingPolicy::PauseForRestart))
-                                .or(scheduling_policy
-                                    .eq(String::from(NodeSchedulingPolicy::Draining)))
-                                .or(scheduling_policy
-                                    .eq(String::from(NodeSchedulingPolicy::Filling))),
-                        )
-                        .set(scheduling_policy.eq(String::from(NodeSchedulingPolicy::Active)))
-                        .execute(conn)
-                        .await?;
+                    if let Some(node) = node {
+                        let old_scheduling_policy =
+                            NodeSchedulingPolicy::from_str(&node.scheduling_policy).unwrap();
+                        let new_scheduling_policy = match old_scheduling_policy {
+                            NodeSchedulingPolicy::Active => NodeSchedulingPolicy::Active,
+                            NodeSchedulingPolicy::PauseForRestart => NodeSchedulingPolicy::Active,
+                            NodeSchedulingPolicy::Draining => NodeSchedulingPolicy::Active,
+                            NodeSchedulingPolicy::Filling => NodeSchedulingPolicy::Active,
+                            NodeSchedulingPolicy::Pause => NodeSchedulingPolicy::Pause,
+                            NodeSchedulingPolicy::Deleting => NodeSchedulingPolicy::Pause,
+                        };
+                        diesel::update(nodes)
+                            .filter(node_id.eq(input_node_id.0 as i64))
+                            .set(scheduling_policy.eq(String::from(new_scheduling_policy)))
+                            .execute(conn)
+                            .await?;
+                    }
 
                     Ok(updated)
                 })

storage_controller/src/service.rs

@@ -1,8 +1,8 @@
 pub mod chaos_injector;
-mod context_iterator;
 pub mod feature_flag;
 pub(crate) mod safekeeper_reconciler;
 mod safekeeper_service;
+mod tenant_shard_iterator;
 
 use std::borrow::Cow;
 use std::cmp::Ordering;
@@ -16,7 +16,6 @@ use std::sync::{Arc, OnceLock};
 use std::time::{Duration, Instant, SystemTime};
 
 use anyhow::Context;
-use context_iterator::TenantShardContextIterator;
 use control_plane::storage_controller::{
     AttachHookRequest, AttachHookResponse, InspectRequest, InspectResponse,
 };
@@ -55,6 +54,7 @@ use pageserver_client::{BlockUnblock, mgmt_api};
 use reqwest::{Certificate, StatusCode};
 use safekeeper_api::models::SafekeeperUtilization;
 use safekeeper_reconciler::SafekeeperReconcilers;
+use tenant_shard_iterator::{TenantShardExclusiveIterator, create_shared_shard_iterator};
 use tokio::sync::TryAcquireError;
 use tokio::sync::mpsc::error::TrySendError;
 use tokio_util::sync::CancellationToken;
@@ -68,10 +68,9 @@ use utils::sync::gate::{Gate, GateGuard};
 use utils::{failpoint_support, pausable_failpoint};
 
 use crate::background_node_operations::{
-    Drain, Fill, MAX_RECONCILES_PER_OPERATION, Operation, OperationError, OperationHandler,
+    Delete, Drain, Fill, MAX_RECONCILES_PER_OPERATION, Operation, OperationError, OperationHandler,
 };
 use crate::compute_hook::{self, ComputeHook, NotifyError};
-use crate::drain_utils::{self, TenantShardDrain, TenantShardIterator};
 use crate::heartbeater::{Heartbeater, PageserverState, SafekeeperState};
 use crate::id_lock_map::{
     IdLockMap, TracingExclusiveGuard, trace_exclusive_lock, trace_shared_lock,
@@ -79,6 +78,7 @@ use crate::id_lock_map::{
 use crate::leadership::Leadership;
 use crate::metrics;
 use crate::node::{AvailabilityTransition, Node};
+use crate::operation_utils::{self, TenantShardDrain};
 use crate::pageserver_client::PageserverClient;
 use crate::peer_client::GlobalObservedState;
 use crate::persistence::split_state::SplitState;
@@ -105,7 +105,7 @@ use crate::timeline_import::{
     TimelineImportFinalizeError, TimelineImportState, UpcallClient,
 };
 
-const WAITER_FILL_DRAIN_POLL_TIMEOUT: Duration = Duration::from_millis(500);
+const WAITER_OPERATION_POLL_TIMEOUT: Duration = Duration::from_millis(500);
 
 // For operations that should be quick, like attaching a new tenant
 const SHORT_RECONCILE_TIMEOUT: Duration = Duration::from_secs(5);
@@ -581,7 +581,9 @@ impl From<ReconcileWaitError> for ApiError {
 impl From<OperationError> for ApiError {
     fn from(value: OperationError) -> Self {
         match value {
-            OperationError::NodeStateChanged(err) | OperationError::FinalizeError(err) => {
+            OperationError::NodeStateChanged(err)
+            | OperationError::FinalizeError(err)
+            | OperationError::ImpossibleConstraint(err) => {
                 ApiError::InternalServerError(anyhow::anyhow!(err))
             }
             OperationError::Cancelled => ApiError::Conflict("Operation was cancelled".into()),
@@ -2414,6 +2416,7 @@ impl Service {
                 NodeSchedulingPolicy::PauseForRestart
                     | NodeSchedulingPolicy::Draining
                     | NodeSchedulingPolicy::Filling
+                    | NodeSchedulingPolicy::Deleting
             );
 
             let mut new_nodes = (**nodes).clone();
@@ -7055,7 +7058,7 @@ impl Service {
     /// If a node has any work on it, it will be rescheduled: this is "clean" in the sense
     /// that we don't leave any bad state behind in the storage controller, but unclean
     /// in the sense that we are not carefully draining the node.
-    pub(crate) async fn node_delete(&self, node_id: NodeId) -> Result<(), ApiError> {
+    pub(crate) async fn node_delete_old(&self, node_id: NodeId) -> Result<(), ApiError> {
         let _node_lock =
             trace_exclusive_lock(&self.node_op_locks, node_id, NodeOperations::Delete).await;
 
@@ -7089,7 +7092,7 @@ impl Service {
             }
 
             for (_tenant_id, mut schedule_context, shards) in
-                TenantShardContextIterator::new(tenants, ScheduleMode::Normal)
+                TenantShardExclusiveIterator::new(tenants, ScheduleMode::Normal)
             {
                 for shard in shards {
                     if shard.deref_node(node_id) {
@@ -7158,6 +7161,171 @@ impl Service {
         Ok(())
     }
 
+    pub(crate) async fn delete_node(
+        self: &Arc<Self>,
+        node_id: NodeId,
+        policy_on_start: NodeSchedulingPolicy,
+        cancel: CancellationToken,
+    ) -> Result<(), OperationError> {
+        let reconciler_config = ReconcilerConfigBuilder::new(ReconcilerPriority::Normal).build();
+
+        let mut waiters: Vec<ReconcilerWaiter> = Vec::new();
+        let mut tid_iter = create_shared_shard_iterator(self.clone());
+
+        while !tid_iter.finished() {
+            if cancel.is_cancelled() {
+                match self
+                    .node_configure(node_id, None, Some(policy_on_start))
+                    .await
+                {
+                    Ok(()) => return Err(OperationError::Cancelled),
+                    Err(err) => {
+                        return Err(OperationError::FinalizeError(
+                            format!(
+                                "Failed to finalise delete cancel of {} by setting scheduling policy to {}: {}",
+                                node_id, String::from(policy_on_start), err
+                            )
+                            .into(),
+                        ));
+                    }
+                }
+            }
+
+            operation_utils::validate_node_state(
+                &node_id,
+                self.inner.read().unwrap().nodes.clone(),
+                NodeSchedulingPolicy::Deleting,
+            )?;
+
+            while waiters.len() < MAX_RECONCILES_PER_OPERATION {
+                let tid = match tid_iter.next() {
+                    Some(tid) => tid,
+                    None => {
+                        break;
+                    }
+                };
+
+                let mut locked = self.inner.write().unwrap();
+                let (nodes, tenants, scheduler) = locked.parts_mut();
+
+                let tenant_shard = match tenants.get_mut(&tid) {
+                    Some(tenant_shard) => tenant_shard,
+                    None => {
+                        // Tenant shard was deleted by another operation. Skip it.
+                        continue;
+                    }
+                };
+
+                match tenant_shard.get_scheduling_policy() {
+                    ShardSchedulingPolicy::Active | ShardSchedulingPolicy::Essential => {
+                        // A migration during delete is classed as 'essential' because it is required to
+                        // uphold our availability goals for the tenant: this shard is elegible for migration.
+                    }
+                    ShardSchedulingPolicy::Pause | ShardSchedulingPolicy::Stop => {
+                        // If we have been asked to avoid rescheduling this shard, then do not migrate it during a deletion
+                        tracing::warn!(
+                            tenant_id=%tid.tenant_id, shard_id=%tid.shard_slug(),
+                            "Skip migration during deletion because shard scheduling policy {:?} disallows it",
+                            tenant_shard.get_scheduling_policy(),
+                        );
+                        continue;
+                    }
+                }
+
+                if tenant_shard.deref_node(node_id) {
+                    // TODO(ephemeralsad): we should process all shards in a tenant at once, so
+                    // we can avoid settling the tenant unevenly.
+                    let mut schedule_context = ScheduleContext::new(ScheduleMode::Normal);
+                    if let Err(e) = tenant_shard.schedule(scheduler, &mut schedule_context) {
+                        tracing::error!(
+                            "Refusing to delete node, shard {} can't be rescheduled: {e}",
+                            tenant_shard.tenant_shard_id
+                        );
+                        return Err(OperationError::ImpossibleConstraint(e.to_string().into()));
+                    } else {
+                        tracing::info!(
+                            "Rescheduled shard {} away from node during deletion",
+                            tenant_shard.tenant_shard_id
+                        )
+                    }
+
+                    let waiter = self.maybe_configured_reconcile_shard(
+                        tenant_shard,
+                        nodes,
+                        reconciler_config,
+                    );
+                    if let Some(some) = waiter {
+                        waiters.push(some);
+                    }
+                }
+            }
+
+            waiters = self
+                .await_waiters_remainder(waiters, WAITER_OPERATION_POLL_TIMEOUT)
+                .await;
+
+            failpoint_support::sleep_millis_async!("sleepy-delete-loop", &cancel);
+        }
+
+        while !waiters.is_empty() {
+            if cancel.is_cancelled() {
+                match self
+                    .node_configure(node_id, None, Some(policy_on_start))
+                    .await
+                {
+                    Ok(()) => return Err(OperationError::Cancelled),
+                    Err(err) => {
+                        return Err(OperationError::FinalizeError(
+                            format!(
+                                "Failed to finalise drain cancel of {} by setting scheduling policy to {}: {}",
+                                node_id, String::from(policy_on_start), err
+                            )
+                            .into(),
+                        ));
+                    }
+                }
+            }
+
+            tracing::info!("Awaiting {} pending delete reconciliations", waiters.len());
+
+            waiters = self
+                .await_waiters_remainder(waiters, SHORT_RECONCILE_TIMEOUT)
+                .await;
+        }
+
+        self.persistence
+            .set_tombstone(node_id)
+            .await
+            .map_err(|e| OperationError::FinalizeError(e.to_string().into()))?;
+
+        {
+            let mut locked = self.inner.write().unwrap();
+            let (nodes, _, scheduler) = locked.parts_mut();
+
+            scheduler.node_remove(node_id);
+
+            let mut nodes_mut = (**nodes).clone();
+            if let Some(mut removed_node) = nodes_mut.remove(&node_id) {
+                // Ensure that any reconciler holding an Arc<> to this node will
+                // drop out when trying to RPC to it (setting Offline state sets the
+                // cancellation token on the Node object).
+                removed_node.set_availability(NodeAvailability::Offline);
+            }
+            *nodes = Arc::new(nodes_mut);
+
+            metrics::METRICS_REGISTRY
+                .metrics_group
+                .storage_controller_pageserver_nodes
+                .set(nodes.len() as i64);
+            metrics::METRICS_REGISTRY
+                .metrics_group
+                .storage_controller_https_pageserver_nodes
+                .set(nodes.values().filter(|n| n.has_https_port()).count() as i64);
+        }
+
+        Ok(())
+    }
+
     pub(crate) async fn node_list(&self) -> Result<Vec<Node>, ApiError> {
         let nodes = {
             self.inner
@@ -7546,7 +7714,7 @@ impl Service {
                 let mut tenants_affected: usize = 0;
 
                 for (_tenant_id, mut schedule_context, shards) in
-                    TenantShardContextIterator::new(tenants, ScheduleMode::Normal)
+                    TenantShardExclusiveIterator::new(tenants, ScheduleMode::Normal)
                 {
                     for tenant_shard in shards {
                         let tenant_shard_id = tenant_shard.tenant_shard_id;
@@ -7717,6 +7885,142 @@ impl Service {
         self.node_configure(node_id, availability, scheduling).await
     }
 
+    pub(crate) async fn start_node_delete(
+        self: &Arc<Self>,
+        node_id: NodeId,
+    ) -> Result<(), ApiError> {
+        let (ongoing_op, node_policy, schedulable_nodes_count) = {
+            let locked = self.inner.read().unwrap();
+            let nodes = &locked.nodes;
+            let node = nodes.get(&node_id).ok_or(ApiError::NotFound(
+                anyhow::anyhow!("Node {} not registered", node_id).into(),
+            ))?;
+            let schedulable_nodes_count = nodes
+                .iter()
+                .filter(|(_, n)| matches!(n.may_schedule(), MaySchedule::Yes(_)))
+                .count();
+
+            (
+                locked
+                    .ongoing_operation
+                    .as_ref()
+                    .map(|ongoing| ongoing.operation),
+                node.get_scheduling(),
+                schedulable_nodes_count,
+            )
+        };
+
+        if let Some(ongoing) = ongoing_op {
+            return Err(ApiError::PreconditionFailed(
+                format!("Background operation already ongoing for node: {ongoing}").into(),
+            ));
+        }
+
+        if schedulable_nodes_count == 0 {
+            return Err(ApiError::PreconditionFailed(
+                "No other schedulable nodes to move shards".into(),
+            ));
+        }
+
+        match node_policy {
+            NodeSchedulingPolicy::Active | NodeSchedulingPolicy::Pause => {
+                self.node_configure(node_id, None, Some(NodeSchedulingPolicy::Deleting))
+                    .await?;
+
+                let cancel = self.cancel.child_token();
+                let gate_guard = self.gate.enter().map_err(|_| ApiError::ShuttingDown)?;
+                let policy_on_start = node_policy;
+
+                self.inner.write().unwrap().ongoing_operation = Some(OperationHandler {
+                    operation: Operation::Delete(Delete { node_id }),
+                    cancel: cancel.clone(),
+                });
+
+                let span = tracing::info_span!(parent: None, "delete_node", %node_id);
+
+                tokio::task::spawn(
+                    {
+                        let service = self.clone();
+                        let cancel = cancel.clone();
+                        async move {
+                            let _gate_guard = gate_guard;
+
+                            scopeguard::defer! {
+                                let prev = service.inner.write().unwrap().ongoing_operation.take();
+
+                                if let Some(Operation::Delete(removed_delete)) = prev.map(|h| h.operation) {
+                                    assert_eq!(removed_delete.node_id, node_id, "We always take the same operation");
+                                } else {
+                                    panic!("We always remove the same operation")
+                                }
+                            }
+
+                            tracing::info!("Delete background operation starting");
+                            let res = service
+                                .delete_node(node_id, policy_on_start, cancel)
+                                .await;
+                            match res {
+                                Ok(()) => {
+                                    tracing::info!(
+                                        "Delete background operation completed successfully"
+                                    );
+                                }
+                                Err(OperationError::Cancelled) => {
+                                    tracing::info!("Delete background operation was cancelled");
+                                }
+                                Err(err) => {
+                                    tracing::error!(
+                                        "Delete background operation encountered: {err}"
+                                    )
+                                }
+                            }
+                        }
+                    }
+                    .instrument(span),
+                );
+            }
+            NodeSchedulingPolicy::Deleting => {
+                return Err(ApiError::Conflict(format!(
+                    "Node {node_id} has delete in progress"
+                )));
+            }
+            policy => {
+                return Err(ApiError::PreconditionFailed(
+                    format!("Node {node_id} cannot be deleted due to {policy:?} policy").into(),
+                ));
+            }
+        }
+
+        Ok(())
+    }
+
+    pub(crate) async fn cancel_node_delete(
+        self: &Arc<Self>,
+        node_id: NodeId,
+    ) -> Result<(), ApiError> {
+        {
+            let locked = self.inner.read().unwrap();
+            let nodes = &locked.nodes;
+            nodes.get(&node_id).ok_or(ApiError::NotFound(
+                anyhow::anyhow!("Node {} not registered", node_id).into(),
+            ))?;
+        }
+
+        if let Some(op_handler) = self.inner.read().unwrap().ongoing_operation.as_ref() {
+            if let Operation::Delete(delete) = op_handler.operation {
+                if delete.node_id == node_id {
+                    tracing::info!("Cancelling background delete operation for node {node_id}");
+                    op_handler.cancel.cancel();
+                    return Ok(());
+                }
+            }
+        }
+
+        Err(ApiError::PreconditionFailed(
+            format!("Node {node_id} has no delete in progress").into(),
+        ))
+    }
+
     pub(crate) async fn start_node_drain(
         self: &Arc<Self>,
         node_id: NodeId,
@@ -8293,7 +8597,7 @@ impl Service {
         // to ignore the utilisation component of the score.
 
         for (_tenant_id, schedule_context, shards) in
-            TenantShardContextIterator::new(tenants, ScheduleMode::Speculative)
+            TenantShardExclusiveIterator::new(tenants, ScheduleMode::Speculative)
         {
             for shard in shards {
                 if work.len() >= MAX_OPTIMIZATIONS_PLAN_PER_PASS {
@@ -9020,25 +9324,7 @@ impl Service {
 
         let mut waiters = Vec::new();
 
-        let mut tid_iter = TenantShardIterator::new({
-            let service = self.clone();
-            move |last_inspected_shard: Option<TenantShardId>| {
-                let locked = &service.inner.read().unwrap();
-                let tenants = &locked.tenants;
-                let entry = match last_inspected_shard {
-                    Some(skip_past) => {
-                        // Skip to the last seen tenant shard id
-                        let mut cursor = tenants.iter().skip_while(|(tid, _)| **tid != skip_past);
-
-                        // Skip past the last seen
-                        cursor.nth(1)
-                    }
-                    None => tenants.first_key_value(),
-                };
-
-                entry.map(|(tid, _)| tid).copied()
-            }
-        });
+        let mut tid_iter = create_shared_shard_iterator(self.clone());
 
         while !tid_iter.finished() {
             if cancel.is_cancelled() {
@@ -9058,7 +9344,11 @@ impl Service {
                 }
             }
 
-            drain_utils::validate_node_state(&node_id, self.inner.read().unwrap().nodes.clone())?;
+            operation_utils::validate_node_state(
+                &node_id,
+                self.inner.read().unwrap().nodes.clone(),
+                NodeSchedulingPolicy::Draining,
+            )?;
 
             while waiters.len() < MAX_RECONCILES_PER_OPERATION {
                 let tid = match tid_iter.next() {
@@ -9138,7 +9428,7 @@ impl Service {
             }
 
             waiters = self
-                .await_waiters_remainder(waiters, WAITER_FILL_DRAIN_POLL_TIMEOUT)
+                .await_waiters_remainder(waiters, WAITER_OPERATION_POLL_TIMEOUT)
                 .await;
 
             failpoint_support::sleep_millis_async!("sleepy-drain-loop", &cancel);
@@ -9432,7 +9722,7 @@ impl Service {
             }
 
             waiters = self
-                .await_waiters_remainder(waiters, WAITER_FILL_DRAIN_POLL_TIMEOUT)
+                .await_waiters_remainder(waiters, WAITER_OPERATION_POLL_TIMEOUT)
                 .await;
         }
 

storage_controller/src/service/safekeeper_service.rs

@@ -914,13 +914,13 @@ impl Service {
                         // so it isn't counted toward the quorum.
                         if let Some(min_position) = min_position {
                             if let Ok(ok_res) = &res {
-                                if (ok_res.term, ok_res.flush_lsn) < min_position {
+                                if (ok_res.last_log_term, ok_res.flush_lsn) < min_position {
                                     // Use Error::Timeout to make this error retriable.
                                     res = Err(mgmt_api::Error::Timeout(
                                         format!(
                                         "safekeeper {} returned position {:?} which is less than minimum required position {:?}",
                                         client.node_id_label(),
-                                        (ok_res.term, ok_res.flush_lsn),
+                                        (ok_res.last_log_term, ok_res.flush_lsn),
                                         min_position
                                         )
                                     ));
@@ -1216,7 +1216,7 @@ impl Service {
 
         let mut sync_position = (INITIAL_TERM, Lsn::INVALID);
         for res in results.into_iter().flatten() {
-            let sk_position = (res.term, res.flush_lsn);
+            let sk_position = (res.last_log_term, res.flush_lsn);
             if sync_position < sk_position {
                 sync_position = sk_position;
             }

storage_controller/src/service/tenant_shard_iterator.rs

@@ -1,4 +1,5 @@
 use std::collections::BTreeMap;
+use std::sync::Arc;
 
 use utils::id::TenantId;
 use utils::shard::TenantShardId;
@@ -6,16 +7,21 @@ use utils::shard::TenantShardId;
 use crate::scheduler::{ScheduleContext, ScheduleMode};
 use crate::tenant_shard::TenantShard;
 
+use super::Service;
+
+/// Exclusive iterator over all tenant shards.
+/// It is used to iterate over consistent tenants state at specific point in time.
+///
 /// When making scheduling decisions, it is useful to have the ScheduleContext for a whole
 /// tenant while considering the individual shards within it.  This iterator is a helper
 /// that gathers all the shards in a tenant and then yields them together with a ScheduleContext
 /// for the tenant.
-pub(super) struct TenantShardContextIterator<'a> {
+pub(super) struct TenantShardExclusiveIterator<'a> {
     schedule_mode: ScheduleMode,
     inner: std::collections::btree_map::IterMut<'a, TenantShardId, TenantShard>,
 }
 
-impl<'a> TenantShardContextIterator<'a> {
+impl<'a> TenantShardExclusiveIterator<'a> {
     pub(super) fn new(
         tenants: &'a mut BTreeMap<TenantShardId, TenantShard>,
         schedule_mode: ScheduleMode,
@@ -27,7 +33,7 @@ impl<'a> TenantShardContextIterator<'a> {
     }
 }
 
-impl<'a> Iterator for TenantShardContextIterator<'a> {
+impl<'a> Iterator for TenantShardExclusiveIterator<'a> {
     type Item = (TenantId, ScheduleContext, Vec<&'a mut TenantShard>);
 
     fn next(&mut self) -> Option<Self::Item> {
@@ -52,13 +58,93 @@ impl<'a> Iterator for TenantShardContextIterator<'a> {
     }
 }
 
+/// Shared iterator over all tenant shards.
+/// It is used to iterate over all tenants without blocking another code, working with tenants
+///
+/// A simple iterator which can be used in tandem with [`crate::service::Service`]
+/// to iterate over all known tenant shard ids without holding the lock on the
+/// service state at all times.
+pub(crate) struct TenantShardSharedIterator<F> {
+    tenants_accessor: F,
+    inspected_all_shards: bool,
+    last_inspected_shard: Option<TenantShardId>,
+}
+
+impl<F> TenantShardSharedIterator<F>
+where
+    F: Fn(Option<TenantShardId>) -> Option<TenantShardId>,
+{
+    pub(crate) fn new(tenants_accessor: F) -> Self {
+        Self {
+            tenants_accessor,
+            inspected_all_shards: false,
+            last_inspected_shard: None,
+        }
+    }
+
+    pub(crate) fn finished(&self) -> bool {
+        self.inspected_all_shards
+    }
+}
+
+impl<F> Iterator for TenantShardSharedIterator<F>
+where
+    F: Fn(Option<TenantShardId>) -> Option<TenantShardId>,
+{
+    // TODO(ephemeralsad): consider adding schedule context to the iterator
+    type Item = TenantShardId;
+
+    /// Returns the next tenant shard id if one exists
+    fn next(&mut self) -> Option<Self::Item> {
+        if self.inspected_all_shards {
+            return None;
+        }
+
+        match (self.tenants_accessor)(self.last_inspected_shard) {
+            Some(tid) => {
+                self.last_inspected_shard = Some(tid);
+                Some(tid)
+            }
+            None => {
+                self.inspected_all_shards = true;
+                None
+            }
+        }
+    }
+}
+
+pub(crate) fn create_shared_shard_iterator(
+    service: Arc<Service>,
+) -> TenantShardSharedIterator<impl Fn(Option<TenantShardId>) -> Option<TenantShardId>> {
+    let tenants_accessor = move |last_inspected_shard: Option<TenantShardId>| {
+        let locked = &service.inner.read().unwrap();
+        let tenants = &locked.tenants;
+        let entry = match last_inspected_shard {
+            Some(skip_past) => {
+                // Skip to the last seen tenant shard id
+                let mut cursor = tenants.iter().skip_while(|(tid, _)| **tid != skip_past);
+
+                // Skip past the last seen
+                cursor.nth(1)
+            }
+            None => tenants.first_key_value(),
+        };
+
+        entry.map(|(tid, _)| tid).copied()
+    };
+
+    TenantShardSharedIterator::new(tenants_accessor)
+}
+
 #[cfg(test)]
 mod tests {
     use std::collections::BTreeMap;
     use std::str::FromStr;
+    use std::sync::Arc;
 
     use pageserver_api::controller_api::PlacementPolicy;
-    use utils::shard::{ShardCount, ShardNumber};
+    use utils::id::TenantId;
+    use utils::shard::{ShardCount, ShardNumber, TenantShardId};
 
     use super::*;
     use crate::scheduler::test_utils::make_test_nodes;
@@ -66,7 +152,7 @@ mod tests {
     use crate::tenant_shard::tests::make_test_tenant_with_id;
 
     #[test]
-    fn test_context_iterator() {
+    fn test_exclusive_shard_iterator() {
         // Hand-crafted tenant IDs to ensure they appear in the expected order when put into
         // a btreemap & iterated
         let mut t_1_shards = make_test_tenant_with_id(
@@ -106,7 +192,7 @@ mod tests {
             shard.schedule(&mut scheduler, &mut context).unwrap();
         }
 
-        let mut iter = TenantShardContextIterator::new(&mut tenants, ScheduleMode::Speculative);
+        let mut iter = TenantShardExclusiveIterator::new(&mut tenants, ScheduleMode::Speculative);
         let (tenant_id, context, shards) = iter.next().unwrap();
         assert_eq!(tenant_id, t1_id);
         assert_eq!(shards[0].tenant_shard_id.shard_number, ShardNumber(0));
@@ -132,4 +218,46 @@ mod tests {
             shard.intent.clear(&mut scheduler);
         }
     }
+
+    #[test]
+    fn test_shared_shard_iterator() {
+        let tenant_id = TenantId::generate();
+        let shard_count = ShardCount(8);
+
+        let mut tenant_shards = Vec::default();
+        for i in 0..shard_count.0 {
+            tenant_shards.push((
+                TenantShardId {
+                    tenant_id,
+                    shard_number: ShardNumber(i),
+                    shard_count,
+                },
+                (),
+            ))
+        }
+
+        let tenant_shards = Arc::new(tenant_shards);
+
+        let tid_iter = TenantShardSharedIterator::new({
+            let tenants = tenant_shards.clone();
+            move |last_inspected_shard: Option<TenantShardId>| {
+                let entry = match last_inspected_shard {
+                    Some(skip_past) => {
+                        let mut cursor = tenants.iter().skip_while(|(tid, _)| *tid != skip_past);
+                        cursor.nth(1)
+                    }
+                    None => tenants.first(),
+                };
+
+                entry.map(|(tid, _)| tid).copied()
+            }
+        });
+
+        let mut iterated_over = Vec::default();
+        for tid in tid_iter {
+            iterated_over.push((tid, ()));
+        }
+
+        assert_eq!(iterated_over, *tenant_shards);
+    }
 }

test_runner/fixtures/endpoint/http.py

@@ -57,6 +57,8 @@ class EndpointHttpClient(requests.Session):
         self.auth = BearerAuth(jwt)
 
         self.mount("http://", HTTPAdapter())
+        self.prewarm_url = f"http://localhost:{external_port}/lfc/prewarm"
+        self.offload_url = f"http://localhost:{external_port}/lfc/offload"
 
     def dbs_and_roles(self):
         res = self.get(f"http://localhost:{self.external_port}/dbs_and_roles", auth=self.auth)
@@ -64,33 +66,39 @@ class EndpointHttpClient(requests.Session):
         return res.json()
 
     def prewarm_lfc_status(self) -> dict[str, str]:
-        res = self.get(f"http://localhost:{self.external_port}/lfc/prewarm")
+        res = self.get(self.prewarm_url)
         res.raise_for_status()
         json: dict[str, str] = res.json()
         return json
 
     def prewarm_lfc(self, from_endpoint_id: str | None = None):
-        url: str = f"http://localhost:{self.external_port}/lfc/prewarm"
         params = {"from_endpoint": from_endpoint_id} if from_endpoint_id else dict()
-        self.post(url, params=params).raise_for_status()
+        self.post(self.prewarm_url, params=params).raise_for_status()
+        self.prewarm_lfc_wait()
 
+    def prewarm_lfc_wait(self):
         def prewarmed():
             json = self.prewarm_lfc_status()
             status, err = json["status"], json.get("error")
-            assert status == "completed", f"{status}, error {err}"
+            assert status == "completed", f"{status}, {err=}"
 
         wait_until(prewarmed, timeout=60)
 
-    def offload_lfc(self):
-        url = f"http://localhost:{self.external_port}/lfc/offload"
-        self.post(url).raise_for_status()
+    def offload_lfc_status(self) -> dict[str, str]:
+        res = self.get(self.offload_url)
+        res.raise_for_status()
+        json: dict[str, str] = res.json()
+        return json
 
+    def offload_lfc(self):
+        self.post(self.offload_url).raise_for_status()
+        self.offload_lfc_wait()
+
+    def offload_lfc_wait(self):
         def offloaded():
-            res = self.get(url)
-            res.raise_for_status()
-            json = res.json()
+            json = self.offload_lfc_status()
             status, err = json["status"], json.get("error")
-            assert status == "completed", f"{status}, error {err}"
+            assert status == "completed", f"{status}, {err=}"
 
         wait_until(offloaded)
 

test_runner/fixtures/neon_cli.py

@@ -568,6 +568,8 @@ class NeonLocalCli(AbstractNeonCli):
         timeout: str | None = None,
         env: dict[str, str] | None = None,
         dev: bool = False,
+        autoprewarm: bool = False,
+        offload_lfc_interval_seconds: int | None = None,
     ) -> subprocess.CompletedProcess[str]:
         args = [
             "endpoint",
@@ -593,6 +595,10 @@ class NeonLocalCli(AbstractNeonCli):
             args.extend(["--create-test-user"])
         if timeout is not None:
             args.extend(["--start-timeout", str(timeout)])
+        if autoprewarm:
+            args.extend(["--autoprewarm"])
+        if offload_lfc_interval_seconds is not None:
+            args.extend(["--offload-lfc-interval-seconds", str(offload_lfc_interval_seconds)])
         if dev:
             args.extend(["--dev"])
 

test_runner/fixtures/neon_fixtures.py

@@ -1867,6 +1867,7 @@ class PageserverSchedulingPolicy(StrEnum):
     FILLING = "Filling"
     PAUSE = "Pause"
     PAUSE_FOR_RESTART = "PauseForRestart"
+    DELETING = "Deleting"
 
 
 class StorageControllerLeadershipStatus(StrEnum):
@@ -2075,14 +2076,30 @@ class NeonStorageController(MetricsGetter, LogUtils):
             headers=self.headers(TokenScope.ADMIN),
         )
 
-    def node_delete(self, node_id):
-        log.info(f"node_delete({node_id})")
+    def node_delete_old(self, node_id):
+        log.info(f"node_delete_old({node_id})")
         self.request(
             "DELETE",
             f"{self.api}/control/v1/node/{node_id}",
             headers=self.headers(TokenScope.ADMIN),
         )
 
+    def node_delete(self, node_id):
+        log.info(f"node_delete({node_id})")
+        self.request(
+            "PUT",
+            f"{self.api}/control/v1/node/{node_id}/delete",
+            headers=self.headers(TokenScope.ADMIN),
+        )
+
+    def cancel_node_delete(self, node_id):
+        log.info(f"cancel_node_delete({node_id})")
+        self.request(
+            "DELETE",
+            f"{self.api}/control/v1/node/{node_id}/delete",
+            headers=self.headers(TokenScope.ADMIN),
+        )
+
     def tombstone_delete(self, node_id):
         log.info(f"tombstone_delete({node_id})")
         self.request(
@@ -4345,6 +4362,8 @@ class Endpoint(PgProtocol, LogUtils):
         basebackup_request_tries: int | None = None,
         timeout: str | None = None,
         env: dict[str, str] | None = None,
+        autoprewarm: bool = False,
+        offload_lfc_interval_seconds: int | None = None,
     ) -> Self:
         """
         Start the Postgres instance.
@@ -4369,6 +4388,8 @@ class Endpoint(PgProtocol, LogUtils):
             basebackup_request_tries=basebackup_request_tries,
             timeout=timeout,
             env=env,
+            autoprewarm=autoprewarm,
+            offload_lfc_interval_seconds=offload_lfc_interval_seconds,
         )
         self._running.release(1)
         self.log_config_value("shared_buffers")
@@ -4584,6 +4605,8 @@ class Endpoint(PgProtocol, LogUtils):
         pageserver_id: int | None = None,
         allow_multiple: bool = False,
         basebackup_request_tries: int | None = None,
+        autoprewarm: bool = False,
+        offload_lfc_interval_seconds: int | None = None,
     ) -> Self:
         """
         Create an endpoint, apply config, and start Postgres.
@@ -4604,6 +4627,8 @@ class Endpoint(PgProtocol, LogUtils):
             pageserver_id=pageserver_id,
             allow_multiple=allow_multiple,
             basebackup_request_tries=basebackup_request_tries,
+            autoprewarm=autoprewarm,
+            offload_lfc_interval_seconds=offload_lfc_interval_seconds,
         )
 
         return self
@@ -4688,6 +4713,8 @@ class EndpointFactory:
         remote_ext_base_url: str | None = None,
         pageserver_id: int | None = None,
         basebackup_request_tries: int | None = None,
+        autoprewarm: bool = False,
+        offload_lfc_interval_seconds: int | None = None,
     ) -> Endpoint:
         ep = Endpoint(
             self.env,
@@ -4709,6 +4736,8 @@ class EndpointFactory:
             remote_ext_base_url=remote_ext_base_url,
             pageserver_id=pageserver_id,
             basebackup_request_tries=basebackup_request_tries,
+            autoprewarm=autoprewarm,
+            offload_lfc_interval_seconds=offload_lfc_interval_seconds,
         )
 
     def create(

test_runner/fixtures/pageserver/allowed_errors.py

@@ -111,6 +111,7 @@ DEFAULT_PAGESERVER_ALLOWED_ERRORS = (
     ".*stalling layer flushes for compaction backpressure.*",
     ".*layer roll waiting for flush due to compaction backpressure.*",
     ".*BatchSpanProcessor.*",
+    ".*No broker updates received for a while.*",
     *(
         [
             r".*your platform is not a supported production platform, ignoing request for O_DIRECT; this could hide alignment bugs.*"

test_runner/fixtures/safekeeper/http.py

@@ -112,12 +112,18 @@ class TimelineCreateRequest:
 class TimelineMembershipSwitchResponse:
     previous_conf: MembershipConfiguration
     current_conf: MembershipConfiguration
+    last_log_term: int
+    flush_lsn: Lsn
 
     @classmethod
     def from_json(cls, d: dict[str, Any]) -> TimelineMembershipSwitchResponse:
         previous_conf = MembershipConfiguration.from_json(d["previous_conf"])
         current_conf = MembershipConfiguration.from_json(d["current_conf"])
-        return TimelineMembershipSwitchResponse(previous_conf, current_conf)
+        last_log_term = d["last_log_term"]
+        flush_lsn = Lsn(d["flush_lsn"])
+        return TimelineMembershipSwitchResponse(
+            previous_conf, current_conf, last_log_term, flush_lsn
+        )
 
 
 class SafekeeperHttpClient(requests.Session, MetricsGetter):

test_runner/performance/pageserver/pagebench/test_pageserver_max_throughput_getpage_at_latest_lsn.py

@@ -55,9 +55,10 @@ def test_pageserver_characterize_throughput_with_n_tenants(
 @pytest.mark.parametrize("duration", [20 * 60])
 @pytest.mark.parametrize("pgbench_scale", [get_scale_for_db(2048)])
 # we use 1 client to characterize latencies, and 64 clients to characterize throughput/scalability
-# we use 64 clients because typically for a high number of connections we recommend the connection pooler
-# which by default uses 64 connections
-@pytest.mark.parametrize("n_clients", [1, 64])
+# we use 8 clients because we see a latency knee around 6-8 clients on im4gn.2xlarge instance type,
+# which we use for this periodic test - at a cpu utilization of around 70 % - which is considered
+# a good utilization for pageserver.
+@pytest.mark.parametrize("n_clients", [1, 8])
 @pytest.mark.parametrize("n_tenants", [1])
 @pytest.mark.timeout(2400)
 def test_pageserver_characterize_latencies_with_1_client_and_throughput_with_many_clients_one_tenant(

test_runner/regress/test_branching.py

@@ -416,6 +416,8 @@ def test_duplicate_creation(neon_env_builder: NeonEnvBuilder):
         # timeline creation (uploads). mask it out here to avoid flakyness.
         del success_result["remote_consistent_lsn_visible"]
         del repeat_result["remote_consistent_lsn_visible"]
+        del success_result["walreceiver_status"]
+        del repeat_result["walreceiver_status"]
         assert repeat_result == success_result
     finally:
         env.pageserver.stop(immediate=True)

test_runner/regress/test_lfc_prewarm.py

@@ -1,34 +1,38 @@
 import random
 import threading
-import time
-from enum import Enum
+from enum import StrEnum
+from time import sleep
+from typing import Any
 
 import pytest
 from fixtures.endpoint.http import EndpointHttpClient
 from fixtures.log_helper import log
 from fixtures.neon_fixtures import NeonEnv
-from fixtures.utils import USE_LFC
+from fixtures.utils import USE_LFC, wait_until
 from prometheus_client.parser import text_string_to_metric_families as prom_parse_impl
+from psycopg2.extensions import cursor as Cursor
 
 
-class LfcQueryMethod(Enum):
-    COMPUTE_CTL = False
-    POSTGRES = True
+class PrewarmMethod(StrEnum):
+    POSTGRES = "postgres"
+    COMPUTE_CTL = "compute-ctl"
+    AUTOPREWARM = "autoprewarm"
 
 
-PREWARM_LABEL = "compute_ctl_lfc_prewarm_requests_total"
-OFFLOAD_LABEL = "compute_ctl_lfc_offload_requests_total"
-QUERY_OPTIONS = LfcQueryMethod.POSTGRES, LfcQueryMethod.COMPUTE_CTL
+PREWARM_LABEL = "compute_ctl_lfc_prewarms_total"
+OFFLOAD_LABEL = "compute_ctl_lfc_offloads_total"
+METHOD_VALUES = [e for e in PrewarmMethod]
+METHOD_IDS = [e.value for e in PrewarmMethod]
 
 
-def check_pinned_entries(cur):
+def check_pinned_entries(cur: Cursor):
     # some LFC buffer can be temporary locked by autovacuum or background writer
     for _ in range(10):
         cur.execute("select lfc_value from neon_lfc_stats where lfc_key='file_cache_chunks_pinned'")
         n_pinned = cur.fetchall()[0][0]
         if n_pinned == 0:
             break
-        time.sleep(1)
+        sleep(1)
     assert n_pinned == 0
 
 
@@ -41,21 +45,68 @@ def prom_parse(client: EndpointHttpClient) -> dict[str, float]:
     }
 
 
+def offload_lfc(method: PrewarmMethod, client: EndpointHttpClient, cur: Cursor) -> Any:
+    if method == PrewarmMethod.AUTOPREWARM:
+        client.offload_lfc_wait()
+    elif method == PrewarmMethod.COMPUTE_CTL:
+        status = client.prewarm_lfc_status()
+        assert status["status"] == "not_prewarmed"
+        assert "error" not in status
+        client.offload_lfc()
+        assert client.prewarm_lfc_status()["status"] == "not_prewarmed"
+        assert prom_parse(client) == {OFFLOAD_LABEL: 1, PREWARM_LABEL: 0}
+    elif method == PrewarmMethod.POSTGRES:
+        cur.execute("select get_local_cache_state()")
+        return cur.fetchall()[0][0]
+    else:
+        raise AssertionError(f"{method} not in PrewarmMethod")
+
+
+def prewarm_endpoint(
+    method: PrewarmMethod, client: EndpointHttpClient, cur: Cursor, lfc_state: str | None
+):
+    if method == PrewarmMethod.AUTOPREWARM:
+        client.prewarm_lfc_wait()
+    elif method == PrewarmMethod.COMPUTE_CTL:
+        client.prewarm_lfc()
+    elif method == PrewarmMethod.POSTGRES:
+        cur.execute("select prewarm_local_cache(%s)", (lfc_state,))
+
+
+def check_prewarmed(
+    method: PrewarmMethod, client: EndpointHttpClient, desired_status: dict[str, str | int]
+):
+    if method == PrewarmMethod.AUTOPREWARM:
+        assert client.prewarm_lfc_status() == desired_status
+        assert prom_parse(client)[PREWARM_LABEL] == 1
+    elif method == PrewarmMethod.COMPUTE_CTL:
+        assert client.prewarm_lfc_status() == desired_status
+        assert prom_parse(client) == {OFFLOAD_LABEL: 0, PREWARM_LABEL: 1}
+
+
 @pytest.mark.skipif(not USE_LFC, reason="LFC is disabled, skipping")
-@pytest.mark.parametrize("query", QUERY_OPTIONS, ids=["postgres", "compute-ctl"])
-def test_lfc_prewarm(neon_simple_env: NeonEnv, query: LfcQueryMethod):
+@pytest.mark.parametrize("method", METHOD_VALUES, ids=METHOD_IDS)
+def test_lfc_prewarm(neon_simple_env: NeonEnv, method: PrewarmMethod):
     env = neon_simple_env
     n_records = 1000000
-    endpoint = env.endpoints.create_start(
-        branch_name="main",
-        config_lines=[
-            "autovacuum = off",
-            "shared_buffers=1MB",
-            "neon.max_file_cache_size=1GB",
-            "neon.file_cache_size_limit=1GB",
-            "neon.file_cache_prewarm_limit=1000",
-        ],
-    )
+    cfg = [
+        "autovacuum = off",
+        "shared_buffers=1MB",
+        "neon.max_file_cache_size=1GB",
+        "neon.file_cache_size_limit=1GB",
+        "neon.file_cache_prewarm_limit=1000",
+    ]
+    offload_secs = 2
+
+    if method == PrewarmMethod.AUTOPREWARM:
+        endpoint = env.endpoints.create_start(
+            branch_name="main",
+            config_lines=cfg,
+            autoprewarm=True,
+            offload_lfc_interval_seconds=offload_secs,
+        )
+    else:
+        endpoint = env.endpoints.create_start(branch_name="main", config_lines=cfg)
 
     pg_conn = endpoint.connect()
     pg_cur = pg_conn.cursor()
@@ -69,31 +120,21 @@ def test_lfc_prewarm(neon_simple_env: NeonEnv, query: LfcQueryMethod):
     lfc_cur.execute(f"insert into t (pk) values (generate_series(1,{n_records}))")
     log.info(f"Inserted {n_records} rows")
 
-    http_client = endpoint.http_client()
-    if query is LfcQueryMethod.COMPUTE_CTL:
-        status = http_client.prewarm_lfc_status()
-        assert status["status"] == "not_prewarmed"
-        assert "error" not in status
-        http_client.offload_lfc()
-        assert http_client.prewarm_lfc_status()["status"] == "not_prewarmed"
-        assert prom_parse(http_client) == {OFFLOAD_LABEL: 1, PREWARM_LABEL: 0}
-    else:
-        pg_cur.execute("select get_local_cache_state()")
-        lfc_state = pg_cur.fetchall()[0][0]
+    client = endpoint.http_client()
+    lfc_state = offload_lfc(method, client, pg_cur)
 
     endpoint.stop()
-    endpoint.start()
+    if method == PrewarmMethod.AUTOPREWARM:
+        endpoint.start(autoprewarm=True, offload_lfc_interval_seconds=offload_secs)
+    else:
+        endpoint.start()
 
     pg_conn = endpoint.connect()
     pg_cur = pg_conn.cursor()
 
     lfc_conn = endpoint.connect(dbname="lfc")
     lfc_cur = lfc_conn.cursor()
-
-    if query is LfcQueryMethod.COMPUTE_CTL:
-        http_client.prewarm_lfc()
-    else:
-        pg_cur.execute("select prewarm_local_cache(%s)", (lfc_state,))
+    prewarm_endpoint(method, client, pg_cur, lfc_state)
 
     pg_cur.execute("select lfc_value from neon_lfc_stats where lfc_key='file_cache_used_pages'")
     lfc_used_pages = pg_cur.fetchall()[0][0]
@@ -111,33 +152,32 @@ def test_lfc_prewarm(neon_simple_env: NeonEnv, query: LfcQueryMethod):
         and prewarm_info[1] > 0
         and prewarm_info[0] == prewarm_info[1] + prewarm_info[2]
     )
-
     lfc_cur.execute("select sum(pk) from t")
     assert lfc_cur.fetchall()[0][0] == n_records * (n_records + 1) / 2
 
     check_pinned_entries(pg_cur)
-
     desired = {"status": "completed", "total": total, "prewarmed": prewarmed, "skipped": skipped}
-    if query is LfcQueryMethod.COMPUTE_CTL:
-        assert http_client.prewarm_lfc_status() == desired
-        assert prom_parse(http_client) == {OFFLOAD_LABEL: 0, PREWARM_LABEL: 1}
+    check_prewarmed(method, client, desired)
+
+
+# autoprewarm isn't needed as we prewarm manually
+WORKLOAD_VALUES = METHOD_VALUES[:-1]
+WORKLOAD_IDS = METHOD_IDS[:-1]
 
 
 @pytest.mark.skipif(not USE_LFC, reason="LFC is disabled, skipping")
-@pytest.mark.parametrize("query", QUERY_OPTIONS, ids=["postgres", "compute-ctl"])
-def test_lfc_prewarm_under_workload(neon_simple_env: NeonEnv, query: LfcQueryMethod):
+@pytest.mark.parametrize("method", WORKLOAD_VALUES, ids=WORKLOAD_IDS)
+def test_lfc_prewarm_under_workload(neon_simple_env: NeonEnv, method: PrewarmMethod):
     env = neon_simple_env
     n_records = 10000
     n_threads = 4
-    endpoint = env.endpoints.create_start(
-        branch_name="main",
-        config_lines=[
-            "shared_buffers=1MB",
-            "neon.max_file_cache_size=1GB",
-            "neon.file_cache_size_limit=1GB",
-            "neon.file_cache_prewarm_limit=1000000",
-        ],
-    )
+    cfg = [
+        "shared_buffers=1MB",
+        "neon.max_file_cache_size=1GB",
+        "neon.file_cache_size_limit=1GB",
+        "neon.file_cache_prewarm_limit=1000000",
+    ]
+    endpoint = env.endpoints.create_start(branch_name="main", config_lines=cfg)
 
     pg_conn = endpoint.connect()
     pg_cur = pg_conn.cursor()
@@ -154,12 +194,7 @@ def test_lfc_prewarm_under_workload(neon_simple_env: NeonEnv, query: LfcQueryMet
     log.info(f"Inserted {n_records} rows")
 
     http_client = endpoint.http_client()
-    if query is LfcQueryMethod.COMPUTE_CTL:
-        http_client.offload_lfc()
-    else:
-        pg_cur.execute("select get_local_cache_state()")
-        lfc_state = pg_cur.fetchall()[0][0]
-
+    lfc_state = offload_lfc(method, http_client, pg_cur)
     running = True
     n_prewarms = 0
 
@@ -170,8 +205,8 @@ def test_lfc_prewarm_under_workload(neon_simple_env: NeonEnv, query: LfcQueryMet
         while running:
             src = random.randint(1, n_records)
             dst = random.randint(1, n_records)
-            lfc_cur.execute("update accounts set balance=balance-100 where id=%s", (src,))
-            lfc_cur.execute("update accounts set balance=balance+100 where id=%s", (dst,))
+            lfc_cur.execute(f"update accounts set balance=balance-100 where id={src}")
+            lfc_cur.execute(f"update accounts set balance=balance+100 where id={dst}")
             n_transfers += 1
         log.info(f"Number of transfers: {n_transfers}")
 
@@ -183,13 +218,7 @@ def test_lfc_prewarm_under_workload(neon_simple_env: NeonEnv, query: LfcQueryMet
             pg_cur.execute("select pg_reload_conf()")
             pg_cur.execute("alter system set neon.file_cache_size_limit='1GB'")
             pg_cur.execute("select pg_reload_conf()")
-
-            if query is LfcQueryMethod.COMPUTE_CTL:
-                # Same thing as prewarm_lfc(), testing other method
-                http_client.prewarm_lfc(endpoint.endpoint_id)
-            else:
-                pg_cur.execute("select prewarm_local_cache(%s)", (lfc_state,))
-
+            prewarm_endpoint(method, http_client, pg_cur, lfc_state)
             nonlocal n_prewarms
             n_prewarms += 1
         log.info(f"Number of prewarms: {n_prewarms}")
@@ -203,7 +232,10 @@ def test_lfc_prewarm_under_workload(neon_simple_env: NeonEnv, query: LfcQueryMet
     prewarm_thread = threading.Thread(target=prewarm)
     prewarm_thread.start()
 
-    time.sleep(20)
+    def prewarmed():
+        assert n_prewarms > 5
+
+    wait_until(prewarmed)
 
     running = False
     for t in workload_threads:
@@ -215,5 +247,5 @@ def test_lfc_prewarm_under_workload(neon_simple_env: NeonEnv, query: LfcQueryMet
     assert total_balance == 0
 
     check_pinned_entries(pg_cur)
-    if query is LfcQueryMethod.COMPUTE_CTL:
+    if method != PrewarmMethod.POSTGRES:
         assert prom_parse(http_client) == {OFFLOAD_LABEL: 1, PREWARM_LABEL: n_prewarms}

test_runner/regress/test_storage_controller.py

@@ -2618,7 +2618,7 @@ def test_storage_controller_node_deletion(
         wait_until(assert_shards_migrated)
 
     log.info(f"Deleting pageserver {victim.id}")
-    env.storage_controller.node_delete(victim.id)
+    env.storage_controller.node_delete_old(victim.id)
 
     if not while_offline:
 
@@ -2653,6 +2653,60 @@ def test_storage_controller_node_deletion(
     env.storage_controller.consistency_check()
 
 
+def test_storage_controller_node_delete_cancellation(neon_env_builder: NeonEnvBuilder):
+    neon_env_builder.num_pageservers = 3
+    neon_env_builder.num_azs = 3
+    env = neon_env_builder.init_configs()
+    env.start()
+
+    tenant_count = 12
+    shard_count_per_tenant = 16
+    tenant_ids = []
+
+    for _ in range(0, tenant_count):
+        tid = TenantId.generate()
+        tenant_ids.append(tid)
+        env.create_tenant(
+            tid, placement_policy='{"Attached":1}', shard_count=shard_count_per_tenant
+        )
+
+    # Sanity check: initial creations should not leave the system in an unstable scheduling state
+    assert env.storage_controller.reconcile_all() == 0
+
+    nodes = env.storage_controller.node_list()
+    assert len(nodes) == 3
+
+    env.storage_controller.configure_failpoints(("sleepy-delete-loop", "return(10000)"))
+
+    ps_id_to_delete = env.pageservers[0].id
+
+    env.storage_controller.warm_up_all_secondaries()
+    env.storage_controller.retryable_node_operation(
+        lambda ps_id: env.storage_controller.node_delete(ps_id),
+        ps_id_to_delete,
+        max_attempts=3,
+        backoff=2,
+    )
+
+    env.storage_controller.poll_node_status(
+        ps_id_to_delete,
+        PageserverAvailability.ACTIVE,
+        PageserverSchedulingPolicy.DELETING,
+        max_attempts=6,
+        backoff=2,
+    )
+
+    env.storage_controller.cancel_node_delete(ps_id_to_delete)
+
+    env.storage_controller.poll_node_status(
+        ps_id_to_delete,
+        PageserverAvailability.ACTIVE,
+        PageserverSchedulingPolicy.ACTIVE,
+        max_attempts=6,
+        backoff=2,
+    )
+
+
 @pytest.mark.parametrize("shard_count", [None, 2])
 def test_storage_controller_metadata_health(
     neon_env_builder: NeonEnvBuilder,
@@ -3208,7 +3262,7 @@ def test_ps_unavailable_after_delete(neon_env_builder: NeonEnvBuilder):
     assert_nodes_count(3)
 
     ps = env.pageservers[0]
-    env.storage_controller.node_delete(ps.id)
+    env.storage_controller.node_delete_old(ps.id)
 
     # After deletion, the node count must be reduced
     assert_nodes_count(2)

test_runner/regress/test_wal_receiver.py

@@ -13,50 +13,6 @@ if TYPE_CHECKING:
     from fixtures.neon_fixtures import NeonEnv, NeonEnvBuilder
 
 
-# Checks that pageserver's walreceiver state is printed in the logs during WAL wait timeout.
-# Ensures that walreceiver does not run without any data inserted and only starts after the insertion.
-def test_pageserver_lsn_wait_error_start(neon_env_builder: NeonEnvBuilder):
-    # we assert below that the walreceiver is not active before data writes.
-    # with manually created timelines, it is active.
-    # FIXME: remove this test once we remove timelines_onto_safekeepers
-    neon_env_builder.storage_controller_config = {
-        "timelines_onto_safekeepers": False,
-    }
-
-    # Trigger WAL wait timeout faster
-    neon_env_builder.pageserver_config_override = "wait_lsn_timeout = '1s'"
-    env = neon_env_builder.init_start()
-    env.pageserver.http_client()
-
-    # In this test we force 'Timed out while waiting for WAL record error' while
-    # fetching basebackup and don't want any retries.
-    os.environ["NEON_COMPUTE_TESTING_BASEBACKUP_RETRIES"] = "1"
-
-    tenant_id, timeline_id = env.create_tenant()
-    expected_timeout_error = f"Timed out while waiting for WAL record at LSN {future_lsn} to arrive"
-    env.pageserver.allowed_errors.append(f".*{expected_timeout_error}.*")
-
-    try:
-        trigger_wait_lsn_timeout(env, tenant_id)
-    except Exception as e:
-        exception_string = str(e)
-        assert expected_timeout_error in exception_string, "Should time out during waiting for WAL"
-        assert "WalReceiver status: Not active" in exception_string, (
-            "Walreceiver should not be active before any data writes"
-        )
-
-    insert_test_elements(env, tenant_id, start=0, count=1_000)
-    try:
-        trigger_wait_lsn_timeout(env, tenant_id)
-    except Exception as e:
-        exception_string = str(e)
-        assert expected_timeout_error in exception_string, "Should time out during waiting for WAL"
-        assert "WalReceiver status: Not active" not in exception_string, (
-            "Should not be inactive anymore after INSERTs are made"
-        )
-        assert "WalReceiver status" in exception_string, "But still should have some other status"
-
-
 # Checks that all active safekeepers are shown in pageserver's walreceiver state printed on WAL wait timeout.
 # Kills one of the safekeepers and ensures that only the active ones are printed in the state.
 def test_pageserver_lsn_wait_error_safekeeper_stop(neon_env_builder: NeonEnvBuilder):

test_runner/sql_regress/expected/neon-subxacts.out

@@ -0,0 +1,21 @@
+DO $$
+DECLARE
+i numeric;
+BEGIN
+  create role somebody;
+  FOR i IN 1..1000000 LOOP
+    BEGIN
+	  IF i % 1000 = 0 THEN
+	    alter role somebody password 'welcome';
+	  ELSE
+        PERFORM 1;
+	  END IF;
+    EXCEPTION WHEN OTHERS THEN
+      RAISE WARNING 'error';
+    END;
+    IF I = 1000000 THEN
+      PERFORM pg_log_backend_memory_contexts(pg_backend_pid());
+    END IF;
+  END LOOP;
+END;
+$$;

test_runner/sql_regress/parallel_schedule

@@ -10,3 +10,4 @@ test: neon-clog
 test: neon-test-utils
 test: neon-vacuum-full
 test: neon-event-triggers
+test: neon-subxacts

test_runner/sql_regress/sql/neon-subxacts.sql

@@ -0,0 +1,21 @@
+DO $$
+DECLARE
+i numeric;
+BEGIN
+  create role somebody;
+  FOR i IN 1..1000000 LOOP
+    BEGIN
+	  IF i % 1000 = 0 THEN
+	    alter role somebody password 'welcome';
+	  ELSE
+        PERFORM 1;
+	  END IF;
+    EXCEPTION WHEN OTHERS THEN
+      RAISE WARNING 'error';
+    END;
+    IF I = 1000000 THEN
+      PERFORM pg_log_backend_memory_contexts(pg_backend_pid());
+    END IF;
+  END LOOP;
+END;
+$$;

workspace_hack/Cargo.toml

@@ -68,6 +68,7 @@ num-integer = { version = "0.1", features = ["i128"] }
 num-iter = { version = "0.1", default-features = false, features = ["i128", "std"] }
 num-rational = { version = "0.4", default-features = false, features = ["num-bigint-std", "std"] }
 num-traits = { version = "0.2", features = ["i128", "libm"] }
+once_cell = { version = "1" }
 p256 = { version = "0.13", features = ["jwk"] }
 parquet = { version = "53", default-features = false, features = ["zstd"] }
 prost = { version = "0.13", features = ["no-recursion-limit", "prost-derive"] }
@@ -112,10 +113,13 @@ zstd-sys = { version = "2", default-features = false, features = ["legacy", "std
 
 [build-dependencies]
 ahash = { version = "0.8" }
+anstream = { version = "0.6" }
 anyhow = { version = "1", features = ["backtrace"] }
 bytes = { version = "1", features = ["serde"] }
 cc = { version = "1", default-features = false, features = ["parallel"] }
 chrono = { version = "0.4", default-features = false, features = ["clock", "serde", "wasmbind"] }
+clap = { version = "4", features = ["derive", "env", "string"] }
+clap_builder = { version = "4", default-features = false, features = ["color", "env", "help", "std", "string", "suggestions", "usage"] }
 either = { version = "1" }
 getrandom = { version = "0.2", default-features = false, features = ["std"] }
 half = { version = "2", default-features = false, features = ["num-traits"] }
@@ -133,6 +137,7 @@ num-integer = { version = "0.1", features = ["i128"] }
 num-iter = { version = "0.1", default-features = false, features = ["i128", "std"] }
 num-rational = { version = "0.4", default-features = false, features = ["num-bigint-std", "std"] }
 num-traits = { version = "0.2", features = ["i128", "libm"] }
+once_cell = { version = "1" }
 parquet = { version = "53", default-features = false, features = ["zstd"] }
 prettyplease = { version = "0.2", default-features = false, features = ["verbatim"] }
 proc-macro2 = { version = "1" }
@@ -142,6 +147,7 @@ regex = { version = "1" }
 regex-automata = { version = "0.4", default-features = false, features = ["dfa-onepass", "hybrid", "meta", "nfa-backtrack", "perf-inline", "perf-literal", "unicode"] }
 regex-syntax = { version = "0.8" }
 serde = { version = "1", features = ["alloc", "derive"] }
+serde_json = { version = "1", features = ["alloc", "raw_value"] }
 syn = { version = "2", features = ["extra-traits", "fold", "full", "visit", "visit-mut"] }
 time-macros = { version = "0.2", default-features = false, features = ["formatting", "parsing", "serde"] }
 toml_edit = { version = "0.22", features = ["serde"] }