pageserver: update doc comments referencing renamed types

.cargo/config.toml

@@ -11,7 +11,7 @@ rustdocflags = ["-Arustdoc::private_intra_doc_links"]
 # * <https://github.com/rust-lang/rust/pull/122646>
 #
 # NB: the RUSTFLAGS envvar will replace this. Make sure to update e.g. Dockerfile as well.
-rustflags = ["-Cforce-frame-pointers=yes", "--cfg", "tokio_unstable", "--cfg", "tokio_taskdump"]
+rustflags = ["-Cforce-frame-pointers=yes"]
 
 [alias]
 build_testing = ["build", "--features", "testing"]

.github/workflows/build_and_test.yml

@@ -824,7 +824,7 @@ jobs:
           docker compose -f ./docker-compose/docker-compose.yml down
 
   promote-images-dev:
-    needs: [ check-permissions, tag, vm-compute-node-image, neon-image ]
+    needs: [ check-permissions, tag, vm-compute-node-image ]
     runs-on: ubuntu-22.04
 
     permissions:

Cargo.lock

@@ -1312,7 +1312,6 @@ dependencies = [
  "tracing-utils",
  "url",
  "utils",
- "uuid",
  "vm_monitor",
  "workspace_hack",
  "zstd",

compute/compute-node.Dockerfile

@@ -995,50 +995,24 @@ RUN wget https://github.com/kelvich/pg_tiktoken/archive/9118dd4549b7d8c0bbc98e04
 #########################################################################################
 #
 # Layer "pg-pgx-ulid-build"
-# Compile "pgx_ulid" extension for v16 and below
+# Compile "pgx_ulid" extension
 #
 #########################################################################################
 
 FROM rust-extensions-build AS pg-pgx-ulid-build
 ARG PG_VERSION
 
-RUN case "${PG_VERSION}" in \
-    "v14" | "v15" | "v16") \
-        ;; \
-    *) \
-        echo "skipping the version of pgx_ulid for $PG_VERSION" && exit 0 \
-        ;; \
+# doesn't support v17 yet
+# https://github.com/pksunkara/pgx_ulid/pull/52
+RUN case "${PG_VERSION}" in "v17") \
+    echo "pgx_ulid does not support pg17 as of the latest version (0.1.5)" && exit 0;; \
     esac && \
     wget https://github.com/pksunkara/pgx_ulid/archive/refs/tags/v0.1.5.tar.gz -O pgx_ulid.tar.gz && \
-    echo "9d1659a2da65af0133d5451c454de31b37364e3502087dadf579f790bc8bef17  pgx_ulid.tar.gz" | sha256sum --check && \
+    echo "9d1659a2da65af0133d5451c454de31b37364e3502087dadf579f790bc8bef17 pgx_ulid.tar.gz" | sha256sum --check && \
     mkdir pgx_ulid-src && cd pgx_ulid-src && tar xzf ../pgx_ulid.tar.gz --strip-components=1 -C . && \
-    sed -i 's/pgrx       = "^0.11.2"/pgrx       = { version = "0.11.3", features = [ "unsafe-postgres" ] }/g' Cargo.toml && \
+    sed -i 's/pgrx       = "^0.11.2"/pgrx = { version = "=0.11.3", features = [ "unsafe-postgres" ] }/g' Cargo.toml && \
     cargo pgrx install --release && \
-    echo 'trusted = true' >> /usr/local/pgsql/share/extension/ulid.control
-
-#########################################################################################
-#
-# Layer "pg-pgx-ulid-pgrx12-build"
-# Compile "pgx_ulid" extension for v17 and up
-#
-#########################################################################################
-
-FROM rust-extensions-build-pgrx12 AS pg-pgx-ulid-pgrx12-build
-ARG PG_VERSION
-
-RUN case "${PG_VERSION}" in \
-    "v17") \
-        ;; \
-    *) \
-        echo "skipping the version of pgx_ulid for $PG_VERSION" && exit 0 \
-        ;; \
-    esac && \
-    wget https://github.com/pksunkara/pgx_ulid/archive/refs/tags/v0.2.0.tar.gz -O pgx_ulid.tar.gz && \
-    echo "cef6a9a2e5e7bd1a10a18989286586ee9e6c1c06005a4055cff190de41bf3e9f pgx_ulid.tar.gz" | sha256sum --check && \
-    mkdir pgx_ulid-src && cd pgx_ulid-src && tar xzf ../pgx_ulid.tar.gz --strip-components=1 -C . && \
-    sed -i 's/pgrx       = "^0.12.7"/pgrx       = { version = "0.12.9", features = [ "unsafe-postgres" ] }/g' Cargo.toml && \
-    cargo pgrx install --release && \
-    echo 'trusted = true' >> /usr/local/pgsql/share/extension/pgx_ulid.control
+    echo "trusted = true" >> /usr/local/pgsql/share/extension/ulid.control
 
 #########################################################################################
 #
@@ -1183,7 +1157,6 @@ COPY --from=timescaledb-pg-build /usr/local/pgsql/ /usr/local/pgsql/
 COPY --from=pg-hint-plan-pg-build /usr/local/pgsql/ /usr/local/pgsql/
 COPY --from=pg-cron-pg-build /usr/local/pgsql/ /usr/local/pgsql/
 COPY --from=pg-pgx-ulid-build /usr/local/pgsql/ /usr/local/pgsql/
-COPY --from=pg-pgx-ulid-pgrx12-build /usr/local/pgsql/ /usr/local/pgsql/
 COPY --from=pg-session-jwt-build /usr/local/pgsql/ /usr/local/pgsql/
 COPY --from=rdkit-pg-build /usr/local/pgsql/ /usr/local/pgsql/
 COPY --from=pg-uuidv7-pg-build /usr/local/pgsql/ /usr/local/pgsql/

compute_tools/Cargo.toml

@@ -51,7 +51,6 @@ tracing-subscriber.workspace = true
 tracing-utils.workspace = true
 thiserror.workspace = true
 url.workspace = true
-uuid.workspace = true
 prometheus.workspace = true
 
 postgres_initdb.workspace = true

compute_tools/src/http/server.rs

@@ -1,14 +1,15 @@
 use std::{
     net::{IpAddr, Ipv6Addr, SocketAddr},
-    sync::Arc,
+    sync::{
+        atomic::{AtomicU64, Ordering},
+        Arc,
+    },
     thread,
     time::Duration,
 };
 
 use anyhow::Result;
 use axum::{
-    extract::Request,
-    middleware::{self, Next},
     response::{IntoResponse, Response},
     routing::{get, post},
     Router,
@@ -16,9 +17,11 @@ use axum::{
 use http::StatusCode;
 use tokio::net::TcpListener;
 use tower::ServiceBuilder;
-use tower_http::{request_id::PropagateRequestIdLayer, trace::TraceLayer};
+use tower_http::{
+    request_id::{MakeRequestId, PropagateRequestIdLayer, RequestId, SetRequestIdLayer},
+    trace::TraceLayer,
+};
 use tracing::{debug, error, info, Span};
-use uuid::Uuid;
 
 use super::routes::{
     check_writability, configure, database_schema, dbs_and_roles, extension_server, extensions,
@@ -31,24 +34,30 @@ async fn handle_404() -> Response {
     StatusCode::NOT_FOUND.into_response()
 }
 
-const X_REQUEST_ID: &str = "x-request-id";
+#[derive(Clone, Default)]
+struct ComputeMakeRequestId(Arc<AtomicU64>);
 
-/// This middleware function allows compute_ctl to generate its own request ID
-/// if one isn't supplied. The control plane will always send one as a UUID. The
-/// neon Postgres extension on the other hand does not send one.
-async fn maybe_add_request_id_header(mut request: Request, next: Next) -> Response {
-    let headers = request.headers_mut();
+impl MakeRequestId for ComputeMakeRequestId {
+    fn make_request_id<B>(
+        &mut self,
+        _request: &http::Request<B>,
+    ) -> Option<tower_http::request_id::RequestId> {
+        let request_id = self
+            .0
+            .fetch_add(1, Ordering::SeqCst)
+            .to_string()
+            .parse()
+            .unwrap();
 
-    if headers.get(X_REQUEST_ID).is_none() {
-        headers.append(X_REQUEST_ID, Uuid::new_v4().to_string().parse().unwrap());
+        Some(RequestId::new(request_id))
     }
-
-    next.run(request).await
 }
 
 /// Run the HTTP server and wait on it forever.
 #[tokio::main]
 async fn serve(port: u16, compute: Arc<ComputeNode>) {
+    const X_REQUEST_ID: &str = "x-request-id";
+
     let mut app = Router::new()
         .route("/check_writability", post(check_writability::is_writable))
         .route("/configure", post(configure::configure))
@@ -73,8 +82,9 @@ async fn serve(port: u16, compute: Arc<ComputeNode>) {
         .fallback(handle_404)
         .layer(
             ServiceBuilder::new()
-                // Add this middleware since we assume the request ID exists
-                .layer(middleware::from_fn(maybe_add_request_id_header))
+                .layer(SetRequestIdLayer::x_request_id(
+                    ComputeMakeRequestId::default(),
+                ))
                 .layer(
                     TraceLayer::new_for_http()
                         .on_request(|request: &http::Request<_>, _span: &Span| {

control_plane/storcon_cli/src/main.rs

@@ -298,7 +298,14 @@ impl FromStr for SkSchedulingPolicyArg {
     type Err = anyhow::Error;
 
     fn from_str(s: &str) -> Result<Self, Self::Err> {
-        SkSchedulingPolicy::from_str(s).map(Self)
+        match s {
+            "active" => Ok(Self(SkSchedulingPolicy::Active)),
+            "disabled" => Ok(Self(SkSchedulingPolicy::Disabled)),
+            "decomissioned" => Ok(Self(SkSchedulingPolicy::Decomissioned)),
+            _ => Err(anyhow::anyhow!(
+                "Unknown scheduling policy '{s}', try active,disabled,decomissioned"
+            )),
+        }
     }
 }
 

docs/rfcs/041-sharded-ingest.md

@@ -1,255 +0,0 @@
-# 
-Created on Aug 2024
-Implemented on Jan 2025
-
-## Summary
-
-Data in large tenants is split up between multiple pageservers according to key hashes, as
-introduced in the [sharding RFC](031-sharding-static.md) and [shard splitting RFC](032-shard-splitting.md).
-
-Whereas currently we send all WAL to all pageserver shards, and each shard filters out the data that it needs,
-in this RFC we add a mechanism to filter the WAL on the safekeeper, so that each shard receives
-only the data it needs.
-
-This will place some extra CPU load on the safekeepers, in exchange for reducing the network bandwidth
-for ingesting WAL back to scaling as O(1) with shard count, rather than O(N_shards).
-
-## Motivation
-
-1. Large databases require higher shard counts.  Whereas currently we run with up to 8 shards for tenants
-with a few TB of storage, the next order of magnitude capacity increase will require tens of shards, such
-that sending all WAL to all shards is impractical in terms of bandwidth.
-2. For contemporary database sizes (~2TB), the pageserver is the bottleneck for ingest: since each
-   shard has to decode and process the whole WAL, sharding doesn't fully relieve this bottleneck.  To achieve significantly higher ingest speeds, we need to filter the WAL earlier so that each pageserver
-   only has to process relevant parts.
-
-## Non Goals (if relevant)
-
-We do not seek to introduce multiple WALs per timeline, or to share the work of handling a timeline's
-WAL across safekeepers (beyond simple 3x replication).  This RFC may be thought of as an incremental
-move of the ingestion bottleneck up the stack: instead of high write rates bottlenecking on the
-pageserver, they will bottleneck on the safekeeper.
-
-## Impacted components (e.g. pageserver, safekeeper, console, etc)
-
-Safekeeper, pageserver.
-
-There will be no control plane or storage controller coordination needed, as pageservers will directly
-indicate their sharding parameters to the safekeeper when subscribing for WAL.
-
-## Proposed implementation
-
-Terminology:
-- "Data pages" refers to postgres relation blocks, and SLRU blocks.
-- "Metadata pages" refers to everything else the pageserver stores, such as relation sizes and
-  directories of relations.
-
-### Phase 1: Refactor ingest
-
-Currently, pageserver ingest code is structured approximately as follows:
-1. `handle_walreceiver_connection` reads a stream of binary WAL records off a network
-   socket
-2. `WalIngest::ingest_record` to translate the record into a series of page-level modifications
-3. `DatadirModification` accumulates page updates from several `ingest_record` calls, and when
-   its `commit()` method is called, flushes these into a Timeline's open `InMemoryLayer`.
-
-This process currently assumes access to a pageserver `Timeline` throughout `ingest_record` and
-from `DatadirModification`, which is used to do read-modify-write cycles on metadata pages
-such as relation sizes and the master DBDIR page.  It also assumes that records are ingested
-strictly one after the other: they cannot be ingested in parallel because each record assumes
-that earlier records' changes have already been applied to `Timeline`.
-
-This code will be refactored to disentangle the simple, fast decode of relation page writes
-from the more complex logic for updating internal metadata.  An intermediate representation
-called `InterpretedWalRecords` will be introduced.  This is similar to the internal state of
-a `DatadirModification`, but does not require access to a Timeline.  Instead of storing
-metadata updates as materialized writes to pages, it will accumulate these as abstract operations,
-for example rather than including a write to a relation size key, this structure will include
-an operation that indicates "Update relation _foo_'s size to the max of its current value and
-_bar_", such that these may be applied later to a real Timeline.
-
-The `DatadirModification` will be aware of the `EphemeralFile` format, so that as it accumulates
-simple page writes of relation blocks, it can write them directly into a buffer in the serialized
-format.  This will avoid the need to later deserialize/reserialize this data when passing the
-structure between safekeeper and pageserver.
-
-The new pipeline will be:
-1. `handle_walreceiver_connection` reads a stream of binary WAL records off a network
-2. A `InterpretedWalRecords` is generated from the incoming WAL records.  This does not
-   require a reference to a Timeline.
-3. The logic that is current spread between `WalIngest` and `DatadirModification` for updating
-   metadata will be refactored to consume the metadata operations from the `InterpretedWalRecords`
-   and turn them into literal writes to metadata pages.  This part must be done sequentially.
-4. The resulting buffer of metadata page writes is combined with the buffer of relation block
-   writes, and written into the `InMemoryLayer`.
-
-Implemented in:
-1. https://github.com/neondatabase/neon/pull/9472
-2. https://github.com/neondatabase/neon/pull/9504
-3. https://github.com/neondatabase/neon/pull/9524
-
-### Phase 2: Decode & filter on safekeeper
-
-In the previous phase, the ingest code was modified to be able to do most of its work without access to
-a Timeline: this first stage of ingest simply converts a series of binary wal records into
-a buffer of relation/SLRU page writes, and a buffer of abstract metadata writes.
-
-The modified ingest code may be transplanted from pageserver to safekeeper (probably via a
-shared crate).  The safekeeper->pageserver network protocol is modified to:
- - in subscription requests, send the `ShardIdentity` from the pageserver to the safekeeper
- - in responses, transmit a `InterpretedWalRecords` instead of a raw `WalRecord`.
- - use the `ShardIdentity` to filter the `ProcessedWalIngest` to relevant content for
-   the subscribing shard before transmitting it.
-
-The overall behavior of the pageserver->safekeeper interaction remains the same, in terms of
-consistent LSN feedback, and connection management.  Only the payload of the subscriptions
-changes, to express an LSN range of WAL as a filtered `ProcessedWalIngest` instead of the
-raw data.
-
-The ingest code on the pageserver can now skip the part where it does the first phase of
-processing, as it will receive pre-processed, compressed data off the wire.
-
-Note that `InterpretedWalRecord` batches multiple `InterpretedWalRecord(s)` in the same network
-message. Safekeeper reads WAL in chunks of 16 blocks and then decodes as many Postgres WAL records
-as possible. Each Postgres WAL record maps to one `InterpretedWalRecord` for potentially multiple shards.
-Hence, the size of the batch is given by the number of Postgres WAL records that fit in 16 blocks.
-
-The protocol needs to support evolution. Protobuf was chosen here with the view that, in the future,
-we may migrate it to GRPC altogether
-
-Implemented in:
-1. https://github.com/neondatabase/neon/pull/9746
-2. https://github.com/neondatabase/neon/pull/9821
-
-### Phase 3: Fan out interpreted WAL
-
-In the previous phase, the initial processing of WAL was moved to the safekeeper, but it is still
-done once for each shard: this will generate O(N_shards) CPU work on the safekeeper (especially
-when considering converting to Protobuf format and compression).
-
-To avoid this, we fan-out WAL from one (tenant, timeline, shard) to all other shards subscribed on
-the same safekeeper. Under normal operation, the WAL will be read from disk, decoded and interpreted
-_only_ once per (safekeeper, timeline).
-
-When the first shard of a sharded timeline subscribes to a given safekeeper a task is spawned
-for the WAL reader (`InterpretedWalReader`). This task reads WAL, decodes, interprets it and sends
-it to the sender (`InterpretedWalSender`). The sender is a future that is polled from the connection
-task. When further shards subscribe on the safekeeper they will attach themselves to the existing WAL reader.
-There's two cases to consider:
-1. The shard's requested `start_lsn` is ahead of the current position of the WAL reader. In this case, the shard
-will start receiving data when the reader reaches that LSN. The intuition here is that there's little to gain
-by letting shards "front-run" since compute backpressure is based on the laggard LSN.
-2. The shard's requested `start_lsn` is below the current position of the WAL reader. In this case, the WAL reader
-gets reset to this requested position (same intuition). Special care is taken such that advanced shards do not receive
-interpreted WAL records below their current position.
-
-The approach above implies that there is at most one WAL reader per (tenant, timeline) on a given safekeeper at any point in time.
-If this turns out to be operationally problematic, there's a trick we can deploy: `--max-delta-for-fanout` is an optional safekeeper
-argument that controls the max absolute delta between a new shard and the current WAL position of the WAL reader. If the absolute
-delta is above that value, a new reader is spawned. Note that there's currently no concurrency control on the number of WAL readers,
-so it's recommended to use large values to avoid pushing CPU utilisation too high.
-
-Unsharded tenants do not spawn a separate task for the interpreted WAL reader since there's no benefit to it. Instead they poll
-the reader and sender concurrently from the connection task.
-
-Shard splits are interesting here because it is the only case when the same shard might have two subscriptions at the same time.
-This is handled by giving readers a unique identifier. Both shards will receive the same data while respecting their requested start
-position.
-
-Implemented in:
-1. https://github.com/neondatabase/neon/pull/10190
-
-## Deployment
-
-Each phase shall be deployed independently. Special care should be taken around protocol changes.
-
-## Observability Tips
-
-* The safekeeper logs the protocol requested by the pageserver
-along with the pageserver ID, tenant, timeline and shard: `starting streaming from`.
-* There's metrics for the number of wal readers:
-  * `safekeeper_wal_readers{kind="task", target=~"pageserver.*"}` gives the number of wal reader tasks for each SK
-  * `safekeeper_wal_readers{kind="future", target=~"pageserver.*"}` gives the numer of wal readers polled inline by each SK
-  * `safekeeper_interpreted_wal_reader_tasks` gives the number of wal reader tasks per tenant, timeline
-* Interesting log lines for the fan-out reader:
-  * `Spawning interpreted`: first shard creates the interpreted wal reader
-  * `Fanning out`: a subsequent shard attaches itself to an interpreted wal reader
-  * `Aborting interpreted`: all senders have finished and the reader task is being aborted
-
-## Future Optimizations
-
-This sections describes some improvement areas which may be revisited in the future.
-
-### Buffering of Interpreted WAL
-
-The interpreted WAL reader may buffer interpreted WAL records in user space to help with serving
-subscriptions that are lagging behind the current position of the reader.
-
-Counterpoints:
-* Safekeepers serve many thousands of timelines and allocating a buffer for each might be wasteful,
-especially given that it would go unused on the happy path.
-* WAL is buffered in the kernel page cache. Usually we'd only pay the CPU cost of decoding and interpreting.
-
-### Tweaking the Pagserver Safekeeper Selection Algorithm
-
-We could make the pageserver aware of which safekeeper's already host shards for the timeline along
-with their current WAL positions. The pageserver should then prefer safkeepers that are in the same
-AZ _and_ already have a shard with a position close to the desired start position.
-
-We currently run one safekeeper per AZ, so the point is mute until that changes.
-
-### Pipelining first ingest phase
-
-The first ingest phase is a stateless transformation of a binary WAL record into a pre-processed
-output per shard.  To put multiple CPUs to work, we may pipeline this processing up to some defined buffer
-depth.
-
-## Alternatives considered
-
-### Give safekeepers enough state to fully decode WAL
-
-In this RFC, we only do the first phase of ingest on the safekeeper, because this is
-the phase that is stateless.  Subsequent changes then happen on the pageserver, with
-access to the `Timeline` state.
-
-We could do more work on the safekeeper if we transmitted metadata state to the safekeeper
-when subscribing to the WAL: for example, by telling the safekeeper all the relation sizes,
-so that it could then generate all the metadata writes for relation sizes.
-
-We avoid doing this for several reasons:
-1. Complexity: it's a more invasive protocol change
-2. Decoupling: having the safekeeper understand the `ProcessedWalIngest` already somewhat
-   infects it with knowledge of the pageserver, but this is mainly an abstract structure
-   that describes postgres writes.  However, if we taught the safekeeper about the exact
-   way that pageserver deals with metadata keys, this would be a much tighter coupling.
-3. Load: once the WAL has been processed to the point that it can be split between shards,
-   it is preferable to share out work on the remaining shards rather than adding extra CPU
-   load to the safekeeper.
-
-### Do pre-processing on the compute instead of the safekeeper
-
-Since our first stage of ingest is stateless, it could be done at any stage in the pipeline,
-all the way up to the compute.
-
-We choose not to do this, because it is useful for the safekeeper to store the raw WAL rather
-than just the preprocessed WAL:
-- The safekeeper still needs to be able to serve raw WAL back to postgres for e.g. physical replication
-- It simplifies our paxos implementation to have the offset in the write log be literally
-  the same as the LSN
-- Raw WAL must have a stable protocol since we might have to re-ingest it at arbitrary points in the future.
-  Storing raw WAL give us more flexibility to evolve the pageserver, safekeeper protocol.
-
-### Do wal pre-processing on shard 0 or a separate service, send it to other shards from there
-
-If we wanted to keep the safekeepers as entirely pure stores of raw WAL bytes, then
-we could do the initial decode and shard-splitting in some other location:
-- Shard 0 could subscribe to the full WAL and then send writes to other shards
-- A new intermediate service between the safekeeper and pageserver could do the splitting.
-
-So why not?
-- Extra network hop from shard 0 to the final destination shard
-- Clearly there is more infrastructure involved here compared with doing it inline on the safekeeper.
-- Safekeepers already have very light CPU load: typical cloud instances shapes with appropriate
-  disks for the safekeepers effectively have "free" CPU resources.
-- Doing extra work on shard 0 would complicate scheduling of shards on pageservers, because
-  shard 0 would have significantly higher CPU load under write workloads than other shards.

libs/pageserver_api/src/controller_api.rs

@@ -324,7 +324,7 @@ impl From<NodeSchedulingPolicy> for String {
 #[derive(Serialize, Deserialize, Clone, Copy, Eq, PartialEq, Debug)]
 pub enum SkSchedulingPolicy {
     Active,
-    Pause,
+    Disabled,
     Decomissioned,
 }
 
@@ -334,13 +334,9 @@ impl FromStr for SkSchedulingPolicy {
     fn from_str(s: &str) -> Result<Self, Self::Err> {
         Ok(match s {
             "active" => Self::Active,
-            "pause" => Self::Pause,
+            "disabled" => Self::Disabled,
             "decomissioned" => Self::Decomissioned,
-            _ => {
-                return Err(anyhow::anyhow!(
-                    "Unknown scheduling policy '{s}', try active,pause,decomissioned"
-                ))
-            }
+            _ => return Err(anyhow::anyhow!("Unknown scheduling state '{s}'")),
         })
     }
 }
@@ -350,7 +346,7 @@ impl From<SkSchedulingPolicy> for String {
         use SkSchedulingPolicy::*;
         match value {
             Active => "active",
-            Pause => "pause",
+            Disabled => "disabled",
             Decomissioned => "decomissioned",
         }
         .to_string()

libs/pageserver_api/src/models.rs

@@ -33,6 +33,7 @@ use crate::{
     reltag::RelTag,
     shard::{ShardCount, ShardStripeSize, TenantShardId},
 };
+use anyhow::bail;
 use bytes::{Buf, BufMut, Bytes, BytesMut};
 
 /// The state of a tenant in this pageserver.
@@ -1399,8 +1400,6 @@ pub enum PagestreamFeMessage {
     GetPage(PagestreamGetPageRequest),
     DbSize(PagestreamDbSizeRequest),
     GetSlruSegment(PagestreamGetSlruSegmentRequest),
-    #[cfg(feature = "testing")]
-    Test(PagestreamTestRequest),
 }
 
 // Wrapped in libpq CopyData
@@ -1412,22 +1411,6 @@ pub enum PagestreamBeMessage {
     Error(PagestreamErrorResponse),
     DbSize(PagestreamDbSizeResponse),
     GetSlruSegment(PagestreamGetSlruSegmentResponse),
-    #[cfg(feature = "testing")]
-    Test(PagestreamTestResponse),
-}
-
-// Keep in sync with `pagestore_client.h`
-#[repr(u8)]
-enum PagestreamFeMessageTag {
-    Exists = 0,
-    Nblocks = 1,
-    GetPage = 2,
-    DbSize = 3,
-    GetSlruSegment = 4,
-    /* future tags above this line */
-    /// For testing purposes, not available in production.
-    #[cfg(feature = "testing")]
-    Test = 99,
 }
 
 // Keep in sync with `pagestore_client.h`
@@ -1439,28 +1422,7 @@ enum PagestreamBeMessageTag {
     Error = 103,
     DbSize = 104,
     GetSlruSegment = 105,
-    /* future tags above this line */
-    /// For testing purposes, not available in production.
-    #[cfg(feature = "testing")]
-    Test = 199,
 }
-
-impl TryFrom<u8> for PagestreamFeMessageTag {
-    type Error = u8;
-    fn try_from(value: u8) -> Result<Self, u8> {
-        match value {
-            0 => Ok(PagestreamFeMessageTag::Exists),
-            1 => Ok(PagestreamFeMessageTag::Nblocks),
-            2 => Ok(PagestreamFeMessageTag::GetPage),
-            3 => Ok(PagestreamFeMessageTag::DbSize),
-            4 => Ok(PagestreamFeMessageTag::GetSlruSegment),
-            #[cfg(feature = "testing")]
-            99 => Ok(PagestreamFeMessageTag::Test),
-            _ => Err(value),
-        }
-    }
-}
-
 impl TryFrom<u8> for PagestreamBeMessageTag {
     type Error = u8;
     fn try_from(value: u8) -> Result<Self, u8> {
@@ -1471,8 +1433,6 @@ impl TryFrom<u8> for PagestreamBeMessageTag {
             103 => Ok(PagestreamBeMessageTag::Error),
             104 => Ok(PagestreamBeMessageTag::DbSize),
             105 => Ok(PagestreamBeMessageTag::GetSlruSegment),
-            #[cfg(feature = "testing")]
-            199 => Ok(PagestreamBeMessageTag::Test),
             _ => Err(value),
         }
     }
@@ -1590,20 +1550,6 @@ pub struct PagestreamDbSizeResponse {
     pub db_size: i64,
 }
 
-#[cfg(feature = "testing")]
-#[derive(Debug, PartialEq, Eq, Clone)]
-pub struct PagestreamTestRequest {
-    pub hdr: PagestreamRequest,
-    pub batch_key: u64,
-    pub message: String,
-}
-
-#[cfg(feature = "testing")]
-#[derive(Debug)]
-pub struct PagestreamTestResponse {
-    pub req: PagestreamTestRequest,
-}
-
 // This is a cut-down version of TenantHistorySize from the pageserver crate, omitting fields
 // that require pageserver-internal types.  It is sufficient to get the total size.
 #[derive(Serialize, Deserialize, Debug)]
@@ -1623,7 +1569,7 @@ impl PagestreamFeMessage {
 
         match self {
             Self::Exists(req) => {
-                bytes.put_u8(PagestreamFeMessageTag::Exists as u8);
+                bytes.put_u8(0);
                 bytes.put_u64(req.hdr.reqid);
                 bytes.put_u64(req.hdr.request_lsn.0);
                 bytes.put_u64(req.hdr.not_modified_since.0);
@@ -1634,7 +1580,7 @@ impl PagestreamFeMessage {
             }
 
             Self::Nblocks(req) => {
-                bytes.put_u8(PagestreamFeMessageTag::Nblocks as u8);
+                bytes.put_u8(1);
                 bytes.put_u64(req.hdr.reqid);
                 bytes.put_u64(req.hdr.request_lsn.0);
                 bytes.put_u64(req.hdr.not_modified_since.0);
@@ -1645,7 +1591,7 @@ impl PagestreamFeMessage {
             }
 
             Self::GetPage(req) => {
-                bytes.put_u8(PagestreamFeMessageTag::GetPage as u8);
+                bytes.put_u8(2);
                 bytes.put_u64(req.hdr.reqid);
                 bytes.put_u64(req.hdr.request_lsn.0);
                 bytes.put_u64(req.hdr.not_modified_since.0);
@@ -1657,7 +1603,7 @@ impl PagestreamFeMessage {
             }
 
             Self::DbSize(req) => {
-                bytes.put_u8(PagestreamFeMessageTag::DbSize as u8);
+                bytes.put_u8(3);
                 bytes.put_u64(req.hdr.reqid);
                 bytes.put_u64(req.hdr.request_lsn.0);
                 bytes.put_u64(req.hdr.not_modified_since.0);
@@ -1665,24 +1611,13 @@ impl PagestreamFeMessage {
             }
 
             Self::GetSlruSegment(req) => {
-                bytes.put_u8(PagestreamFeMessageTag::GetSlruSegment as u8);
+                bytes.put_u8(4);
                 bytes.put_u64(req.hdr.reqid);
                 bytes.put_u64(req.hdr.request_lsn.0);
                 bytes.put_u64(req.hdr.not_modified_since.0);
                 bytes.put_u8(req.kind);
                 bytes.put_u32(req.segno);
             }
-            #[cfg(feature = "testing")]
-            Self::Test(req) => {
-                bytes.put_u8(PagestreamFeMessageTag::Test as u8);
-                bytes.put_u64(req.hdr.reqid);
-                bytes.put_u64(req.hdr.request_lsn.0);
-                bytes.put_u64(req.hdr.not_modified_since.0);
-                bytes.put_u64(req.batch_key);
-                let message = req.message.as_bytes();
-                bytes.put_u64(message.len() as u64);
-                bytes.put_slice(message);
-            }
         }
 
         bytes.into()
@@ -1710,66 +1645,56 @@ impl PagestreamFeMessage {
             ),
         };
 
-        match PagestreamFeMessageTag::try_from(msg_tag)
-            .map_err(|tag: u8| anyhow::anyhow!("invalid tag {tag}"))?
-        {
-            PagestreamFeMessageTag::Exists => {
-                Ok(PagestreamFeMessage::Exists(PagestreamExistsRequest {
-                    hdr: PagestreamRequest {
-                        reqid,
-                        request_lsn,
-                        not_modified_since,
-                    },
-                    rel: RelTag {
-                        spcnode: body.read_u32::<BigEndian>()?,
-                        dbnode: body.read_u32::<BigEndian>()?,
-                        relnode: body.read_u32::<BigEndian>()?,
-                        forknum: body.read_u8()?,
-                    },
-                }))
-            }
-            PagestreamFeMessageTag::Nblocks => {
-                Ok(PagestreamFeMessage::Nblocks(PagestreamNblocksRequest {
-                    hdr: PagestreamRequest {
-                        reqid,
-                        request_lsn,
-                        not_modified_since,
-                    },
-                    rel: RelTag {
-                        spcnode: body.read_u32::<BigEndian>()?,
-                        dbnode: body.read_u32::<BigEndian>()?,
-                        relnode: body.read_u32::<BigEndian>()?,
-                        forknum: body.read_u8()?,
-                    },
-                }))
-            }
-            PagestreamFeMessageTag::GetPage => {
-                Ok(PagestreamFeMessage::GetPage(PagestreamGetPageRequest {
-                    hdr: PagestreamRequest {
-                        reqid,
-                        request_lsn,
-                        not_modified_since,
-                    },
-                    rel: RelTag {
-                        spcnode: body.read_u32::<BigEndian>()?,
-                        dbnode: body.read_u32::<BigEndian>()?,
-                        relnode: body.read_u32::<BigEndian>()?,
-                        forknum: body.read_u8()?,
-                    },
-                    blkno: body.read_u32::<BigEndian>()?,
-                }))
-            }
-            PagestreamFeMessageTag::DbSize => {
-                Ok(PagestreamFeMessage::DbSize(PagestreamDbSizeRequest {
-                    hdr: PagestreamRequest {
-                        reqid,
-                        request_lsn,
-                        not_modified_since,
-                    },
+        match msg_tag {
+            0 => Ok(PagestreamFeMessage::Exists(PagestreamExistsRequest {
+                hdr: PagestreamRequest {
+                    reqid,
+                    request_lsn,
+                    not_modified_since,
+                },
+                rel: RelTag {
+                    spcnode: body.read_u32::<BigEndian>()?,
                     dbnode: body.read_u32::<BigEndian>()?,
-                }))
-            }
-            PagestreamFeMessageTag::GetSlruSegment => Ok(PagestreamFeMessage::GetSlruSegment(
+                    relnode: body.read_u32::<BigEndian>()?,
+                    forknum: body.read_u8()?,
+                },
+            })),
+            1 => Ok(PagestreamFeMessage::Nblocks(PagestreamNblocksRequest {
+                hdr: PagestreamRequest {
+                    reqid,
+                    request_lsn,
+                    not_modified_since,
+                },
+                rel: RelTag {
+                    spcnode: body.read_u32::<BigEndian>()?,
+                    dbnode: body.read_u32::<BigEndian>()?,
+                    relnode: body.read_u32::<BigEndian>()?,
+                    forknum: body.read_u8()?,
+                },
+            })),
+            2 => Ok(PagestreamFeMessage::GetPage(PagestreamGetPageRequest {
+                hdr: PagestreamRequest {
+                    reqid,
+                    request_lsn,
+                    not_modified_since,
+                },
+                rel: RelTag {
+                    spcnode: body.read_u32::<BigEndian>()?,
+                    dbnode: body.read_u32::<BigEndian>()?,
+                    relnode: body.read_u32::<BigEndian>()?,
+                    forknum: body.read_u8()?,
+                },
+                blkno: body.read_u32::<BigEndian>()?,
+            })),
+            3 => Ok(PagestreamFeMessage::DbSize(PagestreamDbSizeRequest {
+                hdr: PagestreamRequest {
+                    reqid,
+                    request_lsn,
+                    not_modified_since,
+                },
+                dbnode: body.read_u32::<BigEndian>()?,
+            })),
+            4 => Ok(PagestreamFeMessage::GetSlruSegment(
                 PagestreamGetSlruSegmentRequest {
                     hdr: PagestreamRequest {
                         reqid,
@@ -1780,21 +1705,7 @@ impl PagestreamFeMessage {
                     segno: body.read_u32::<BigEndian>()?,
                 },
             )),
-            #[cfg(feature = "testing")]
-            PagestreamFeMessageTag::Test => Ok(PagestreamFeMessage::Test(PagestreamTestRequest {
-                hdr: PagestreamRequest {
-                    reqid,
-                    request_lsn,
-                    not_modified_since,
-                },
-                batch_key: body.read_u64::<BigEndian>()?,
-                message: {
-                    let len = body.read_u64::<BigEndian>()?;
-                    let mut buf = vec![0; len as usize];
-                    body.read_exact(&mut buf)?;
-                    String::from_utf8(buf)?
-                },
-            })),
+            _ => bail!("unknown smgr message tag: {:?}", msg_tag),
         }
     }
 }
@@ -1837,15 +1748,6 @@ impl PagestreamBeMessage {
                         bytes.put_u32((resp.segment.len() / BLCKSZ as usize) as u32);
                         bytes.put(&resp.segment[..]);
                     }
-
-                    #[cfg(feature = "testing")]
-                    Self::Test(resp) => {
-                        bytes.put_u8(Tag::Test as u8);
-                        bytes.put_u64(resp.req.batch_key);
-                        let message = resp.req.message.as_bytes();
-                        bytes.put_u64(message.len() as u64);
-                        bytes.put_slice(message);
-                    }
                 }
             }
             PagestreamProtocolVersion::V3 => {
@@ -1914,18 +1816,6 @@ impl PagestreamBeMessage {
                         bytes.put_u32((resp.segment.len() / BLCKSZ as usize) as u32);
                         bytes.put(&resp.segment[..]);
                     }
-
-                    #[cfg(feature = "testing")]
-                    Self::Test(resp) => {
-                        bytes.put_u8(Tag::Test as u8);
-                        bytes.put_u64(resp.req.hdr.reqid);
-                        bytes.put_u64(resp.req.hdr.request_lsn.0);
-                        bytes.put_u64(resp.req.hdr.not_modified_since.0);
-                        bytes.put_u64(resp.req.batch_key);
-                        let message = resp.req.message.as_bytes();
-                        bytes.put_u64(message.len() as u64);
-                        bytes.put_slice(message);
-                    }
                 }
             }
         }
@@ -2068,28 +1958,6 @@ impl PagestreamBeMessage {
                         segment: segment.into(),
                     })
                 }
-                #[cfg(feature = "testing")]
-                Tag::Test => {
-                    let reqid = buf.read_u64::<BigEndian>()?;
-                    let request_lsn = Lsn(buf.read_u64::<BigEndian>()?);
-                    let not_modified_since = Lsn(buf.read_u64::<BigEndian>()?);
-                    let batch_key = buf.read_u64::<BigEndian>()?;
-                    let len = buf.read_u64::<BigEndian>()?;
-                    let mut msg = vec![0; len as usize];
-                    buf.read_exact(&mut msg)?;
-                    let message = String::from_utf8(msg)?;
-                    Self::Test(PagestreamTestResponse {
-                        req: PagestreamTestRequest {
-                            hdr: PagestreamRequest {
-                                reqid,
-                                request_lsn,
-                                not_modified_since,
-                            },
-                            batch_key,
-                            message,
-                        },
-                    })
-                }
             };
         let remaining = buf.into_inner();
         if !remaining.is_empty() {
@@ -2109,8 +1977,6 @@ impl PagestreamBeMessage {
             Self::Error(_) => "Error",
             Self::DbSize(_) => "DbSize",
             Self::GetSlruSegment(_) => "GetSlruSegment",
-            #[cfg(feature = "testing")]
-            Self::Test(_) => "Test",
         }
     }
 }

libs/safekeeper_api/src/models.rs

@@ -277,8 +277,3 @@ pub struct TimelineTermBumpResponse {
     pub previous_term: u64,
     pub current_term: u64,
 }
-
-#[derive(Debug, Clone, Deserialize, Serialize)]
-pub struct SafekeeperUtilization {
-    pub timeline_count: u64,
-}

libs/utils/src/guard_arc_swap.rs

@@ -1,54 +0,0 @@
-//! A wrapper around `ArcSwap` that ensures there is only one writer at a time and writes
-//! don't block reads.
-
-use arc_swap::ArcSwap;
-use std::sync::Arc;
-use tokio::sync::TryLockError;
-
-pub struct GuardArcSwap<T> {
-    inner: ArcSwap<T>,
-    guard: tokio::sync::Mutex<()>,
-}
-
-pub struct Guard<'a, T> {
-    _guard: tokio::sync::MutexGuard<'a, ()>,
-    inner: &'a ArcSwap<T>,
-}
-
-impl<T> GuardArcSwap<T> {
-    pub fn new(inner: T) -> Self {
-        Self {
-            inner: ArcSwap::new(Arc::new(inner)),
-            guard: tokio::sync::Mutex::new(()),
-        }
-    }
-
-    pub fn read(&self) -> Arc<T> {
-        self.inner.load_full()
-    }
-
-    pub async fn write_guard(&self) -> Guard<'_, T> {
-        Guard {
-            _guard: self.guard.lock().await,
-            inner: &self.inner,
-        }
-    }
-
-    pub fn try_write_guard(&self) -> Result<Guard<'_, T>, TryLockError> {
-        let guard = self.guard.try_lock()?;
-        Ok(Guard {
-            _guard: guard,
-            inner: &self.inner,
-        })
-    }
-}
-
-impl<T> Guard<'_, T> {
-    pub fn read(&self) -> Arc<T> {
-        self.inner.load_full()
-    }
-
-    pub fn write(&mut self, value: T) {
-        self.inner.store(Arc::new(value));
-    }
-}

libs/utils/src/lib.rs

@@ -98,8 +98,6 @@ pub mod try_rcu;
 
 pub mod pprof;
 
-pub mod guard_arc_swap;
-
 // Re-export used in macro. Avoids adding git-version as dep in target crates.
 #[doc(hidden)]
 pub use git_version;

pageserver/Cargo.toml

@@ -8,7 +8,7 @@ license.workspace = true
 default = []
 # Enables test-only APIs, incuding failpoints. In particular, enables the `fail_point!` macro,
 # which adds some runtime cost to run tests on outage conditions
-testing = ["fail/failpoints", "pageserver_api/testing", "wal_decoder/testing", "pageserver_client/testing"]
+testing = ["fail/failpoints", "pageserver_api/testing", "wal_decoder/testing"]
 
 [dependencies]
 anyhow.workspace = true
@@ -114,7 +114,3 @@ harness = false
 [[bench]]
 name = "upload_queue"
 harness = false
-
-[[bin]]
-name = "test_helper_slow_client_reads"
-required-features = [ "testing" ]

pageserver/client/Cargo.toml

@@ -4,9 +4,6 @@ version = "0.1.0"
 edition.workspace = true
 license.workspace = true
 
-[features]
-testing = [ "pageserver_api/testing" ]
-
 [dependencies]
 pageserver_api.workspace = true
 thiserror.workspace = true

pageserver/client/src/page_service.rs

@@ -1,9 +1,6 @@
-use std::sync::{Arc, Mutex};
+use std::pin::Pin;
 
-use futures::{
-    stream::{SplitSink, SplitStream},
-    SinkExt, StreamExt,
-};
+use futures::SinkExt;
 use pageserver_api::{
     models::{
         PagestreamBeMessage, PagestreamFeMessage, PagestreamGetPageRequest,
@@ -13,6 +10,7 @@ use pageserver_api::{
 };
 use tokio::task::JoinHandle;
 use tokio_postgres::CopyOutStream;
+use tokio_stream::StreamExt;
 use tokio_util::sync::CancellationToken;
 use utils::{
     id::{TenantId, TimelineId},
@@ -64,28 +62,15 @@ impl Client {
             .client
             .copy_both_simple(&format!("pagestream_v3 {tenant_id} {timeline_id}"))
             .await?;
-        let (sink, stream) = copy_both.split(); // TODO: actually support splitting of the CopyBothDuplex so the lock inside this split adaptor goes away.
         let Client {
             cancel_on_client_drop,
             conn_task,
             client: _,
         } = self;
-        let shared = Arc::new(Mutex::new(PagestreamShared::ConnTaskRunning(
-            ConnTaskRunning {
-                cancel_on_client_drop,
-                conn_task,
-            },
-        )));
         Ok(PagestreamClient {
-            sink: PagestreamSender {
-                shared: shared.clone(),
-                sink,
-            },
-            stream: PagestreamReceiver {
-                shared: shared.clone(),
-                stream,
-            },
-            shared,
+            copy_both: Box::pin(copy_both),
+            conn_task,
+            cancel_on_client_drop,
         })
     }
 
@@ -112,28 +97,7 @@ impl Client {
 
 /// Create using [`Client::pagestream`].
 pub struct PagestreamClient {
-    shared: Arc<Mutex<PagestreamShared>>,
-    sink: PagestreamSender,
-    stream: PagestreamReceiver,
-}
-
-pub struct PagestreamSender {
-    #[allow(dead_code)]
-    shared: Arc<Mutex<PagestreamShared>>,
-    sink: SplitSink<tokio_postgres::CopyBothDuplex<bytes::Bytes>, bytes::Bytes>,
-}
-
-pub struct PagestreamReceiver {
-    #[allow(dead_code)]
-    shared: Arc<Mutex<PagestreamShared>>,
-    stream: SplitStream<tokio_postgres::CopyBothDuplex<bytes::Bytes>>,
-}
-
-enum PagestreamShared {
-    ConnTaskRunning(ConnTaskRunning),
-    ConnTaskCancelledJoinHandleReturnedOrDropped,
-}
-struct ConnTaskRunning {
+    copy_both: Pin<Box<tokio_postgres::CopyBothDuplex<bytes::Bytes>>>,
     cancel_on_client_drop: Option<tokio_util::sync::DropGuard>,
     conn_task: JoinHandle<()>,
 }
@@ -146,11 +110,11 @@ pub struct RelTagBlockNo {
 impl PagestreamClient {
     pub async fn shutdown(self) {
         let Self {
-            shared,
-            sink,
-            stream,
-        } = { self };
-        // The `copy_both` split into `sink` and `stream` contains internal channel sender, the receiver of which is polled by `conn_task`.
+            copy_both,
+            cancel_on_client_drop: cancel_conn_task,
+            conn_task,
+        } = self;
+        // The `copy_both` contains internal channel sender, the receiver of which is polled by `conn_task`.
         // When `conn_task` observes the sender has been dropped, it sends a `FeMessage::CopyFail` into the connection.
         // (see https://github.com/neondatabase/rust-postgres/blob/2005bf79573b8add5cf205b52a2b208e356cc8b0/tokio-postgres/src/copy_both.rs#L56).
         //
@@ -167,77 +131,27 @@ impl PagestreamClient {
         //
         // NB: page_service doesn't have a use case to exit the `pagestream` mode currently.
         // => https://github.com/neondatabase/neon/issues/6390
-        let ConnTaskRunning {
-            cancel_on_client_drop,
-            conn_task,
-        } = {
-            let mut guard = shared.lock().unwrap();
-            match std::mem::replace(
-                &mut *guard,
-                PagestreamShared::ConnTaskCancelledJoinHandleReturnedOrDropped,
-            ) {
-                PagestreamShared::ConnTaskRunning(conn_task_running) => conn_task_running,
-                PagestreamShared::ConnTaskCancelledJoinHandleReturnedOrDropped => unreachable!(),
-            }
-        };
-        let _ = cancel_on_client_drop.unwrap();
+        let _ = cancel_conn_task.unwrap();
         conn_task.await.unwrap();
-
-        // Now drop the split copy_both.
-        drop(sink);
-        drop(stream);
-    }
-
-    pub fn split(self) -> (PagestreamSender, PagestreamReceiver) {
-        let Self {
-            shared: _,
-            sink,
-            stream,
-        } = self;
-        (sink, stream)
+        drop(copy_both);
     }
 
     pub async fn getpage(
         &mut self,
         req: PagestreamGetPageRequest,
     ) -> anyhow::Result<PagestreamGetPageResponse> {
-        self.getpage_send(req).await?;
-        self.getpage_recv().await
-    }
+        let req = PagestreamFeMessage::GetPage(req);
+        let req: bytes::Bytes = req.serialize();
+        // let mut req = tokio_util::io::ReaderStream::new(&req);
+        let mut req = tokio_stream::once(Ok(req));
 
-    pub async fn getpage_send(&mut self, req: PagestreamGetPageRequest) -> anyhow::Result<()> {
-        self.sink.getpage_send(req).await
-    }
+        self.copy_both.send_all(&mut req).await?;
 
-    pub async fn getpage_recv(&mut self) -> anyhow::Result<PagestreamGetPageResponse> {
-        self.stream.getpage_recv().await
-    }
-}
-
-impl PagestreamSender {
-    // TODO: maybe make this impl Sink instead for better composability?
-    pub async fn send(&mut self, msg: PagestreamFeMessage) -> anyhow::Result<()> {
-        let msg = msg.serialize();
-        self.sink.send_all(&mut tokio_stream::once(Ok(msg))).await?;
-        Ok(())
-    }
-
-    pub async fn getpage_send(&mut self, req: PagestreamGetPageRequest) -> anyhow::Result<()> {
-        self.send(PagestreamFeMessage::GetPage(req)).await
-    }
-}
-
-impl PagestreamReceiver {
-    // TODO: maybe make this impl Stream instead for better composability?
-    pub async fn recv(&mut self) -> anyhow::Result<PagestreamBeMessage> {
-        let next: Option<Result<bytes::Bytes, _>> = self.stream.next().await;
+        let next: Option<Result<bytes::Bytes, _>> = self.copy_both.next().await;
         let next: bytes::Bytes = next.unwrap()?;
-        PagestreamBeMessage::deserialize(next)
-    }
 
-    pub async fn getpage_recv(&mut self) -> anyhow::Result<PagestreamGetPageResponse> {
-        let next: PagestreamBeMessage = self.recv().await?;
-        match next {
+        let msg = PagestreamBeMessage::deserialize(next)?;
+        match msg {
             PagestreamBeMessage::GetPage(p) => Ok(p),
             PagestreamBeMessage::Error(e) => anyhow::bail!("Error: {:?}", e),
             PagestreamBeMessage::Exists(_)
@@ -246,14 +160,7 @@ impl PagestreamReceiver {
             | PagestreamBeMessage::GetSlruSegment(_) => {
                 anyhow::bail!(
                     "unexpected be message kind in response to getpage request: {}",
-                    next.kind()
-                )
-            }
-            #[cfg(feature = "testing")]
-            PagestreamBeMessage::Test(_) => {
-                anyhow::bail!(
-                    "unexpected be message kind in response to getpage request: {}",
-                    next.kind()
+                    msg.kind()
                 )
             }
         }

pageserver/src/bin/test_helper_slow_client_reads.rs

@@ -1,65 +0,0 @@
-use std::{
-    io::{stdin, stdout, Read, Write},
-    time::Duration,
-};
-
-use clap::Parser;
-use pageserver_api::models::{PagestreamRequest, PagestreamTestRequest};
-use utils::{
-    id::{TenantId, TimelineId},
-    lsn::Lsn,
-};
-
-#[derive(clap::Parser)]
-struct Args {
-    connstr: String,
-    tenant_id: TenantId,
-    timeline_id: TimelineId,
-}
-
-#[tokio::main]
-async fn main() -> anyhow::Result<()> {
-    let Args {
-        connstr,
-        tenant_id,
-        timeline_id,
-    } = Args::parse();
-    let client = pageserver_client::page_service::Client::new(connstr).await?;
-    let client = client.pagestream(tenant_id, timeline_id).await?;
-    let (mut sender, _receiver) = client.split();
-
-    eprintln!("filling the pipe");
-    let mut msg = 0;
-    loop {
-        msg += 1;
-        let fut = sender.send(pageserver_api::models::PagestreamFeMessage::Test(
-            PagestreamTestRequest {
-                hdr: PagestreamRequest {
-                    reqid: 0,
-                    request_lsn: Lsn(23),
-                    not_modified_since: Lsn(23),
-                },
-                batch_key: 42,
-                message: format!("message {}", msg),
-            },
-        ));
-        let Ok(res) = tokio::time::timeout(Duration::from_secs(10), fut).await else {
-            eprintln!("pipe seems full");
-            break;
-        };
-        let _: () = res?;
-    }
-
-    let n = stdout().write(b"R")?;
-    assert_eq!(n, 1);
-    stdout().flush()?;
-
-    eprintln!("waiting for signal to tell us to exit");
-
-    let mut buf = [0u8; 1];
-    stdin().read_exact(&mut buf)?;
-
-    eprintln!("termination signal received, exiting");
-
-    anyhow::Ok(())
-}

pageserver/src/config.rs

@@ -109,13 +109,13 @@ pub struct PageServerConf {
     /// A lower value implicitly deprioritizes loading such tenants, vs. other work in the system.
     pub concurrent_tenant_warmup: ConfigurableSemaphore,
 
-    /// Number of concurrent [`Tenant::gather_size_inputs`](crate::tenant::Tenant::gather_size_inputs) allowed.
+    /// Number of concurrent [`TenantShard::gather_size_inputs`](crate::tenant::TenantShard::gather_size_inputs) allowed.
     pub concurrent_tenant_size_logical_size_queries: ConfigurableSemaphore,
-    /// Limit of concurrent [`Tenant::gather_size_inputs`] issued by module `eviction_task`.
+    /// Limit of concurrent [`TenantShard::gather_size_inputs`] issued by module `eviction_task`.
     /// The number of permits is the same as `concurrent_tenant_size_logical_size_queries`.
     /// See the comment in `eviction_task` for details.
     ///
-    /// [`Tenant::gather_size_inputs`]: crate::tenant::Tenant::gather_size_inputs
+    /// [`TenantShard::gather_size_inputs`]: crate::tenant::TenantShard::gather_size_inputs
     pub eviction_task_immitated_concurrent_logical_size_queries: ConfigurableSemaphore,
 
     // How often to collect metrics and send them to the metrics endpoint.
@@ -509,10 +509,10 @@ impl ConfigurableSemaphore {
     /// Initializse using a non-zero amount of permits.
     ///
     /// Require a non-zero initial permits, because using permits == 0 is a crude way to disable a
-    /// feature such as [`Tenant::gather_size_inputs`]. Otherwise any semaphore using future will
+    /// feature such as [`TenantShard::gather_size_inputs`]. Otherwise any semaphore using future will
     /// behave like [`futures::future::pending`], just waiting until new permits are added.
     ///
-    /// [`Tenant::gather_size_inputs`]: crate::tenant::Tenant::gather_size_inputs
+    /// [`TenantShard::gather_size_inputs`]: crate::tenant::TenantShard::gather_size_inputs
     pub fn new(initial_permits: NonZeroUsize) -> Self {
         ConfigurableSemaphore {
             initial_permits,

pageserver/src/consumption_metrics.rs

@@ -7,7 +7,7 @@ use crate::context::{DownloadBehavior, RequestContext};
 use crate::task_mgr::{self, TaskKind, BACKGROUND_RUNTIME};
 use crate::tenant::size::CalculateSyntheticSizeError;
 use crate::tenant::tasks::BackgroundLoopKind;
-use crate::tenant::{mgr::TenantManager, LogicalSizeCalculationCause, Tenant};
+use crate::tenant::{mgr::TenantShardManager, LogicalSizeCalculationCause, TenantShard};
 use camino::Utf8PathBuf;
 use consumption_metrics::EventType;
 use itertools::Itertools as _;
@@ -95,7 +95,7 @@ type Cache = HashMap<MetricsKey, NewRawMetric>;
 
 pub async fn run(
     conf: &'static PageServerConf,
-    tenant_manager: Arc<TenantManager>,
+    tenant_manager: Arc<TenantShardManager>,
     cancel: CancellationToken,
 ) {
     let Some(metric_collection_endpoint) = conf.metric_collection_endpoint.as_ref() else {
@@ -150,7 +150,7 @@ pub async fn run(
 /// Main thread that serves metrics collection
 #[allow(clippy::too_many_arguments)]
 async fn collect_metrics(
-    tenant_manager: Arc<TenantManager>,
+    tenant_manager: Arc<TenantShardManager>,
     metric_collection_endpoint: &Url,
     metric_collection_bucket: &Option<RemoteStorageConfig>,
     metric_collection_interval: Duration,
@@ -362,7 +362,7 @@ async fn reschedule(
 
 /// Caclculate synthetic size for each active tenant
 async fn calculate_synthetic_size_worker(
-    tenant_manager: Arc<TenantManager>,
+    tenant_manager: Arc<TenantShardManager>,
     synthetic_size_calculation_interval: Duration,
     cancel: CancellationToken,
     ctx: RequestContext,
@@ -425,7 +425,7 @@ async fn calculate_synthetic_size_worker(
     }
 }
 
-async fn calculate_and_log(tenant: &Tenant, cancel: &CancellationToken, ctx: &RequestContext) {
+async fn calculate_and_log(tenant: &TenantShard, cancel: &CancellationToken, ctx: &RequestContext) {
     const CAUSE: LogicalSizeCalculationCause =
         LogicalSizeCalculationCause::ConsumptionMetricsSyntheticSize;
 

pageserver/src/consumption_metrics/metrics.rs

@@ -1,4 +1,4 @@
-use crate::tenant::mgr::TenantManager;
+use crate::tenant::mgr::TenantShardManager;
 use crate::{context::RequestContext, tenant::timeline::logical_size::CurrentLogicalSize};
 use chrono::{DateTime, Utc};
 use consumption_metrics::EventType;
@@ -174,9 +174,9 @@ impl MetricsKey {
         .absolute_values()
     }
 
-    /// [`Tenant::remote_size`]
+    /// [`TenantShard::remote_size`]
     ///
-    /// [`Tenant::remote_size`]: crate::tenant::Tenant::remote_size
+    /// [`TenantShard::remote_size`]: crate::tenant::TenantShard::remote_size
     const fn remote_storage_size(tenant_id: TenantId) -> AbsoluteValueFactory {
         MetricsKey {
             tenant_id,
@@ -198,9 +198,9 @@ impl MetricsKey {
         .absolute_values()
     }
 
-    /// [`Tenant::cached_synthetic_size`] as refreshed by [`calculate_synthetic_size_worker`].
+    /// [`TenantShard::cached_synthetic_size`] as refreshed by [`calculate_synthetic_size_worker`].
     ///
-    /// [`Tenant::cached_synthetic_size`]: crate::tenant::Tenant::cached_synthetic_size
+    /// [`TenantShard::cached_synthetic_size`]: crate::tenant::TenantShard::cached_synthetic_size
     /// [`calculate_synthetic_size_worker`]: super::calculate_synthetic_size_worker
     const fn synthetic_size(tenant_id: TenantId) -> AbsoluteValueFactory {
         MetricsKey {
@@ -213,7 +213,7 @@ impl MetricsKey {
 }
 
 pub(super) async fn collect_all_metrics(
-    tenant_manager: &Arc<TenantManager>,
+    tenant_manager: &Arc<TenantShardManager>,
     cached_metrics: &Cache,
     ctx: &RequestContext,
 ) -> Vec<NewRawMetric> {
@@ -253,7 +253,7 @@ pub(super) async fn collect_all_metrics(
 
 async fn collect<S>(tenants: S, cache: &Cache, ctx: &RequestContext) -> Vec<NewRawMetric>
 where
-    S: futures::stream::Stream<Item = (TenantId, Arc<crate::tenant::Tenant>)>,
+    S: futures::stream::Stream<Item = (TenantId, Arc<crate::tenant::TenantShard>)>,
 {
     let mut current_metrics: Vec<NewRawMetric> = Vec::new();
 
@@ -307,7 +307,7 @@ impl TenantSnapshot {
     ///
     /// `resident_size` is calculated of the timelines we had access to for other metrics, so we
     /// cannot just list timelines here.
-    fn collect(t: &Arc<crate::tenant::Tenant>, resident_size: u64) -> Self {
+    fn collect(t: &Arc<crate::tenant::TenantShard>, resident_size: u64) -> Self {
         TenantSnapshot {
             resident_size,
             remote_size: t.remote_size(),

pageserver/src/deletion_queue.rs

@@ -703,7 +703,7 @@ mod test {
 
     use crate::{
         controller_upcall_client::RetryForeverError,
-        tenant::{harness::TenantHarness, storage_layer::DeltaLayerName},
+        tenant::{harness::TenantShardHarness, storage_layer::DeltaLayerName},
     };
 
     use super::*;
@@ -722,7 +722,7 @@ mod test {
     });
 
     struct TestSetup {
-        harness: TenantHarness,
+        harness: TenantShardHarness,
         remote_fs_dir: Utf8PathBuf,
         storage: GenericRemoteStorage,
         mock_control_plane: MockControlPlane,
@@ -825,7 +825,7 @@ mod test {
 
     async fn setup(test_name: &str) -> anyhow::Result<TestSetup> {
         let test_name = Box::leak(Box::new(format!("deletion_queue__{test_name}")));
-        let harness = TenantHarness::create(test_name).await?;
+        let harness = TenantShardHarness::create(test_name).await?;
 
         // We do not load() the harness: we only need its config and remote_storage
 

pageserver/src/disk_usage_eviction_task.rs

@@ -57,7 +57,7 @@ use crate::{
     metrics::disk_usage_based_eviction::METRICS,
     task_mgr::{self, BACKGROUND_RUNTIME},
     tenant::{
-        mgr::TenantManager,
+        mgr::TenantShardManager,
         remote_timeline_client::LayerFileMetadata,
         secondary::SecondaryTenant,
         storage_layer::{AsLayerDesc, EvictionError, Layer, LayerName, LayerVisibilityHint},
@@ -166,7 +166,7 @@ pub fn launch_disk_usage_global_eviction_task(
     conf: &'static PageServerConf,
     storage: GenericRemoteStorage,
     state: Arc<State>,
-    tenant_manager: Arc<TenantManager>,
+    tenant_manager: Arc<TenantShardManager>,
     background_jobs_barrier: completion::Barrier,
 ) -> Option<DiskUsageEvictionTask> {
     let Some(task_config) = &conf.disk_usage_based_eviction else {
@@ -203,7 +203,7 @@ async fn disk_usage_eviction_task(
     state: &State,
     task_config: &DiskUsageEvictionTaskConfig,
     storage: &GenericRemoteStorage,
-    tenant_manager: Arc<TenantManager>,
+    tenant_manager: Arc<TenantShardManager>,
     cancel: CancellationToken,
 ) {
     scopeguard::defer! {
@@ -265,7 +265,7 @@ async fn disk_usage_eviction_task_iteration(
     state: &State,
     task_config: &DiskUsageEvictionTaskConfig,
     storage: &GenericRemoteStorage,
-    tenant_manager: &Arc<TenantManager>,
+    tenant_manager: &Arc<TenantShardManager>,
     cancel: &CancellationToken,
 ) -> anyhow::Result<()> {
     let tenants_dir = tenant_manager.get_conf().tenants_path();
@@ -361,7 +361,7 @@ pub(crate) async fn disk_usage_eviction_task_iteration_impl<U: Usage>(
     state: &State,
     _storage: &GenericRemoteStorage,
     usage_pre: U,
-    tenant_manager: &Arc<TenantManager>,
+    tenant_manager: &Arc<TenantShardManager>,
     eviction_order: EvictionOrder,
     cancel: &CancellationToken,
 ) -> anyhow::Result<IterationOutcome<U>> {
@@ -788,7 +788,7 @@ enum EvictionCandidates {
 /// - tenant B 1 layer
 /// - tenant C 8 layers
 async fn collect_eviction_candidates(
-    tenant_manager: &Arc<TenantManager>,
+    tenant_manager: &Arc<TenantShardManager>,
     eviction_order: EvictionOrder,
     cancel: &CancellationToken,
 ) -> anyhow::Result<EvictionCandidates> {

pageserver/src/http/routes.rs

@@ -74,7 +74,7 @@ use crate::task_mgr::TaskKind;
 use crate::tenant::config::{LocationConf, TenantConfOpt};
 use crate::tenant::mgr::GetActiveTenantError;
 use crate::tenant::mgr::{
-    GetTenantError, TenantManager, TenantMapError, TenantMapInsertError, TenantSlotError,
+    GetTenantError, TenantMapError, TenantMapInsertError, TenantShardManager, TenantSlotError,
     TenantSlotUpsertError, TenantStateError,
 };
 use crate::tenant::mgr::{TenantSlot, UpsertLocationError};
@@ -130,7 +130,7 @@ pub(crate) const ACTIVE_TENANT_TIMEOUT: Duration = Duration::from_millis(30000);
 
 pub struct State {
     conf: &'static PageServerConf,
-    tenant_manager: Arc<TenantManager>,
+    tenant_manager: Arc<TenantShardManager>,
     auth: Option<Arc<SwappableJwtAuth>>,
     allowlist_routes: &'static [&'static str],
     remote_storage: GenericRemoteStorage,
@@ -145,7 +145,7 @@ impl State {
     #[allow(clippy::too_many_arguments)]
     pub fn new(
         conf: &'static PageServerConf,
-        tenant_manager: Arc<TenantManager>,
+        tenant_manager: Arc<TenantShardManager>,
         auth: Option<Arc<SwappableJwtAuth>>,
         remote_storage: GenericRemoteStorage,
         broker_client: storage_broker::BrokerClientChannel,
@@ -1761,7 +1761,7 @@ async fn update_tenant_config_handler(
         &ShardParameters::default(),
     );
 
-    crate::tenant::Tenant::persist_tenant_config(state.conf, &tenant_shard_id, &location_conf)
+    crate::tenant::TenantShard::persist_tenant_config(state.conf, &tenant_shard_id, &location_conf)
         .await
         .map_err(|e| ApiError::InternalServerError(anyhow::anyhow!(e)))?;
 
@@ -1802,7 +1802,7 @@ async fn patch_tenant_config_handler(
         &ShardParameters::default(),
     );
 
-    crate::tenant::Tenant::persist_tenant_config(state.conf, &tenant_shard_id, &location_conf)
+    crate::tenant::TenantShard::persist_tenant_config(state.conf, &tenant_shard_id, &location_conf)
         .await
         .map_err(|e| ApiError::InternalServerError(anyhow::anyhow!(e)))?;
 
@@ -2571,7 +2571,7 @@ async fn timeline_collect_keyspace(
 }
 
 async fn active_timeline_of_active_tenant(
-    tenant_manager: &TenantManager,
+    tenant_manager: &TenantShardManager,
     tenant_shard_id: TenantShardId,
     timeline_id: TimelineId,
 ) -> Result<Arc<Timeline>, ApiError> {
@@ -2923,33 +2923,6 @@ async fn get_utilization(
         .map_err(ApiError::InternalServerError)
 }
 
-async fn get_taskdump(
-    request: Request<Body>,
-    _cancel: CancellationToken,
-) -> Result<Response<Body>, ApiError> {
-    check_permission(&request, None)?;
-
-    use std::io::Write;
-    use tokio::runtime::Handle;
-
-    let out = async {
-        let mut out = Vec::new();
-        // Inside an async block or function.
-        let handle = Handle::current();
-        let dump = handle.dump().await;
-        for (i, task) in dump.tasks().iter().enumerate() {
-            let trace = task.trace();
-            writeln!(out, "TASK {i}:")?;
-            writeln!(out, "{trace}\n")?;
-        }
-        Ok(String::from_utf8(out).unwrap())
-    }
-    .await
-    .map_err(|e: anyhow::Error| ApiError::InternalServerError(e))?;
-
-    json_response(StatusCode::OK, out)
-}
-
 async fn list_aux_files(
     mut request: Request<Body>,
     _cancel: CancellationToken,
@@ -3635,9 +3608,5 @@ pub fn make_router(
             "/v1/tenant/:tenant_id/timeline/:timeline_id/import_wal",
             |r| api_handler(r, put_tenant_timeline_import_wal),
         )
-        .get(
-            "/debug/taskdump",
-            |r| api_handler(r,  get_taskdump),
-        )
         .any(handler_404))
 }

pageserver/src/lib.rs

@@ -36,7 +36,7 @@ pub mod walredo;
 use camino::Utf8Path;
 use deletion_queue::DeletionQueue;
 use tenant::{
-    mgr::{BackgroundPurges, TenantManager},
+    mgr::{BackgroundPurges, TenantShardManager},
     secondary,
 };
 use tracing::{info, info_span};
@@ -81,7 +81,7 @@ pub async fn shutdown_pageserver(
     page_service: page_service::Listener,
     consumption_metrics_worker: ConsumptionMetricsTasks,
     disk_usage_eviction_task: Option<DiskUsageEvictionTask>,
-    tenant_manager: &TenantManager,
+    tenant_manager: &TenantShardManager,
     background_purges: BackgroundPurges,
     mut deletion_queue: DeletionQueue,
     secondary_controller_tasks: secondary::GlobalTasks,

pageserver/src/metrics.rs

@@ -100,32 +100,6 @@ pub(crate) static VEC_READ_NUM_LAYERS_VISITED: Lazy<Histogram> = Lazy::new(|| {
     .expect("failed to define a metric")
 });
 
-pub(crate) static CONCURRENT_INITDBS: Lazy<UIntGauge> = Lazy::new(|| {
-    register_uint_gauge!(
-        "pageserver_concurrent_initdb",
-        "Number of initdb processes running"
-    )
-    .expect("failed to define a metric")
-});
-
-pub(crate) static INITDB_SEMAPHORE_ACQUISITION_TIME: Lazy<Histogram> = Lazy::new(|| {
-    register_histogram!(
-        "pageserver_initdb_semaphore_seconds_global",
-        "Time spent getting a permit from the global initdb semaphore",
-        STORAGE_OP_BUCKETS.into()
-    )
-    .expect("failed to define metric")
-});
-
-pub(crate) static INITDB_RUN_TIME: Lazy<Histogram> = Lazy::new(|| {
-    register_histogram!(
-        "pageserver_initdb_seconds_global",
-        "Time spent performing initdb",
-        STORAGE_OP_BUCKETS.into()
-    )
-    .expect("failed to define metric")
-});
-
 // Metrics collected on operations on the storage repository.
 #[derive(
     Clone, Copy, enum_map::Enum, strum_macros::EnumString, strum_macros::Display, IntoStaticStr,
@@ -946,7 +920,7 @@ pub(crate) static TIMELINE_EPHEMERAL_BYTES: Lazy<UIntGauge> = Lazy::new(|| {
     .expect("Failed to register metric")
 });
 
-/// Metrics related to the lifecycle of a [`crate::tenant::Tenant`] object: things
+/// Metrics related to the lifecycle of a [`crate::tenant::TenantShard`] object: things
 /// like how long it took to load.
 ///
 /// Note that these are process-global metrics, _not_ per-tenant metrics.  Per-tenant
@@ -1489,8 +1463,6 @@ pub enum SmgrQueryType {
     GetPageAtLsn,
     GetDbSize,
     GetSlruSegment,
-    #[cfg(feature = "testing")]
-    Test,
 }
 
 pub(crate) struct SmgrQueryTimePerTimeline {

pageserver/src/page_service.rs

@@ -59,7 +59,7 @@ use crate::span::debug_assert_current_span_has_tenant_and_timeline_id_no_shard_i
 use crate::task_mgr::TaskKind;
 use crate::task_mgr::{self, COMPUTE_REQUEST_RUNTIME};
 use crate::tenant::mgr::ShardSelector;
-use crate::tenant::mgr::TenantManager;
+use crate::tenant::mgr::TenantShardManager;
 use crate::tenant::mgr::{GetActiveTenantError, GetTenantError, ShardResolveResult};
 use crate::tenant::timeline::{self, WaitLsnError};
 use crate::tenant::GetTimelineError;
@@ -94,7 +94,7 @@ pub struct Connections {
 
 pub fn spawn(
     conf: &'static PageServerConf,
-    tenant_manager: Arc<TenantManager>,
+    tenant_manager: Arc<TenantShardManager>,
     pg_auth: Option<Arc<SwappableJwtAuth>>,
     tcp_listener: tokio::net::TcpListener,
 ) -> Listener {
@@ -159,7 +159,7 @@ impl Connections {
 /// open connections.
 ///
 pub async fn libpq_listener_main(
-    tenant_manager: Arc<TenantManager>,
+    tenant_manager: Arc<TenantShardManager>,
     auth: Option<Arc<SwappableJwtAuth>>,
     listener: tokio::net::TcpListener,
     auth_type: AuthType,
@@ -218,7 +218,7 @@ type ConnectionHandlerResult = anyhow::Result<()>;
 
 #[instrument(skip_all, fields(peer_addr))]
 async fn page_service_conn_main(
-    tenant_manager: Arc<TenantManager>,
+    tenant_manager: Arc<TenantShardManager>,
     auth: Option<Arc<SwappableJwtAuth>>,
     socket: tokio::net::TcpStream,
     auth_type: AuthType,
@@ -337,7 +337,7 @@ struct TimelineHandles {
 }
 
 impl TimelineHandles {
-    fn new(tenant_manager: Arc<TenantManager>) -> Self {
+    fn new(tenant_manager: Arc<TenantShardManager>) -> Self {
         Self {
             wrapper: TenantManagerWrapper {
                 tenant_manager,
@@ -379,7 +379,7 @@ impl TimelineHandles {
 }
 
 pub(crate) struct TenantManagerWrapper {
-    tenant_manager: Arc<TenantManager>,
+    tenant_manager: Arc<TenantShardManager>,
     // We do not support switching tenant_id on a connection at this point.
     // We can can add support for this later if needed without changing
     // the protocol.
@@ -555,52 +555,37 @@ struct BatchedGetPageRequest {
     timer: SmgrOpTimer,
 }
 
-#[cfg(feature = "testing")]
-struct BatchedTestRequest {
-    req: models::PagestreamTestRequest,
-    timer: SmgrOpTimer,
-}
-
-/// NB: we only hold [`timeline::handle::WeakHandle`] inside this enum,
-/// so that we don't keep the [`Timeline::gate`] open while the batch
-/// is being built up inside the [`spsc_fold`] (pagestream pipelining).
 enum BatchedFeMessage {
     Exists {
         span: Span,
         timer: SmgrOpTimer,
-        shard: timeline::handle::WeakHandle<TenantManagerTypes>,
+        shard: timeline::handle::Handle<TenantManagerTypes>,
         req: models::PagestreamExistsRequest,
     },
     Nblocks {
         span: Span,
         timer: SmgrOpTimer,
-        shard: timeline::handle::WeakHandle<TenantManagerTypes>,
+        shard: timeline::handle::Handle<TenantManagerTypes>,
         req: models::PagestreamNblocksRequest,
     },
     GetPage {
         span: Span,
-        shard: timeline::handle::WeakHandle<TenantManagerTypes>,
+        shard: timeline::handle::Handle<TenantManagerTypes>,
         effective_request_lsn: Lsn,
         pages: smallvec::SmallVec<[BatchedGetPageRequest; 1]>,
     },
     DbSize {
         span: Span,
         timer: SmgrOpTimer,
-        shard: timeline::handle::WeakHandle<TenantManagerTypes>,
+        shard: timeline::handle::Handle<TenantManagerTypes>,
         req: models::PagestreamDbSizeRequest,
     },
     GetSlruSegment {
         span: Span,
         timer: SmgrOpTimer,
-        shard: timeline::handle::WeakHandle<TenantManagerTypes>,
+        shard: timeline::handle::Handle<TenantManagerTypes>,
         req: models::PagestreamGetSlruSegmentRequest,
     },
-    #[cfg(feature = "testing")]
-    Test {
-        span: Span,
-        shard: timeline::handle::WeakHandle<TenantManagerTypes>,
-        requests: Vec<BatchedTestRequest>,
-    },
     RespondError {
         span: Span,
         error: BatchedPageStreamError,
@@ -621,12 +606,6 @@ impl BatchedFeMessage {
                     page.timer.observe_execution_start(at);
                 }
             }
-            #[cfg(feature = "testing")]
-            BatchedFeMessage::Test { requests, .. } => {
-                for req in requests {
-                    req.timer.observe_execution_start(at);
-                }
-            }
             BatchedFeMessage::RespondError { .. } => {}
         }
     }
@@ -634,7 +613,7 @@ impl BatchedFeMessage {
 
 impl PageServerHandler {
     pub fn new(
-        tenant_manager: Arc<TenantManager>,
+        tenant_manager: Arc<TenantShardManager>,
         auth: Option<Arc<SwappableJwtAuth>>,
         pipelining_config: PageServicePipeliningConfig,
         connection_ctx: RequestContext,
@@ -756,7 +735,7 @@ impl PageServerHandler {
                 BatchedFeMessage::Exists {
                     span,
                     timer,
-                    shard: shard.downgrade(),
+                    shard,
                     req,
                 }
             }
@@ -775,7 +754,7 @@ impl PageServerHandler {
                 BatchedFeMessage::Nblocks {
                     span,
                     timer,
-                    shard: shard.downgrade(),
+                    shard,
                     req,
                 }
             }
@@ -794,7 +773,7 @@ impl PageServerHandler {
                 BatchedFeMessage::DbSize {
                     span,
                     timer,
-                    shard: shard.downgrade(),
+                    shard,
                     req,
                 }
             }
@@ -813,7 +792,7 @@ impl PageServerHandler {
                 BatchedFeMessage::GetSlruSegment {
                     span,
                     timer,
-                    shard: shard.downgrade(),
+                    shard,
                     req,
                 }
             }
@@ -865,7 +844,6 @@ impl PageServerHandler {
                 )
                 .await?;
 
-                // We're holding the Handle
                 let effective_request_lsn = match Self::wait_or_get_last_lsn(
                     &shard,
                     req.hdr.request_lsn,
@@ -883,27 +861,11 @@ impl PageServerHandler {
                 };
                 BatchedFeMessage::GetPage {
                     span,
-                    shard: shard.downgrade(),
+                    shard,
                     effective_request_lsn,
                     pages: smallvec::smallvec![BatchedGetPageRequest { req, timer }],
                 }
             }
-            #[cfg(feature = "testing")]
-            PagestreamFeMessage::Test(req) => {
-                let span = tracing::info_span!(parent: parent_span, "handle_test_request");
-                let shard = timeline_handles
-                    .get(tenant_id, timeline_id, ShardSelector::Zero)
-                    .instrument(span.clone()) // sets `shard_id` field
-                    .await?;
-                let timer =
-                    record_op_start_and_throttle(&shard, metrics::SmgrQueryType::Test, received_at)
-                        .await?;
-                BatchedFeMessage::Test {
-                    span,
-                    shard: shard.downgrade(),
-                    requests: vec![BatchedTestRequest { req, timer }],
-                }
-            }
         };
         Ok(Some(batched_msg))
     }
@@ -945,7 +907,9 @@ impl PageServerHandler {
                     assert_eq!(accum_pages.len(), max_batch_size.get());
                     return false;
                 }
-                if !accum_shard.is_same_handle_as(&this_shard) {
+                if (accum_shard.tenant_shard_id, accum_shard.timeline_id)
+                    != (this_shard.tenant_shard_id, this_shard.timeline_id)
+                {
                     trace!(%accum_lsn, %this_lsn, "stopping batching because timeline object mismatch");
                     // TODO: we _could_ batch & execute each shard seperately (and in parallel).
                     // But the current logic for keeping responses in order does not support that.
@@ -964,44 +928,6 @@ impl PageServerHandler {
                 accum_pages.extend(this_pages);
                 Ok(())
             }
-            #[cfg(feature = "testing")]
-            (
-                Ok(BatchedFeMessage::Test {
-                    shard: accum_shard,
-                    requests: accum_requests,
-                    ..
-                }),
-                BatchedFeMessage::Test {
-                    shard: this_shard,
-                    requests: this_requests,
-                    ..
-                },
-            ) if (|| {
-                assert!(this_requests.len() == 1);
-                if accum_requests.len() >= max_batch_size.get() {
-                    trace!(%max_batch_size, "stopping batching because of batch size");
-                    assert_eq!(accum_requests.len(), max_batch_size.get());
-                    return false;
-                }
-                if !accum_shard.is_same_handle_as(&this_shard) {
-                    trace!("stopping batching because timeline object mismatch");
-                    // TODO: we _could_ batch & execute each shard seperately (and in parallel).
-                    // But the current logic for keeping responses in order does not support that.
-                    return false;
-                }
-                let this_batch_key = this_requests[0].req.batch_key;
-                let accum_batch_key = accum_requests[0].req.batch_key;
-                if this_requests[0].req.batch_key != accum_requests[0].req.batch_key {
-                    trace!(%accum_batch_key, %this_batch_key, "stopping batching because batch key changed");
-                    return false;
-                }
-                true
-            })() =>
-            {
-                // ok to batch
-                accum_requests.extend(this_requests);
-                Ok(())
-            }
             // something batched already but this message is unbatchable
             (_, this_msg) => {
                 // by default, don't continue batching
@@ -1043,7 +969,7 @@ impl PageServerHandler {
                 fail::fail_point!("ps::handle-pagerequest-message::exists");
                 (
                     vec![self
-                        .handle_get_rel_exists_request(&*shard.upgrade()?, &req, ctx)
+                        .handle_get_rel_exists_request(&shard, &req, ctx)
                         .instrument(span.clone())
                         .await
                         .map(|msg| (msg, timer))
@@ -1060,7 +986,7 @@ impl PageServerHandler {
                 fail::fail_point!("ps::handle-pagerequest-message::nblocks");
                 (
                     vec![self
-                        .handle_get_nblocks_request(&*shard.upgrade()?, &req, ctx)
+                        .handle_get_nblocks_request(&shard, &req, ctx)
                         .instrument(span.clone())
                         .await
                         .map(|msg| (msg, timer))
@@ -1081,7 +1007,7 @@ impl PageServerHandler {
                         trace!(npages, "handling getpage request");
                         let res = self
                             .handle_get_page_at_lsn_request_batched(
-                                &*shard.upgrade()?,
+                                &shard,
                                 effective_request_lsn,
                                 pages,
                                 ctx,
@@ -1103,7 +1029,7 @@ impl PageServerHandler {
                 fail::fail_point!("ps::handle-pagerequest-message::dbsize");
                 (
                     vec![self
-                        .handle_db_size_request(&*shard.upgrade()?, &req, ctx)
+                        .handle_db_size_request(&shard, &req, ctx)
                         .instrument(span.clone())
                         .await
                         .map(|msg| (msg, timer))
@@ -1120,7 +1046,7 @@ impl PageServerHandler {
                 fail::fail_point!("ps::handle-pagerequest-message::slrusegment");
                 (
                     vec![self
-                        .handle_get_slru_segment_request(&*shard.upgrade()?, &req, ctx)
+                        .handle_get_slru_segment_request(&shard, &req, ctx)
                         .instrument(span.clone())
                         .await
                         .map(|msg| (msg, timer))
@@ -1128,27 +1054,6 @@ impl PageServerHandler {
                     span,
                 )
             }
-            #[cfg(feature = "testing")]
-            BatchedFeMessage::Test {
-                span,
-                shard,
-                requests,
-            } => {
-                fail::fail_point!("ps::handle-pagerequest-message::test");
-                (
-                    {
-                        let npages = requests.len();
-                        trace!(npages, "handling getpage request");
-                        let res = self
-                            .handle_test_request_batch(&*shard.upgrade()?, requests, ctx)
-                            .instrument(span.clone())
-                            .await;
-                        assert_eq!(res.len(), npages);
-                        res
-                    },
-                    span,
-                )
-            }
             BatchedFeMessage::RespondError { span, error } => {
                 // We've already decided to respond with an error, so we don't need to
                 // call the handler.
@@ -1886,51 +1791,6 @@ impl PageServerHandler {
         ))
     }
 
-    // NB: this impl mimics what we do for batched getpage requests.
-    #[cfg(feature = "testing")]
-    #[instrument(skip_all, fields(shard_id))]
-    async fn handle_test_request_batch(
-        &mut self,
-        timeline: &Timeline,
-        requests: Vec<BatchedTestRequest>,
-        _ctx: &RequestContext,
-    ) -> Vec<Result<(PagestreamBeMessage, SmgrOpTimer), BatchedPageStreamError>> {
-        // real requests would do something with the timeline
-        let mut results = Vec::with_capacity(requests.len());
-        for _req in requests.iter() {
-            tokio::task::yield_now().await;
-
-            results.push({
-                if timeline.cancel.is_cancelled() {
-                    Err(PageReconstructError::Cancelled)
-                } else {
-                    Ok(())
-                }
-            });
-        }
-
-        // TODO: avoid creating the new Vec here
-        Vec::from_iter(
-            requests
-                .into_iter()
-                .zip(results.into_iter())
-                .map(|(req, res)| {
-                    res.map(|()| {
-                        (
-                            PagestreamBeMessage::Test(models::PagestreamTestResponse {
-                                req: req.req.clone(),
-                            }),
-                            req.timer,
-                        )
-                    })
-                    .map_err(|e| BatchedPageStreamError {
-                        err: PageStreamError::from(e),
-                        req: req.req.hdr,
-                    })
-                }),
-        )
-    }
-
     /// Note on "fullbackup":
     /// Full basebackups should only be used for debugging purposes.
     /// Originally, it was introduced to enable breaking storage format changes,
@@ -2546,14 +2406,6 @@ impl From<GetActiveTimelineError> for QueryError {
     }
 }
 
-impl From<crate::tenant::timeline::handle::HandleUpgradeError> for QueryError {
-    fn from(e: crate::tenant::timeline::handle::HandleUpgradeError) -> Self {
-        match e {
-            crate::tenant::timeline::handle::HandleUpgradeError::ShutDown => QueryError::Shutdown,
-        }
-    }
-}
-
 fn set_tracing_field_shard_id(timeline: &Timeline) {
     debug_assert_current_span_has_tenant_and_timeline_id_no_shard_id();
     tracing::Span::current().record(

pageserver/src/pgdatadir_mapping.rs

@@ -2404,13 +2404,13 @@ mod tests {
 
     use super::*;
 
-    use crate::{tenant::harness::TenantHarness, DEFAULT_PG_VERSION};
+    use crate::{tenant::harness::TenantShardHarness, DEFAULT_PG_VERSION};
 
     /// Test a round trip of aux file updates, from DatadirModification to reading back from the Timeline
     #[tokio::test]
     async fn aux_files_round_trip() -> anyhow::Result<()> {
         let name = "aux_files_round_trip";
-        let harness = TenantHarness::create(name).await?;
+        let harness = TenantShardHarness::create(name).await?;
 
         pub const TIMELINE_ID: TimelineId =
             TimelineId::from_array(hex!("11223344556677881122334455667788"));

pageserver/src/tenant.rs

@@ -95,9 +95,6 @@ use crate::deletion_queue::DeletionQueueError;
 use crate::import_datadir;
 use crate::is_uninit_mark;
 use crate::l0_flush::L0FlushGlobalState;
-use crate::metrics::CONCURRENT_INITDBS;
-use crate::metrics::INITDB_RUN_TIME;
-use crate::metrics::INITDB_SEMAPHORE_ACQUISITION_TIME;
 use crate::metrics::TENANT;
 use crate::metrics::{
     remove_tenant_metrics, BROKEN_TENANTS_SET, CIRCUIT_BREAKERS_BROKEN, CIRCUIT_BREAKERS_UNBROKEN,
@@ -196,7 +193,7 @@ pub struct TenantSharedResources {
     pub l0_flush_global_state: L0FlushGlobalState,
 }
 
-/// A [`Tenant`] is really an _attached_ tenant.  The configuration
+/// A [`TenantShard`] is really an _attached_ tenant.  The configuration
 /// for an attached tenant is a subset of the [`LocationConf`], represented
 /// in this struct.
 #[derive(Clone)]
@@ -276,7 +273,7 @@ pub(crate) enum SpawnMode {
 ///
 /// Tenant consists of multiple timelines. Keep them in a hash table.
 ///
-pub struct Tenant {
+pub struct TenantShard {
     // Global pageserver config parameters
     pub conf: &'static PageServerConf,
 
@@ -298,7 +295,7 @@ pub struct Tenant {
     shard_identity: ShardIdentity,
 
     /// The remote storage generation, used to protect S3 objects from split-brain.
-    /// Does not change over the lifetime of the [`Tenant`] object.
+    /// Does not change over the lifetime of the [`TenantShard`] object.
     ///
     /// This duplicates the generation stored in LocationConf, but that structure is mutable:
     /// this copy enforces the invariant that generatio doesn't change during a Tenant's lifetime.
@@ -337,7 +334,7 @@ pub struct Tenant {
     // Access to global deletion queue for when this tenant wants to schedule a deletion
     deletion_queue_client: DeletionQueueClient,
 
-    /// Cached logical sizes updated updated on each [`Tenant::gather_size_inputs`].
+    /// Cached logical sizes updated updated on each [`TenantShard::gather_size_inputs`].
     cached_logical_sizes: tokio::sync::Mutex<HashMap<(TimelineId, Lsn), u64>>,
     cached_synthetic_tenant_size: Arc<AtomicU64>,
 
@@ -367,7 +364,7 @@ pub struct Tenant {
     pub(crate) gate: Gate,
 
     /// Throttle applied at the top of [`Timeline::get`].
-    /// All [`Tenant::timelines`] of a given [`Tenant`] instance share the same [`throttle::Throttle`] instance.
+    /// All [`TenantShard::timelines`] of a given [`TenantShard`] instance share the same [`throttle::Throttle`] instance.
     pub(crate) pagestream_throttle: Arc<throttle::Throttle>,
 
     pub(crate) pagestream_throttle_metrics: Arc<crate::metrics::tenant_throttling::Pagestream>,
@@ -387,7 +384,7 @@ pub struct Tenant {
 
     l0_flush_global_state: L0FlushGlobalState,
 }
-impl std::fmt::Debug for Tenant {
+impl std::fmt::Debug for TenantShard {
     fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
         write!(f, "{} ({})", self.tenant_shard_id, self.current_state())
     }
@@ -863,7 +860,7 @@ impl Debug for SetStoppingError {
     }
 }
 
-/// Arguments to [`Tenant::create_timeline`].
+/// Arguments to [`TenantShard::create_timeline`].
 ///
 /// Not usable as an idempotency key for timeline creation because if [`CreateTimelineParamsBranch::ancestor_start_lsn`]
 /// is `None`, the result of the timeline create call is not deterministic.
@@ -898,7 +895,7 @@ pub(crate) struct CreateTimelineParamsImportPgdata {
     pub(crate) idempotency_key: import_pgdata::index_part_format::IdempotencyKey,
 }
 
-/// What is used to determine idempotency of a [`Tenant::create_timeline`] call in  [`Tenant::start_creating_timeline`] in  [`Tenant::start_creating_timeline`].
+/// What is used to determine idempotency of a [`TenantShard::create_timeline`] call in  [`TenantShard::start_creating_timeline`] in  [`TenantShard::start_creating_timeline`].
 ///
 /// Each [`Timeline`] object holds [`Self`] as an immutable property in [`Timeline::create_idempotency`].
 ///
@@ -936,7 +933,7 @@ pub(crate) struct CreatingTimelineIdempotencyImportPgdata {
     idempotency_key: import_pgdata::index_part_format::IdempotencyKey,
 }
 
-/// What is returned by [`Tenant::start_creating_timeline`].
+/// What is returned by [`TenantShard::start_creating_timeline`].
 #[must_use]
 enum StartCreatingTimelineResult {
     CreateGuard(TimelineCreateGuard),
@@ -964,13 +961,13 @@ struct TimelineInitAndSyncNeedsSpawnImportPgdata {
     guard: TimelineCreateGuard,
 }
 
-/// What is returned by [`Tenant::create_timeline`].
+/// What is returned by [`TenantShard::create_timeline`].
 enum CreateTimelineResult {
     Created(Arc<Timeline>),
     Idempotent(Arc<Timeline>),
-    /// IMPORTANT: This [`Arc<Timeline>`] object is not in [`Tenant::timelines`] when
+    /// IMPORTANT: This [`Arc<Timeline>`] object is not in [`TenantShard::timelines`] when
     /// we return this result, nor will this concrete object ever be added there.
-    /// Cf method comment on [`Tenant::create_timeline_import_pgdata`].
+    /// Cf method comment on [`TenantShard::create_timeline_import_pgdata`].
     ImportSpawned(Arc<Timeline>),
 }
 
@@ -1102,7 +1099,7 @@ pub(crate) enum LoadConfigError {
     NotFound(Utf8PathBuf),
 }
 
-impl Tenant {
+impl TenantShard {
     /// Yet another helper for timeline initialization.
     ///
     /// - Initializes the Timeline struct and inserts it into the tenant's hash map
@@ -1280,7 +1277,7 @@ impl Tenant {
         init_order: Option<InitializationOrder>,
         mode: SpawnMode,
         ctx: &RequestContext,
-    ) -> Result<Arc<Tenant>, GlobalShutDown> {
+    ) -> Result<Arc<TenantShard>, GlobalShutDown> {
         let wal_redo_manager =
             WalRedoManager::new(PostgresRedoManager::new(conf, tenant_shard_id))?;
 
@@ -1294,7 +1291,7 @@ impl Tenant {
         let attach_mode = attached_conf.location.attach_mode;
         let generation = attached_conf.location.generation;
 
-        let tenant = Arc::new(Tenant::new(
+        let tenant = Arc::new(TenantShard::new(
             TenantState::Attaching,
             conf,
             attached_conf,
@@ -1339,13 +1336,13 @@ impl Tenant {
                     }
                 }
 
-                // Ideally we should use Tenant::set_broken_no_wait, but it is not supposed to be used when tenant is in loading state.
+                // Ideally we should use TenantShard::set_broken_no_wait, but it is not supposed to be used when tenant is in loading state.
                 enum BrokenVerbosity {
                     Error,
                     Info
                 }
                 let make_broken =
-                    |t: &Tenant, err: anyhow::Error, verbosity: BrokenVerbosity| {
+                    |t: &TenantShard, err: anyhow::Error, verbosity: BrokenVerbosity| {
                         match verbosity {
                             BrokenVerbosity::Info => {
                                 info!("attach cancelled, setting tenant state to Broken: {err}");
@@ -1568,7 +1565,7 @@ impl Tenant {
     /// No background tasks are started as part of this routine.
     ///
     async fn attach(
-        self: &Arc<Tenant>,
+        self: &Arc<TenantShard>,
         preload: Option<TenantPreload>,
         ctx: &RequestContext,
     ) -> anyhow::Result<()> {
@@ -1707,7 +1704,7 @@ impl Tenant {
 
             match effect {
                 TimelineInitAndSyncResult::ReadyToActivate(_) => {
-                    // activation happens later, on Tenant::activate
+                    // activation happens later, on TenantShard::activate
                 }
                 TimelineInitAndSyncResult::NeedsSpawnImportPgdata(
                     TimelineInitAndSyncNeedsSpawnImportPgdata {
@@ -1888,7 +1885,7 @@ impl Tenant {
     }
 
     async fn load_timelines_metadata(
-        self: &Arc<Tenant>,
+        self: &Arc<TenantShard>,
         timeline_ids: HashSet<TimelineId>,
         remote_storage: &GenericRemoteStorage,
         cancel: CancellationToken,
@@ -1943,7 +1940,7 @@ impl Tenant {
     }
 
     fn load_timeline_metadata(
-        self: &Arc<Tenant>,
+        self: &Arc<TenantShard>,
         timeline_id: TimelineId,
         remote_storage: GenericRemoteStorage,
         cancel: CancellationToken,
@@ -2336,14 +2333,14 @@ impl Tenant {
     /// This is used by tests & import-from-basebackup.
     ///
     /// The returned [`UninitializedTimeline`] contains no data nor metadata and it is in
-    /// a state that will fail [`Tenant::load_remote_timeline`] because `disk_consistent_lsn=Lsn(0)`.
+    /// a state that will fail [`TenantShard::load_remote_timeline`] because `disk_consistent_lsn=Lsn(0)`.
     ///
     /// The caller is responsible for getting the timeline into a state that will be accepted
-    /// by [`Tenant::load_remote_timeline`] / [`Tenant::attach`].
+    /// by [`TenantShard::load_remote_timeline`] / [`TenantShard::attach`].
     /// Then they may call [`UninitializedTimeline::finish_creation`] to add the timeline
-    /// to the [`Tenant::timelines`].
+    /// to the [`TenantShard::timelines`].
     ///
-    /// Tests should use `Tenant::create_test_timeline` to set up the minimum required metadata keys.
+    /// Tests should use `TenantShard::create_test_timeline` to set up the minimum required metadata keys.
     pub(crate) async fn create_empty_timeline(
         self: &Arc<Self>,
         new_timeline_id: TimelineId,
@@ -2483,7 +2480,7 @@ impl Tenant {
     /// the same timeline ID already exists, returns CreateTimelineError::AlreadyExists.
     #[allow(clippy::too_many_arguments)]
     pub(crate) async fn create_timeline(
-        self: &Arc<Tenant>,
+        self: &Arc<TenantShard>,
         params: CreateTimelineParams,
         broker_client: storage_broker::BrokerClientChannel,
         ctx: &RequestContext,
@@ -2638,13 +2635,13 @@ impl Tenant {
         Ok(activated_timeline)
     }
 
-    /// The returned [`Arc<Timeline>`] is NOT in the [`Tenant::timelines`] map until the import
+    /// The returned [`Arc<Timeline>`] is NOT in the [`TenantShard::timelines`] map until the import
     /// completes in the background. A DIFFERENT [`Arc<Timeline>`] will be inserted into the
-    /// [`Tenant::timelines`] map when the import completes.
+    /// [`TenantShard::timelines`] map when the import completes.
     /// We only return an [`Arc<Timeline>`] here so the API handler can create a [`pageserver_api::models::TimelineInfo`]
     /// for the response.
     async fn create_timeline_import_pgdata(
-        self: &Arc<Tenant>,
+        self: &Arc<TenantShard>,
         params: CreateTimelineParamsImportPgdata,
         activate: ActivateTimelineArgs,
         ctx: &RequestContext,
@@ -2739,7 +2736,7 @@ impl Tenant {
 
     #[instrument(skip_all, fields(tenant_id=%self.tenant_shard_id.tenant_id, shard_id=%self.tenant_shard_id.shard_slug(), timeline_id=%timeline.timeline_id))]
     async fn create_timeline_import_pgdata_task(
-        self: Arc<Tenant>,
+        self: Arc<TenantShard>,
         timeline: Arc<Timeline>,
         index_part: import_pgdata::index_part_format::Root,
         activate: ActivateTimelineArgs,
@@ -2765,7 +2762,7 @@ impl Tenant {
     }
 
     async fn create_timeline_import_pgdata_task_impl(
-        self: Arc<Tenant>,
+        self: Arc<TenantShard>,
         timeline: Arc<Timeline>,
         index_part: import_pgdata::index_part_format::Root,
         activate: ActivateTimelineArgs,
@@ -2783,10 +2780,10 @@ impl Tenant {
         // Reload timeline from remote.
         // This proves that the remote state is attachable, and it reuses the code.
         //
-        // TODO: think about whether this is safe to do with concurrent Tenant::shutdown.
+        // TODO: think about whether this is safe to do with concurrent TenantShard::shutdown.
         // timeline_create_guard hols the tenant gate open, so, shutdown cannot _complete_ until we exit.
-        // But our activate() call might launch new background tasks after Tenant::shutdown
-        // already went past shutting down the Tenant::timelines, which this timeline here is no part of.
+        // But our activate() call might launch new background tasks after TenantShard::shutdown
+        // already went past shutting down the TenantShard::timelines, which this timeline here is no part of.
         // I think the same problem exists with the bootstrap & branch mgmt API tasks (tenant shutting
         // down while bootstrapping/branching + activating), but, the race condition is much more likely
         // to manifest because of the long runtime of this import task.
@@ -2801,7 +2798,7 @@ impl Tenant {
         // };
         let timeline_id = timeline.timeline_id;
 
-        // load from object storage like Tenant::attach does
+        // load from object storage like TenantShard::attach does
         let resources = self.build_timeline_resources(timeline_id);
         let index_part = resources
             .remote_client
@@ -3271,7 +3268,7 @@ impl Tenant {
             }
             Err(SetStoppingError::AlreadyStopping(other)) => {
                 // give caller the option to wait for this this shutdown
-                info!("Tenant::shutdown: AlreadyStopping");
+                info!("TenantShard::shutdown: AlreadyStopping");
                 return Err(other);
             }
         };
@@ -3764,7 +3761,7 @@ enum ActivateTimelineArgs {
     No,
 }
 
-impl Tenant {
+impl TenantShard {
     pub fn tenant_specific_overrides(&self) -> TenantConfOpt {
         self.tenant_conf.load().tenant_conf.clone()
     }
@@ -3885,7 +3882,7 @@ impl Tenant {
         update: F,
     ) -> anyhow::Result<TenantConfOpt> {
         // Use read-copy-update in order to avoid overwriting the location config
-        // state if this races with [`Tenant::set_new_location_config`]. Note that
+        // state if this races with [`TenantShard::set_new_location_config`]. Note that
         // this race is not possible if both request types come from the storage
         // controller (as they should!) because an exclusive op lock is required
         // on the storage controller side.
@@ -3997,7 +3994,7 @@ impl Tenant {
         Ok(timeline)
     }
 
-    /// [`Tenant::shutdown`] must be called before dropping the returned [`Tenant`] object
+    /// [`TenantShard::shutdown`] must be called before dropping the returned [`TenantShard`] object
     /// to ensure proper cleanup of background tasks and metrics.
     //
     // Allow too_many_arguments because a constructor's argument list naturally grows with the
@@ -4013,7 +4010,7 @@ impl Tenant {
         remote_storage: GenericRemoteStorage,
         deletion_queue_client: DeletionQueueClient,
         l0_flush_global_state: L0FlushGlobalState,
-    ) -> Tenant {
+    ) -> TenantShard {
         debug_assert!(
             !attached_conf.location.generation.is_none() || conf.control_plane_api.is_none()
         );
@@ -4073,7 +4070,7 @@ impl Tenant {
             }
         });
 
-        Tenant {
+        TenantShard {
             tenant_shard_id,
             shard_identity,
             generation: attached_conf.location.generation,
@@ -4107,7 +4104,7 @@ impl Tenant {
             cancel: CancellationToken::default(),
             gate: Gate::default(),
             pagestream_throttle: Arc::new(throttle::Throttle::new(
-                Tenant::get_pagestream_throttle_config(conf, &attached_conf.tenant_conf),
+                TenantShard::get_pagestream_throttle_config(conf, &attached_conf.tenant_conf),
             )),
             pagestream_throttle_metrics: Arc::new(
                 crate::metrics::tenant_throttling::Pagestream::new(&tenant_shard_id),
@@ -4246,11 +4243,11 @@ impl Tenant {
 
         // Perform GC for each timeline.
         //
-        // Note that we don't hold the `Tenant::gc_cs` lock here because we don't want to delay the
+        // Note that we don't hold the `TenantShard::gc_cs` lock here because we don't want to delay the
         // branch creation task, which requires the GC lock. A GC iteration can run concurrently
         // with branch creation.
         //
-        // See comments in [`Tenant::branch_timeline`] for more information about why branch
+        // See comments in [`TenantShard::branch_timeline`] for more information about why branch
         // creation task can run concurrently with timeline's GC iteration.
         for timeline in gc_timelines {
             if cancel.is_cancelled() {
@@ -4280,7 +4277,7 @@ impl Tenant {
 
     /// Refreshes the Timeline::gc_info for all timelines, returning the
     /// vector of timelines which have [`Timeline::get_last_record_lsn`] past
-    /// [`Tenant::get_gc_horizon`].
+    /// [`TenantShard::get_gc_horizon`].
     ///
     /// This is usually executed as part of periodic gc, but can now be triggered more often.
     pub(crate) async fn refresh_gc_info(
@@ -5350,17 +5347,8 @@ async fn run_initdb(
         initdb_bin_path, initdb_target_dir, initdb_lib_dir,
     );
 
-    let _permit = {
-        let _timer = INITDB_SEMAPHORE_ACQUISITION_TIME.start_timer();
-        INIT_DB_SEMAPHORE.acquire().await
-    };
+    let _permit = INIT_DB_SEMAPHORE.acquire().await;
 
-    CONCURRENT_INITDBS.inc();
-    scopeguard::defer! {
-        CONCURRENT_INITDBS.dec();
-    }
-
-    let _timer = INITDB_RUN_TIME.start_timer();
     let res = postgres_initdb::do_run_initdb(postgres_initdb::RunInitdbArgs {
         superuser: &conf.superuser,
         locale: &conf.locale,
@@ -5479,7 +5467,7 @@ pub(crate) mod harness {
         }
     }
 
-    pub struct TenantHarness {
+    pub struct TenantShardHarness {
         pub conf: &'static PageServerConf,
         pub tenant_conf: TenantConf,
         pub tenant_shard_id: TenantShardId,
@@ -5505,7 +5493,7 @@ pub(crate) mod harness {
         });
     }
 
-    impl TenantHarness {
+    impl TenantShardHarness {
         pub async fn create_custom(
             test_name: &'static str,
             tenant_conf: TenantConf,
@@ -5582,7 +5570,7 @@ pub(crate) mod harness {
             info_span!("TenantHarness", tenant_id=%self.tenant_shard_id.tenant_id, shard_id=%self.tenant_shard_id.shard_slug())
         }
 
-        pub(crate) async fn load(&self) -> (Arc<Tenant>, RequestContext) {
+        pub(crate) async fn load(&self) -> (Arc<TenantShard>, RequestContext) {
             let ctx = RequestContext::new(TaskKind::UnitTest, DownloadBehavior::Error);
             (
                 self.do_try_load(&ctx)
@@ -5596,10 +5584,10 @@ pub(crate) mod harness {
         pub(crate) async fn do_try_load(
             &self,
             ctx: &RequestContext,
-        ) -> anyhow::Result<Arc<Tenant>> {
+        ) -> anyhow::Result<Arc<TenantShard>> {
             let walredo_mgr = Arc::new(WalRedoManager::from(TestRedoManager));
 
-            let tenant = Arc::new(Tenant::new(
+            let tenant = Arc::new(TenantShard::new(
                 TenantState::Attaching,
                 self.conf,
                 AttachedTenantConf::try_from(LocationConf::attached_single(
@@ -5728,7 +5716,7 @@ mod tests {
 
     #[tokio::test]
     async fn test_basic() -> anyhow::Result<()> {
-        let (tenant, ctx) = TenantHarness::create("test_basic").await?.load().await;
+        let (tenant, ctx) = TenantShardHarness::create("test_basic").await?.load().await;
         let tline = tenant
             .create_test_timeline(TIMELINE_ID, Lsn(0x08), DEFAULT_PG_VERSION, &ctx)
             .await?;
@@ -5775,7 +5763,7 @@ mod tests {
 
     #[tokio::test]
     async fn no_duplicate_timelines() -> anyhow::Result<()> {
-        let (tenant, ctx) = TenantHarness::create("no_duplicate_timelines")
+        let (tenant, ctx) = TenantShardHarness::create("no_duplicate_timelines")
             .await?
             .load()
             .await;
@@ -5811,7 +5799,10 @@ mod tests {
     async fn test_branch() -> anyhow::Result<()> {
         use std::str::from_utf8;
 
-        let (tenant, ctx) = TenantHarness::create("test_branch").await?.load().await;
+        let (tenant, ctx) = TenantShardHarness::create("test_branch")
+            .await?
+            .load()
+            .await;
         let tline = tenant
             .create_test_timeline(TIMELINE_ID, Lsn(0x10), DEFAULT_PG_VERSION, &ctx)
             .await?;
@@ -5933,7 +5924,7 @@ mod tests {
     #[tokio::test(start_paused = true)]
     async fn test_prohibit_branch_creation_on_garbage_collected_data() -> anyhow::Result<()> {
         let (tenant, ctx) =
-            TenantHarness::create("test_prohibit_branch_creation_on_garbage_collected_data")
+            TenantShardHarness::create("test_prohibit_branch_creation_on_garbage_collected_data")
                 .await?
                 .load()
                 .await;
@@ -5985,7 +5976,7 @@ mod tests {
     #[tokio::test]
     async fn test_prohibit_branch_creation_on_pre_initdb_lsn() -> anyhow::Result<()> {
         let (tenant, ctx) =
-            TenantHarness::create("test_prohibit_branch_creation_on_pre_initdb_lsn")
+            TenantShardHarness::create("test_prohibit_branch_creation_on_pre_initdb_lsn")
                 .await?
                 .load()
                 .await;
@@ -6041,7 +6032,7 @@ mod tests {
     #[tokio::test]
     async fn test_get_branchpoints_from_an_inactive_timeline() -> anyhow::Result<()> {
         let (tenant, ctx) =
-            TenantHarness::create("test_get_branchpoints_from_an_inactive_timeline")
+            TenantShardHarness::create("test_get_branchpoints_from_an_inactive_timeline")
                 .await?
                 .load()
                 .await;
@@ -6105,7 +6096,7 @@ mod tests {
     #[tokio::test]
     async fn test_retain_data_in_parent_which_is_needed_for_child() -> anyhow::Result<()> {
         let (tenant, ctx) =
-            TenantHarness::create("test_retain_data_in_parent_which_is_needed_for_child")
+            TenantShardHarness::create("test_retain_data_in_parent_which_is_needed_for_child")
                 .await?
                 .load()
                 .await;
@@ -6136,10 +6127,11 @@ mod tests {
     }
     #[tokio::test]
     async fn test_parent_keeps_data_forever_after_branching() -> anyhow::Result<()> {
-        let (tenant, ctx) = TenantHarness::create("test_parent_keeps_data_forever_after_branching")
-            .await?
-            .load()
-            .await;
+        let (tenant, ctx) =
+            TenantShardHarness::create("test_parent_keeps_data_forever_after_branching")
+                .await?
+                .load()
+                .await;
         let tline = tenant
             .create_test_timeline(TIMELINE_ID, Lsn(0x10), DEFAULT_PG_VERSION, &ctx)
             .await?;
@@ -6177,7 +6169,7 @@ mod tests {
     #[tokio::test]
     async fn timeline_load() -> anyhow::Result<()> {
         const TEST_NAME: &str = "timeline_load";
-        let harness = TenantHarness::create(TEST_NAME).await?;
+        let harness = TenantShardHarness::create(TEST_NAME).await?;
         {
             let (tenant, ctx) = harness.load().await;
             let tline = tenant
@@ -6204,7 +6196,7 @@ mod tests {
     #[tokio::test]
     async fn timeline_load_with_ancestor() -> anyhow::Result<()> {
         const TEST_NAME: &str = "timeline_load_with_ancestor";
-        let harness = TenantHarness::create(TEST_NAME).await?;
+        let harness = TenantShardHarness::create(TEST_NAME).await?;
         // create two timelines
         {
             let (tenant, ctx) = harness.load().await;
@@ -6252,7 +6244,7 @@ mod tests {
     #[tokio::test]
     async fn delta_layer_dumping() -> anyhow::Result<()> {
         use storage_layer::AsLayerDesc;
-        let (tenant, ctx) = TenantHarness::create("test_layer_dumping")
+        let (tenant, ctx) = TenantShardHarness::create("test_layer_dumping")
             .await?
             .load()
             .await;
@@ -6282,7 +6274,10 @@ mod tests {
 
     #[tokio::test]
     async fn test_images() -> anyhow::Result<()> {
-        let (tenant, ctx) = TenantHarness::create("test_images").await?.load().await;
+        let (tenant, ctx) = TenantShardHarness::create("test_images")
+            .await?
+            .load()
+            .await;
         let tline = tenant
             .create_test_timeline(TIMELINE_ID, Lsn(0x08), DEFAULT_PG_VERSION, &ctx)
             .await?;
@@ -6380,7 +6375,7 @@ mod tests {
     }
 
     async fn bulk_insert_compact_gc(
-        tenant: &Tenant,
+        tenant: &TenantShard,
         timeline: &Arc<Timeline>,
         ctx: &RequestContext,
         lsn: Lsn,
@@ -6392,7 +6387,7 @@ mod tests {
     }
 
     async fn bulk_insert_maybe_compact_gc(
-        tenant: &Tenant,
+        tenant: &TenantShard,
         timeline: &Arc<Timeline>,
         ctx: &RequestContext,
         mut lsn: Lsn,
@@ -6456,7 +6451,7 @@ mod tests {
     //
     #[tokio::test]
     async fn test_bulk_insert() -> anyhow::Result<()> {
-        let harness = TenantHarness::create("test_bulk_insert").await?;
+        let harness = TenantShardHarness::create("test_bulk_insert").await?;
         let (tenant, ctx) = harness.load().await;
         let tline = tenant
             .create_test_timeline(TIMELINE_ID, Lsn(0x08), DEFAULT_PG_VERSION, &ctx)
@@ -6487,7 +6482,7 @@ mod tests {
     // so the search can stop at the first delta layer and doesn't traverse any deeper.
     #[tokio::test]
     async fn test_get_vectored() -> anyhow::Result<()> {
-        let harness = TenantHarness::create("test_get_vectored").await?;
+        let harness = TenantShardHarness::create("test_get_vectored").await?;
         let (tenant, ctx) = harness.load().await;
         let tline = tenant
             .create_test_timeline(TIMELINE_ID, Lsn(0x08), DEFAULT_PG_VERSION, &ctx)
@@ -6597,7 +6592,7 @@ mod tests {
 
     #[tokio::test]
     async fn test_get_vectored_aux_files() -> anyhow::Result<()> {
-        let harness = TenantHarness::create("test_get_vectored_aux_files").await?;
+        let harness = TenantShardHarness::create("test_get_vectored_aux_files").await?;
 
         let (tenant, ctx) = harness.load().await;
         let tline = tenant
@@ -6673,7 +6668,7 @@ mod tests {
             ..TenantConf::default()
         };
 
-        let harness = TenantHarness::create_custom(
+        let harness = TenantShardHarness::create_custom(
             "test_get_vectored_key_gap",
             tenant_conf,
             TenantId::generate(),
@@ -6823,7 +6818,7 @@ mod tests {
     // ```
     #[tokio::test]
     async fn test_get_vectored_ancestor_descent() -> anyhow::Result<()> {
-        let harness = TenantHarness::create("test_get_vectored_on_lsn_axis").await?;
+        let harness = TenantShardHarness::create("test_get_vectored_on_lsn_axis").await?;
         let (tenant, ctx) = harness.load().await;
 
         let start_key = Key::from_hex("010000000033333333444444445500000000").unwrap();
@@ -6972,7 +6967,7 @@ mod tests {
         name: &'static str,
         compaction_algorithm: CompactionAlgorithm,
     ) -> anyhow::Result<()> {
-        let mut harness = TenantHarness::create(name).await?;
+        let mut harness = TenantShardHarness::create(name).await?;
         harness.tenant_conf.compaction_algorithm = CompactionAlgorithmSettings {
             kind: compaction_algorithm,
         };
@@ -7056,7 +7051,7 @@ mod tests {
 
     #[tokio::test]
     async fn test_traverse_branches() -> anyhow::Result<()> {
-        let (tenant, ctx) = TenantHarness::create("test_traverse_branches")
+        let (tenant, ctx) = TenantShardHarness::create("test_traverse_branches")
             .await?
             .load()
             .await;
@@ -7147,7 +7142,7 @@ mod tests {
 
     #[tokio::test]
     async fn test_traverse_ancestors() -> anyhow::Result<()> {
-        let (tenant, ctx) = TenantHarness::create("test_traverse_ancestors")
+        let (tenant, ctx) = TenantShardHarness::create("test_traverse_ancestors")
             .await?
             .load()
             .await;
@@ -7214,7 +7209,7 @@ mod tests {
 
     #[tokio::test]
     async fn test_write_at_initdb_lsn_takes_optimization_code_path() -> anyhow::Result<()> {
-        let (tenant, ctx) = TenantHarness::create("test_empty_test_timeline_is_usable")
+        let (tenant, ctx) = TenantShardHarness::create("test_empty_test_timeline_is_usable")
             .await?
             .load()
             .await;
@@ -7284,13 +7279,13 @@ mod tests {
     #[tokio::test]
     async fn test_create_guard_crash() -> anyhow::Result<()> {
         let name = "test_create_guard_crash";
-        let harness = TenantHarness::create(name).await?;
+        let harness = TenantShardHarness::create(name).await?;
         {
             let (tenant, ctx) = harness.load().await;
             let tline = tenant
                 .create_empty_timeline(TIMELINE_ID, Lsn(0), DEFAULT_PG_VERSION, &ctx)
                 .await?;
-            // Leave the timeline ID in [`Tenant::timelines_creating`] to exclude attempting to create it again
+            // Leave the timeline ID in [`TenantShard::timelines_creating`] to exclude attempting to create it again
             let raw_tline = tline.raw_timeline().unwrap();
             raw_tline
                 .shutdown(super::timeline::ShutdownMode::Hard)
@@ -7337,7 +7332,7 @@ mod tests {
         name: &'static str,
         compaction_algorithm: CompactionAlgorithm,
     ) -> anyhow::Result<()> {
-        let mut harness = TenantHarness::create(name).await?;
+        let mut harness = TenantShardHarness::create(name).await?;
         harness.tenant_conf.compaction_algorithm = CompactionAlgorithmSettings {
             kind: compaction_algorithm,
         };
@@ -7361,7 +7356,7 @@ mod tests {
 
     #[tokio::test]
     async fn test_metadata_scan() -> anyhow::Result<()> {
-        let harness = TenantHarness::create("test_metadata_scan").await?;
+        let harness = TenantShardHarness::create("test_metadata_scan").await?;
         let (tenant, ctx) = harness.load().await;
         let tline = tenant
             .create_test_timeline(TIMELINE_ID, Lsn(0x10), DEFAULT_PG_VERSION, &ctx)
@@ -7480,7 +7475,7 @@ mod tests {
 
     #[tokio::test]
     async fn test_metadata_compaction_trigger() -> anyhow::Result<()> {
-        let harness = TenantHarness::create("test_metadata_compaction_trigger").await?;
+        let harness = TenantShardHarness::create("test_metadata_compaction_trigger").await?;
         let (tenant, ctx) = harness.load().await;
         let tline = tenant
             .create_test_timeline(TIMELINE_ID, Lsn(0x10), DEFAULT_PG_VERSION, &ctx)
@@ -7528,7 +7523,9 @@ mod tests {
 
     #[tokio::test]
     async fn test_aux_file_e2e() {
-        let harness = TenantHarness::create("test_aux_file_e2e").await.unwrap();
+        let harness = TenantShardHarness::create("test_aux_file_e2e")
+            .await
+            .unwrap();
 
         let (tenant, ctx) = harness.load().await;
 
@@ -7584,7 +7581,7 @@ mod tests {
 
     #[tokio::test]
     async fn test_metadata_image_creation() -> anyhow::Result<()> {
-        let harness = TenantHarness::create("test_metadata_image_creation").await?;
+        let harness = TenantShardHarness::create("test_metadata_image_creation").await?;
         let (tenant, ctx) = harness.load().await;
         let tline = tenant
             .create_test_timeline(TIMELINE_ID, Lsn(0x10), DEFAULT_PG_VERSION, &ctx)
@@ -7683,7 +7680,7 @@ mod tests {
 
     #[tokio::test]
     async fn test_vectored_missing_data_key_reads() -> anyhow::Result<()> {
-        let harness = TenantHarness::create("test_vectored_missing_data_key_reads").await?;
+        let harness = TenantShardHarness::create("test_vectored_missing_data_key_reads").await?;
         let (tenant, ctx) = harness.load().await;
 
         let base_key = Key::from_hex("000000000033333333444444445500000000").unwrap();
@@ -7755,7 +7752,8 @@ mod tests {
 
     #[tokio::test]
     async fn test_vectored_missing_metadata_key_reads() -> anyhow::Result<()> {
-        let harness = TenantHarness::create("test_vectored_missing_metadata_key_reads").await?;
+        let harness =
+            TenantShardHarness::create("test_vectored_missing_metadata_key_reads").await?;
         let (tenant, ctx) = harness.load().await;
 
         let base_key = Key::from_hex("620000000033333333444444445500000000").unwrap();
@@ -7976,7 +7974,7 @@ mod tests {
 
     #[tokio::test]
     async fn test_metadata_tombstone_reads() -> anyhow::Result<()> {
-        let harness = TenantHarness::create("test_metadata_tombstone_reads").await?;
+        let harness = TenantShardHarness::create("test_metadata_tombstone_reads").await?;
         let (tenant, ctx) = harness.load().await;
         let key0 = Key::from_hex("620000000033333333444444445500000000").unwrap();
         let key1 = Key::from_hex("620000000033333333444444445500000001").unwrap();
@@ -8056,7 +8054,7 @@ mod tests {
 
     #[tokio::test]
     async fn test_metadata_tombstone_image_creation() {
-        let harness = TenantHarness::create("test_metadata_tombstone_image_creation")
+        let harness = TenantShardHarness::create("test_metadata_tombstone_image_creation")
             .await
             .unwrap();
         let (tenant, ctx) = harness.load().await;
@@ -8130,7 +8128,7 @@ mod tests {
 
     #[tokio::test]
     async fn test_metadata_tombstone_empty_image_creation() {
-        let harness = TenantHarness::create("test_metadata_tombstone_empty_image_creation")
+        let harness = TenantShardHarness::create("test_metadata_tombstone_empty_image_creation")
             .await
             .unwrap();
         let (tenant, ctx) = harness.load().await;
@@ -8195,7 +8193,8 @@ mod tests {
 
     #[tokio::test]
     async fn test_simple_bottom_most_compaction_images() -> anyhow::Result<()> {
-        let harness = TenantHarness::create("test_simple_bottom_most_compaction_images").await?;
+        let harness =
+            TenantShardHarness::create("test_simple_bottom_most_compaction_images").await?;
         let (tenant, ctx) = harness.load().await;
 
         fn get_key(id: u32) -> Key {
@@ -8415,7 +8414,7 @@ mod tests {
     #[cfg(feature = "testing")]
     #[tokio::test]
     async fn test_neon_test_record() -> anyhow::Result<()> {
-        let harness = TenantHarness::create("test_neon_test_record").await?;
+        let harness = TenantShardHarness::create("test_neon_test_record").await?;
         let (tenant, ctx) = harness.load().await;
 
         fn get_key(id: u32) -> Key {
@@ -8496,7 +8495,7 @@ mod tests {
 
     #[tokio::test(start_paused = true)]
     async fn test_lsn_lease() -> anyhow::Result<()> {
-        let (tenant, ctx) = TenantHarness::create("test_lsn_lease")
+        let (tenant, ctx) = TenantShardHarness::create("test_lsn_lease")
             .await
             .unwrap()
             .load()
@@ -8629,7 +8628,7 @@ mod tests {
         test_name: &'static str,
         use_delta_bottom_layer: bool,
     ) -> anyhow::Result<()> {
-        let harness = TenantHarness::create(test_name).await?;
+        let harness = TenantShardHarness::create(test_name).await?;
         let (tenant, ctx) = harness.load().await;
 
         fn get_key(id: u32) -> Key {
@@ -8888,7 +8887,7 @@ mod tests {
     #[cfg(feature = "testing")]
     #[tokio::test]
     async fn test_generate_key_retention() -> anyhow::Result<()> {
-        let harness = TenantHarness::create("test_generate_key_retention").await?;
+        let harness = TenantShardHarness::create("test_generate_key_retention").await?;
         let (tenant, ctx) = harness.load().await;
         let tline = tenant
             .create_test_timeline(TIMELINE_ID, Lsn(0x10), DEFAULT_PG_VERSION, &ctx)
@@ -9237,7 +9236,8 @@ mod tests {
     #[tokio::test]
     async fn test_simple_bottom_most_compaction_with_retain_lsns() -> anyhow::Result<()> {
         let harness =
-            TenantHarness::create("test_simple_bottom_most_compaction_with_retain_lsns").await?;
+            TenantShardHarness::create("test_simple_bottom_most_compaction_with_retain_lsns")
+                .await?;
         let (tenant, ctx) = harness.load().await;
 
         fn get_key(id: u32) -> Key {
@@ -9497,9 +9497,10 @@ mod tests {
     #[tokio::test]
     async fn test_simple_bottom_most_compaction_with_retain_lsns_single_key() -> anyhow::Result<()>
     {
-        let harness =
-            TenantHarness::create("test_simple_bottom_most_compaction_with_retain_lsns_single_key")
-                .await?;
+        let harness = TenantShardHarness::create(
+            "test_simple_bottom_most_compaction_with_retain_lsns_single_key",
+        )
+        .await?;
         let (tenant, ctx) = harness.load().await;
 
         fn get_key(id: u32) -> Key {
@@ -9720,7 +9721,8 @@ mod tests {
     async fn test_simple_bottom_most_compaction_on_branch() -> anyhow::Result<()> {
         use models::CompactLsnRange;
 
-        let harness = TenantHarness::create("test_simple_bottom_most_compaction_on_branch").await?;
+        let harness =
+            TenantShardHarness::create("test_simple_bottom_most_compaction_on_branch").await?;
         let (tenant, ctx) = harness.load().await;
 
         fn get_key(id: u32) -> Key {
@@ -9950,7 +9952,8 @@ mod tests {
     #[cfg(feature = "testing")]
     #[tokio::test]
     async fn test_vectored_read_with_nested_image_layer() -> anyhow::Result<()> {
-        let harness = TenantHarness::create("test_vectored_read_with_nested_image_layer").await?;
+        let harness =
+            TenantShardHarness::create("test_vectored_read_with_nested_image_layer").await?;
         let (tenant, ctx) = harness.load().await;
 
         let will_init_keys = [2, 6];
@@ -10115,7 +10118,8 @@ mod tests {
     #[cfg(feature = "testing")]
     #[tokio::test]
     async fn test_simple_partial_bottom_most_compaction() -> anyhow::Result<()> {
-        let harness = TenantHarness::create("test_simple_partial_bottom_most_compaction").await?;
+        let harness =
+            TenantShardHarness::create("test_simple_partial_bottom_most_compaction").await?;
         let (tenant, ctx) = harness.load().await;
 
         fn get_key(id: u32) -> Key {
@@ -10457,7 +10461,7 @@ mod tests {
     #[cfg(feature = "testing")]
     #[tokio::test]
     async fn test_timeline_offload_retain_lsn() -> anyhow::Result<()> {
-        let harness = TenantHarness::create("test_timeline_offload_retain_lsn")
+        let harness = TenantShardHarness::create("test_timeline_offload_retain_lsn")
             .await
             .unwrap();
         let (tenant, ctx) = harness.load().await;
@@ -10507,7 +10511,8 @@ mod tests {
     #[cfg(feature = "testing")]
     #[tokio::test]
     async fn test_simple_bottom_most_compaction_above_lsn() -> anyhow::Result<()> {
-        let harness = TenantHarness::create("test_simple_bottom_most_compaction_above_lsn").await?;
+        let harness =
+            TenantShardHarness::create("test_simple_bottom_most_compaction_above_lsn").await?;
         let (tenant, ctx) = harness.load().await;
 
         fn get_key(id: u32) -> Key {
@@ -10758,7 +10763,8 @@ mod tests {
     #[cfg(feature = "testing")]
     #[tokio::test]
     async fn test_simple_bottom_most_compaction_rectangle() -> anyhow::Result<()> {
-        let harness = TenantHarness::create("test_simple_bottom_most_compaction_rectangle").await?;
+        let harness =
+            TenantShardHarness::create("test_simple_bottom_most_compaction_rectangle").await?;
         let (tenant, ctx) = harness.load().await;
 
         fn get_key(id: u32) -> Key {

pageserver/src/tenant/mgr.rs

@@ -44,7 +44,9 @@ use crate::tenant::config::{
 use crate::tenant::span::debug_assert_current_span_has_tenant_id;
 use crate::tenant::storage_layer::inmemory_layer;
 use crate::tenant::timeline::ShutdownMode;
-use crate::tenant::{AttachedTenantConf, GcError, LoadConfigError, SpawnMode, Tenant, TenantState};
+use crate::tenant::{
+    AttachedTenantConf, GcError, LoadConfigError, SpawnMode, TenantShard, TenantState,
+};
 use crate::virtual_file::MaybeFatalIo;
 use crate::{InitializationOrder, TEMP_FILE_SUFFIX};
 
@@ -69,7 +71,7 @@ use super::{GlobalShutDown, TenantSharedResources};
 /// having a properly acquired generation (Secondary doesn't need a generation)
 #[derive(Clone)]
 pub(crate) enum TenantSlot {
-    Attached(Arc<Tenant>),
+    Attached(Arc<TenantShard>),
     Secondary(Arc<SecondaryTenant>),
     /// In this state, other administrative operations acting on the TenantId should
     /// block, or return a retry indicator equivalent to HTTP 503.
@@ -88,7 +90,7 @@ impl std::fmt::Debug for TenantSlot {
 
 impl TenantSlot {
     /// Return the `Tenant` in this slot if attached, else None
-    fn get_attached(&self) -> Option<&Arc<Tenant>> {
+    fn get_attached(&self) -> Option<&Arc<TenantShard>> {
         match self {
             Self::Attached(t) => Some(t),
             Self::Secondary(_) => None,
@@ -99,13 +101,13 @@ impl TenantSlot {
 
 /// The tenants known to the pageserver.
 /// The enum variants are used to distinguish the different states that the pageserver can be in.
-pub(crate) enum TenantsMap {
+pub(crate) enum TenantShardMap {
     /// [`init_tenant_mgr`] is not done yet.
     Initializing,
     /// [`init_tenant_mgr`] is done, all on-disk tenants have been loaded.
     /// New tenants can be added using [`tenant_map_acquire_slot`].
     Open(BTreeMap<TenantShardId, TenantSlot>),
-    /// The pageserver has entered shutdown mode via [`TenantManager::shutdown`].
+    /// The pageserver has entered shutdown mode via [`TenantShardManager::shutdown`].
     /// Existing tenants are still accessible, but no new tenants can be created.
     ShuttingDown(BTreeMap<TenantShardId, TenantSlot>),
 }
@@ -166,19 +168,19 @@ impl TenantStartupMode {
 /// Result type for looking up a TenantId to a specific shard
 pub(crate) enum ShardResolveResult {
     NotFound,
-    Found(Arc<Tenant>),
+    Found(Arc<TenantShard>),
     // Wait for this barrrier, then query again
     InProgress(utils::completion::Barrier),
 }
 
-impl TenantsMap {
+impl TenantShardMap {
     /// Convenience function for typical usage, where we want to get a `Tenant` object, for
     /// working with attached tenants.  If the TenantId is in the map but in Secondary state,
     /// None is returned.
-    pub(crate) fn get(&self, tenant_shard_id: &TenantShardId) -> Option<&Arc<Tenant>> {
+    pub(crate) fn get(&self, tenant_shard_id: &TenantShardId) -> Option<&Arc<TenantShard>> {
         match self {
-            TenantsMap::Initializing => None,
-            TenantsMap::Open(m) | TenantsMap::ShuttingDown(m) => {
+            TenantShardMap::Initializing => None,
+            TenantShardMap::Open(m) | TenantShardMap::ShuttingDown(m) => {
                 m.get(tenant_shard_id).and_then(|slot| slot.get_attached())
             }
         }
@@ -187,8 +189,8 @@ impl TenantsMap {
     #[cfg(all(debug_assertions, not(test)))]
     pub(crate) fn len(&self) -> usize {
         match self {
-            TenantsMap::Initializing => 0,
-            TenantsMap::Open(m) | TenantsMap::ShuttingDown(m) => m.len(),
+            TenantShardMap::Initializing => 0,
+            TenantShardMap::Open(m) | TenantShardMap::ShuttingDown(m) => m.len(),
         }
     }
 }
@@ -284,23 +286,23 @@ impl BackgroundPurges {
     }
 }
 
-static TENANTS: Lazy<std::sync::RwLock<TenantsMap>> =
-    Lazy::new(|| std::sync::RwLock::new(TenantsMap::Initializing));
+static TENANTS: Lazy<std::sync::RwLock<TenantShardMap>> =
+    Lazy::new(|| std::sync::RwLock::new(TenantShardMap::Initializing));
 
 /// Responsible for storing and mutating the collection of all tenants
 /// that this pageserver has state for.
 ///
-/// Every Tenant and SecondaryTenant instance lives inside the TenantManager.
+/// Every TenantShard and SecondaryTenant instance lives inside the TenantShardManager.
 ///
-/// The most important role of the TenantManager is to prevent conflicts: e.g. trying to attach
+/// The most important role of the TenantShardManager is to prevent conflicts: e.g. trying to attach
 /// the same tenant twice concurrently, or trying to configure the same tenant into secondary
 /// and attached modes concurrently.
-pub struct TenantManager {
+pub struct TenantShardManager {
     conf: &'static PageServerConf,
     // TODO: currently this is a &'static pointing to TENANTs.  When we finish refactoring
     // out of that static variable, the TenantManager can own this.
     // See https://github.com/neondatabase/neon/issues/5796
-    tenants: &'static std::sync::RwLock<TenantsMap>,
+    tenants: &'static std::sync::RwLock<TenantShardMap>,
     resources: TenantSharedResources,
 
     // Long-running operations that happen outside of a [`Tenant`] lifetime should respect this token.
@@ -412,7 +414,7 @@ fn load_tenant_config(
         return None;
     }
 
-    Some(Tenant::load_tenant_config(conf, &tenant_shard_id))
+    Some(TenantShard::load_tenant_config(conf, &tenant_shard_id))
 }
 
 /// Initial stage of load: walk the local tenants directory, clean up any temp files,
@@ -491,7 +493,7 @@ pub async fn init_tenant_mgr(
     resources: TenantSharedResources,
     init_order: InitializationOrder,
     cancel: CancellationToken,
-) -> anyhow::Result<TenantManager> {
+) -> anyhow::Result<TenantShardManager> {
     let mut tenants = BTreeMap::new();
 
     let ctx = RequestContext::todo_child(TaskKind::Startup, DownloadBehavior::Warn);
@@ -606,7 +608,8 @@ pub async fn init_tenant_mgr(
         // Presence of a generation number implies attachment: attach the tenant
         // if it wasn't already, and apply the generation number.
         config_write_futs.push(async move {
-            let r = Tenant::persist_tenant_config(conf, &tenant_shard_id, &location_conf).await;
+            let r =
+                TenantShard::persist_tenant_config(conf, &tenant_shard_id, &location_conf).await;
             (tenant_shard_id, location_conf, r)
         });
     }
@@ -669,11 +672,11 @@ pub async fn init_tenant_mgr(
     info!("Processed {} local tenants at startup", tenants.len());
 
     let mut tenants_map = TENANTS.write().unwrap();
-    assert!(matches!(&*tenants_map, &TenantsMap::Initializing));
+    assert!(matches!(&*tenants_map, &TenantShardMap::Initializing));
 
-    *tenants_map = TenantsMap::Open(tenants);
+    *tenants_map = TenantShardMap::Open(tenants);
 
-    Ok(TenantManager {
+    Ok(TenantShardManager {
         conf,
         tenants: &TENANTS,
         resources,
@@ -682,7 +685,7 @@ pub async fn init_tenant_mgr(
     })
 }
 
-/// Wrapper for Tenant::spawn that checks invariants before running
+/// Wrapper for TenantShard::spawn that checks invariants before running
 #[allow(clippy::too_many_arguments)]
 fn tenant_spawn(
     conf: &'static PageServerConf,
@@ -694,7 +697,7 @@ fn tenant_spawn(
     init_order: Option<InitializationOrder>,
     mode: SpawnMode,
     ctx: &RequestContext,
-) -> Result<Arc<Tenant>, GlobalShutDown> {
+) -> Result<Arc<TenantShard>, GlobalShutDown> {
     // All these conditions should have been satisfied by our caller: the tenant dir exists, is a well formed
     // path, and contains a configuration file.  Assertions that do synchronous I/O are limited to debug mode
     // to avoid impacting prod runtime performance.
@@ -705,7 +708,7 @@ fn tenant_spawn(
         .try_exists()
         .unwrap());
 
-    Tenant::spawn(
+    TenantShard::spawn(
         conf,
         tenant_shard_id,
         resources,
@@ -717,7 +720,7 @@ fn tenant_spawn(
     )
 }
 
-async fn shutdown_all_tenants0(tenants: &std::sync::RwLock<TenantsMap>) {
+async fn shutdown_all_tenants0(tenants: &std::sync::RwLock<TenantShardMap>) {
     let mut join_set = JoinSet::new();
 
     #[cfg(all(debug_assertions, not(test)))]
@@ -732,12 +735,12 @@ async fn shutdown_all_tenants0(tenants: &std::sync::RwLock<TenantsMap>) {
     let (total_in_progress, total_attached) = {
         let mut m = tenants.write().unwrap();
         match &mut *m {
-            TenantsMap::Initializing => {
-                *m = TenantsMap::ShuttingDown(BTreeMap::default());
+            TenantShardMap::Initializing => {
+                *m = TenantShardMap::ShuttingDown(BTreeMap::default());
                 info!("tenants map is empty");
                 return;
             }
-            TenantsMap::Open(tenants) => {
+            TenantShardMap::Open(tenants) => {
                 let mut shutdown_state = BTreeMap::new();
                 let mut total_in_progress = 0;
                 let mut total_attached = 0;
@@ -787,10 +790,10 @@ async fn shutdown_all_tenants0(tenants: &std::sync::RwLock<TenantsMap>) {
                         }
                     }
                 }
-                *m = TenantsMap::ShuttingDown(shutdown_state);
+                *m = TenantShardMap::ShuttingDown(shutdown_state);
                 (total_in_progress, total_attached)
             }
-            TenantsMap::ShuttingDown(_) => {
+            TenantShardMap::ShuttingDown(_) => {
                 error!("already shutting down, this function isn't supposed to be called more than once");
                 return;
             }
@@ -870,7 +873,7 @@ pub(crate) enum UpsertLocationError {
     InternalError(anyhow::Error),
 }
 
-impl TenantManager {
+impl TenantShardManager {
     /// Convenience function so that anyone with a TenantManager can get at the global configuration, without
     /// having to pass it around everywhere as a separate object.
     pub(crate) fn get_conf(&self) -> &'static PageServerConf {
@@ -881,11 +884,11 @@ impl TenantManager {
     /// undergoing a state change (i.e. slot is InProgress).
     ///
     /// The return Tenant is not guaranteed to be active: check its status after obtaing it, or
-    /// use [`Tenant::wait_to_become_active`] before using it if you will do I/O on it.
+    /// use [`TenantShard::wait_to_become_active`] before using it if you will do I/O on it.
     pub(crate) fn get_attached_tenant_shard(
         &self,
         tenant_shard_id: TenantShardId,
-    ) -> Result<Arc<Tenant>, GetTenantError> {
+    ) -> Result<Arc<TenantShard>, GetTenantError> {
         let locked = self.tenants.read().unwrap();
 
         let peek_slot = tenant_map_peek_slot(&locked, &tenant_shard_id, TenantSlotPeekMode::Read)?;
@@ -934,12 +937,12 @@ impl TenantManager {
         flush: Option<Duration>,
         mut spawn_mode: SpawnMode,
         ctx: &RequestContext,
-    ) -> Result<Option<Arc<Tenant>>, UpsertLocationError> {
+    ) -> Result<Option<Arc<TenantShard>>, UpsertLocationError> {
         debug_assert_current_span_has_tenant_id();
         info!("configuring tenant location to state {new_location_config:?}");
 
         enum FastPathModified {
-            Attached(Arc<Tenant>),
+            Attached(Arc<TenantShard>),
             Secondary(Arc<SecondaryTenant>),
         }
 
@@ -996,9 +999,13 @@ impl TenantManager {
         // phase of writing config and/or waiting for flush, before returning.
         match fast_path_taken {
             Some(FastPathModified::Attached(tenant)) => {
-                Tenant::persist_tenant_config(self.conf, &tenant_shard_id, &new_location_config)
-                    .await
-                    .fatal_err("write tenant shard config");
+                TenantShard::persist_tenant_config(
+                    self.conf,
+                    &tenant_shard_id,
+                    &new_location_config,
+                )
+                .await
+                .fatal_err("write tenant shard config");
 
                 // Transition to AttachedStale means we may well hold a valid generation
                 // still, and have been requested to go stale as part of a migration.  If
@@ -1027,9 +1034,13 @@ impl TenantManager {
                 return Ok(Some(tenant));
             }
             Some(FastPathModified::Secondary(_secondary_tenant)) => {
-                Tenant::persist_tenant_config(self.conf, &tenant_shard_id, &new_location_config)
-                    .await
-                    .fatal_err("write tenant shard config");
+                TenantShard::persist_tenant_config(
+                    self.conf,
+                    &tenant_shard_id,
+                    &new_location_config,
+                )
+                .await
+                .fatal_err("write tenant shard config");
 
                 return Ok(None);
             }
@@ -1119,7 +1130,7 @@ impl TenantManager {
         // Before activating either secondary or attached mode, persist the
         // configuration, so that on restart we will re-attach (or re-start
         // secondary) on the tenant.
-        Tenant::persist_tenant_config(self.conf, &tenant_shard_id, &new_location_config)
+        TenantShard::persist_tenant_config(self.conf, &tenant_shard_id, &new_location_config)
             .await
             .fatal_err("write tenant shard config");
 
@@ -1257,7 +1268,7 @@ impl TenantManager {
 
         let tenant_path = self.conf.tenant_path(&tenant_shard_id);
         let timelines_path = self.conf.timelines_path(&tenant_shard_id);
-        let config = Tenant::load_tenant_config(self.conf, &tenant_shard_id)?;
+        let config = TenantShard::load_tenant_config(self.conf, &tenant_shard_id)?;
 
         if drop_cache {
             tracing::info!("Dropping local file cache");
@@ -1292,11 +1303,11 @@ impl TenantManager {
         Ok(())
     }
 
-    pub(crate) fn get_attached_active_tenant_shards(&self) -> Vec<Arc<Tenant>> {
+    pub(crate) fn get_attached_active_tenant_shards(&self) -> Vec<Arc<TenantShard>> {
         let locked = self.tenants.read().unwrap();
         match &*locked {
-            TenantsMap::Initializing => Vec::new(),
-            TenantsMap::Open(map) | TenantsMap::ShuttingDown(map) => map
+            TenantShardMap::Initializing => Vec::new(),
+            TenantShardMap::Open(map) | TenantShardMap::ShuttingDown(map) => map
                 .values()
                 .filter_map(|slot| {
                     slot.get_attached()
@@ -1316,8 +1327,8 @@ impl TenantManager {
         let locked = self.tenants.read().unwrap();
 
         let map = match &*locked {
-            TenantsMap::Initializing | TenantsMap::ShuttingDown(_) => return,
-            TenantsMap::Open(m) => m,
+            TenantShardMap::Initializing | TenantShardMap::ShuttingDown(_) => return,
+            TenantShardMap::Open(m) => m,
         };
 
         for (tenant_id, slot) in map {
@@ -1334,8 +1345,8 @@ impl TenantManager {
     pub(crate) fn list(&self) -> Vec<(TenantShardId, TenantSlot)> {
         let locked = self.tenants.read().unwrap();
         match &*locked {
-            TenantsMap::Initializing => Vec::new(),
-            TenantsMap::Open(map) | TenantsMap::ShuttingDown(map) => {
+            TenantShardMap::Initializing => Vec::new(),
+            TenantShardMap::Open(map) | TenantShardMap::ShuttingDown(map) => {
                 map.iter().map(|(k, v)| (*k, v.clone())).collect()
             }
         }
@@ -1344,8 +1355,8 @@ impl TenantManager {
     pub(crate) fn get(&self, tenant_shard_id: TenantShardId) -> Option<TenantSlot> {
         let locked = self.tenants.read().unwrap();
         match &*locked {
-            TenantsMap::Initializing => None,
-            TenantsMap::Open(map) | TenantsMap::ShuttingDown(map) => {
+            TenantShardMap::Initializing => None,
+            TenantShardMap::Open(map) | TenantShardMap::ShuttingDown(map) => {
                 map.get(&tenant_shard_id).cloned()
             }
         }
@@ -1441,7 +1452,7 @@ impl TenantManager {
     #[instrument(skip_all, fields(tenant_id=%tenant.get_tenant_shard_id().tenant_id, shard_id=%tenant.get_tenant_shard_id().shard_slug(), new_shard_count=%new_shard_count.literal()))]
     pub(crate) async fn shard_split(
         &self,
-        tenant: Arc<Tenant>,
+        tenant: Arc<TenantShard>,
         new_shard_count: ShardCount,
         new_stripe_size: Option<ShardStripeSize>,
         ctx: &RequestContext,
@@ -1471,7 +1482,7 @@ impl TenantManager {
 
     pub(crate) async fn do_shard_split(
         &self,
-        tenant: Arc<Tenant>,
+        tenant: Arc<TenantShard>,
         new_shard_count: ShardCount,
         new_stripe_size: Option<ShardStripeSize>,
         ctx: &RequestContext,
@@ -1519,7 +1530,7 @@ impl TenantManager {
 
         // Phase 1: Write out child shards' remote index files, in the parent tenant's current generation
         if let Err(e) = tenant.split_prepare(&child_shards).await {
-            // If [`Tenant::split_prepare`] fails, we must reload the tenant, because it might
+            // If [`TenantShard::split_prepare`] fails, we must reload the tenant, because it might
             // have been left in a partially-shut-down state.
             tracing::warn!("Failed to prepare for split: {e}, reloading Tenant before returning");
             return Err(e);
@@ -1697,7 +1708,7 @@ impl TenantManager {
     /// For each resident layer in the parent shard, we will hard link it into all of the child shards.
     async fn shard_split_hardlink(
         &self,
-        parent_shard: &Tenant,
+        parent_shard: &TenantShard,
         child_shards: Vec<TenantShardId>,
     ) -> anyhow::Result<()> {
         debug_assert_current_span_has_tenant_id();
@@ -1888,8 +1899,8 @@ impl TenantManager {
     ) -> Result<Vec<(TenantShardId, TenantState, Generation)>, TenantMapListError> {
         let tenants = self.tenants.read().unwrap();
         let m = match &*tenants {
-            TenantsMap::Initializing => return Err(TenantMapListError::Initializing),
-            TenantsMap::Open(m) | TenantsMap::ShuttingDown(m) => m,
+            TenantShardMap::Initializing => return Err(TenantMapListError::Initializing),
+            TenantShardMap::Open(m) | TenantShardMap::ShuttingDown(m) => m,
         };
         Ok(m.iter()
             .filter_map(|(id, tenant)| match tenant {
@@ -1974,7 +1985,7 @@ impl TenantManager {
             }
 
             let tenant_path = self.conf.tenant_path(&tenant_shard_id);
-            let config = Tenant::load_tenant_config(self.conf, &tenant_shard_id)
+            let config = TenantShard::load_tenant_config(self.conf, &tenant_shard_id)
                 .map_err(|e| Error::DetachReparent(e.into()))?;
 
             let shard_identity = config.shard;
@@ -2070,8 +2081,8 @@ impl TenantManager {
         let mut any_in_progress = None;
 
         match &*tenants {
-            TenantsMap::Initializing => ShardResolveResult::NotFound,
-            TenantsMap::Open(m) | TenantsMap::ShuttingDown(m) => {
+            TenantShardMap::Initializing => ShardResolveResult::NotFound,
+            TenantShardMap::Open(m) | TenantShardMap::ShuttingDown(m) => {
                 for slot in m.range(TenantShardId::tenant_range(*tenant_id)) {
                     // Ignore all slots that don't contain an attached tenant
                     let tenant = match &slot.1 {
@@ -2135,8 +2146,8 @@ impl TenantManager {
     pub(crate) fn calculate_utilization(&self) -> Result<(u64, u32), TenantMapListError> {
         let tenants = self.tenants.read().unwrap();
         let m = match &*tenants {
-            TenantsMap::Initializing => return Err(TenantMapListError::Initializing),
-            TenantsMap::Open(m) | TenantsMap::ShuttingDown(m) => m,
+            TenantShardMap::Initializing => return Err(TenantMapListError::Initializing),
+            TenantShardMap::Open(m) | TenantShardMap::ShuttingDown(m) => m,
         };
         let shard_count = m.len();
         let mut wanted_bytes = 0;
@@ -2482,18 +2493,18 @@ impl SlotGuard {
             }
 
             let m = match &mut *locked {
-                TenantsMap::Initializing => {
+                TenantShardMap::Initializing => {
                     return Err(TenantSlotUpsertError::MapState(
                         TenantMapError::StillInitializing,
                     ))
                 }
-                TenantsMap::ShuttingDown(_) => {
+                TenantShardMap::ShuttingDown(_) => {
                     return Err(TenantSlotUpsertError::ShuttingDown((
                         new_value,
                         self.completion.clone(),
                     )));
                 }
-                TenantsMap::Open(m) => m,
+                TenantShardMap::Open(m) => m,
             };
 
             METRICS.slot_inserted(&new_value);
@@ -2590,17 +2601,17 @@ impl Drop for SlotGuard {
         let mut locked = TENANTS.write().unwrap();
 
         let m = match &mut *locked {
-            TenantsMap::Initializing => {
+            TenantShardMap::Initializing => {
                 // There is no map, this should never happen.
                 return;
             }
-            TenantsMap::ShuttingDown(_) => {
+            TenantShardMap::ShuttingDown(_) => {
                 // When we transition to shutdown, InProgress elements are removed
                 // from the map, so we do not need to clean up our Inprogress marker.
                 // See [`shutdown_all_tenants0`]
                 return;
             }
-            TenantsMap::Open(m) => m,
+            TenantShardMap::Open(m) => m,
         };
 
         use std::collections::btree_map::Entry;
@@ -2640,13 +2651,13 @@ enum TenantSlotPeekMode {
 }
 
 fn tenant_map_peek_slot<'a>(
-    tenants: &'a std::sync::RwLockReadGuard<'a, TenantsMap>,
+    tenants: &'a std::sync::RwLockReadGuard<'a, TenantShardMap>,
     tenant_shard_id: &TenantShardId,
     mode: TenantSlotPeekMode,
 ) -> Result<Option<&'a TenantSlot>, TenantMapError> {
     match tenants.deref() {
-        TenantsMap::Initializing => Err(TenantMapError::StillInitializing),
-        TenantsMap::ShuttingDown(m) => match mode {
+        TenantShardMap::Initializing => Err(TenantMapError::StillInitializing),
+        TenantShardMap::ShuttingDown(m) => match mode {
             TenantSlotPeekMode::Read => Ok(Some(
                 // When reading in ShuttingDown state, we must translate None results
                 // into a ShuttingDown error, because absence of a tenant shard ID in the map
@@ -2658,7 +2669,7 @@ fn tenant_map_peek_slot<'a>(
             )),
             TenantSlotPeekMode::Write => Err(TenantMapError::ShuttingDown),
         },
-        TenantsMap::Open(m) => Ok(m.get(tenant_shard_id)),
+        TenantShardMap::Open(m) => Ok(m.get(tenant_shard_id)),
     }
 }
 
@@ -2678,7 +2689,7 @@ fn tenant_map_acquire_slot(
 
 fn tenant_map_acquire_slot_impl(
     tenant_shard_id: &TenantShardId,
-    tenants: &std::sync::RwLock<TenantsMap>,
+    tenants: &std::sync::RwLock<TenantShardMap>,
     mode: TenantSlotAcquireMode,
 ) -> Result<SlotGuard, TenantSlotError> {
     use TenantSlotAcquireMode::*;
@@ -2689,9 +2700,9 @@ fn tenant_map_acquire_slot_impl(
     let _guard = span.enter();
 
     let m = match &mut *locked {
-        TenantsMap::Initializing => return Err(TenantMapError::StillInitializing.into()),
-        TenantsMap::ShuttingDown(_) => return Err(TenantMapError::ShuttingDown.into()),
-        TenantsMap::Open(m) => m,
+        TenantShardMap::Initializing => return Err(TenantMapError::StillInitializing.into()),
+        TenantShardMap::ShuttingDown(_) => return Err(TenantMapError::ShuttingDown.into()),
+        TenantShardMap::Open(m) => m,
     };
 
     use std::collections::btree_map::Entry;
@@ -2744,7 +2755,7 @@ fn tenant_map_acquire_slot_impl(
 /// If the cleanup fails, tenant will stay in memory in [`TenantState::Broken`] state, and another removal
 /// operation would be needed to remove it.
 async fn remove_tenant_from_memory<V, F>(
-    tenants: &std::sync::RwLock<TenantsMap>,
+    tenants: &std::sync::RwLock<TenantShardMap>,
     tenant_shard_id: TenantShardId,
     tenant_cleanup: F,
 ) -> Result<V, TenantStateError>
@@ -2827,14 +2838,14 @@ mod tests {
 
     use crate::tenant::mgr::TenantSlot;
 
-    use super::{super::harness::TenantHarness, TenantsMap};
+    use super::{super::harness::TenantShardHarness, TenantShardMap};
 
     #[tokio::test(start_paused = true)]
     async fn shutdown_awaits_in_progress_tenant() {
         // Test that if an InProgress tenant is in the map during shutdown, the shutdown will gracefully
         // wait for it to complete before proceeding.
 
-        let h = TenantHarness::create("shutdown_awaits_in_progress_tenant")
+        let h = TenantShardHarness::create("shutdown_awaits_in_progress_tenant")
             .await
             .unwrap();
         let (t, _ctx) = h.load().await;
@@ -2848,7 +2859,7 @@ mod tests {
         let _e = span.enter();
 
         let tenants = BTreeMap::from([(id, TenantSlot::Attached(t.clone()))]);
-        let tenants = Arc::new(std::sync::RwLock::new(TenantsMap::Open(tenants)));
+        let tenants = Arc::new(std::sync::RwLock::new(TenantShardMap::Open(tenants)));
 
         // Invoke remove_tenant_from_memory with a cleanup hook that blocks until we manually
         // permit it to proceed: that will stick the tenant in InProgress

pageserver/src/tenant/remote_timeline_client.rs

@@ -133,7 +133,7 @@
 //! - Initiate upload queue with that [`IndexPart`].
 //! - Reschedule all lost operations by comparing the local filesystem state
 //!   and remote state as per [`IndexPart`]. This is done in
-//!   [`Tenant::timeline_init_and_sync`].
+//!   [`TenantShard::timeline_init_and_sync`].
 //!
 //! Note that if we crash during file deletion between the index update
 //! that removes the file from the list of files, and deleting the remote file,
@@ -171,7 +171,7 @@
 //! If no remote storage configuration is provided, the [`RemoteTimelineClient`] is
 //! not created and the uploads are skipped.
 //!
-//! [`Tenant::timeline_init_and_sync`]: super::Tenant::timeline_init_and_sync
+//! [`TenantShard::timeline_init_and_sync`]: super::TenantShard::timeline_init_and_sync
 //! [`Timeline::load_layer_map`]: super::Timeline::load_layer_map
 
 pub(crate) mod download;
@@ -382,12 +382,6 @@ pub(crate) struct RemoteTimelineClient {
     cancel: CancellationToken,
 }
 
-impl Drop for RemoteTimelineClient {
-    fn drop(&mut self) {
-        debug!("dropping RemoteTimelineClient");
-    }
-}
-
 impl RemoteTimelineClient {
     ///
     /// Create a remote storage client for given timeline
@@ -803,12 +797,6 @@ impl RemoteTimelineClient {
 
         upload_queue.dirty.metadata.apply(update);
 
-        // Defense in depth: if we somehow generated invalid metadata, do not persist it.
-        upload_queue
-            .dirty
-            .validate()
-            .map_err(|e| anyhow::anyhow!(e))?;
-
         self.schedule_index_upload(upload_queue);
 
         Ok(())
@@ -2650,9 +2638,9 @@ mod tests {
         context::RequestContext,
         tenant::{
             config::AttachmentMode,
-            harness::{TenantHarness, TIMELINE_ID},
+            harness::{TenantShardHarness, TIMELINE_ID},
             storage_layer::layer::local_layer_path,
-            Tenant, Timeline,
+            TenantShard, Timeline,
         },
         DEFAULT_PG_VERSION,
     };
@@ -2709,8 +2697,8 @@ mod tests {
     }
 
     struct TestSetup {
-        harness: TenantHarness,
-        tenant: Arc<Tenant>,
+        harness: TenantShardHarness,
+        tenant: Arc<TenantShard>,
         timeline: Arc<Timeline>,
         tenant_ctx: RequestContext,
     }
@@ -2718,7 +2706,7 @@ mod tests {
     impl TestSetup {
         async fn new(test_name: &str) -> anyhow::Result<Self> {
             let test_name = Box::leak(Box::new(format!("remote_timeline_client__{test_name}")));
-            let harness = TenantHarness::create(test_name).await?;
+            let harness = TenantShardHarness::create(test_name).await?;
             let (tenant, ctx) = harness.load().await;
 
             let timeline = tenant

pageserver/src/tenant/remote_timeline_client/index.rs

@@ -152,21 +152,6 @@ impl IndexPart {
         };
         is_same_remote_layer_path(name, metadata, name, index_metadata)
     }
-
-    /// Check for invariants in the index: this is useful when uploading an index to ensure that if
-    /// we encounter a bug, we do not persist buggy metadata.
-    pub(crate) fn validate(&self) -> Result<(), String> {
-        if self.import_pgdata.is_none()
-            && self.metadata.ancestor_timeline().is_none()
-            && self.layer_metadata.is_empty()
-        {
-            // Unless we're in the middle of a raw pgdata import, or this is a child timeline,the index must
-            // always have at least one layer.
-            return Err("Index has no ancestor and no layers".to_string());
-        }
-
-        Ok(())
-    }
 }
 
 /// Metadata gathered for each of the layer files.

pageserver/src/tenant/remote_timeline_client/upload.rs

@@ -40,10 +40,6 @@ pub(crate) async fn upload_index_part(
     });
     pausable_failpoint!("before-upload-index-pausable");
 
-    // Safety: refuse to persist invalid index metadata, to mitigate the impact of any bug that produces this
-    // (this should never happen)
-    index_part.validate().map_err(|e| anyhow::anyhow!(e))?;
-
     // FIXME: this error comes too late
     let serialized = index_part.to_json_bytes()?;
     let serialized = Bytes::from(serialized);

pageserver/src/tenant/secondary.rs

@@ -19,7 +19,7 @@ use self::{
 
 use super::{
     config::{SecondaryLocationConfig, TenantConfOpt},
-    mgr::TenantManager,
+    mgr::TenantShardManager,
     span::debug_assert_current_span_has_tenant_id,
     storage_layer::LayerName,
     GetTenantError,
@@ -374,7 +374,7 @@ impl GlobalTasks {
 }
 
 pub fn spawn_tasks(
-    tenant_manager: Arc<TenantManager>,
+    tenant_manager: Arc<TenantShardManager>,
     remote_storage: GenericRemoteStorage,
     background_jobs_can_start: Barrier,
     cancel: CancellationToken,

pageserver/src/tenant/secondary/downloader.rs

@@ -39,7 +39,7 @@ use super::{
 };
 
 use crate::tenant::{
-    mgr::TenantManager,
+    mgr::TenantShardManager,
     remote_timeline_client::{download::download_layer_file, remote_heatmap_path},
 };
 
@@ -69,7 +69,7 @@ use super::{
 const DEFAULT_DOWNLOAD_INTERVAL: Duration = Duration::from_millis(60000);
 
 pub(super) async fn downloader_task(
-    tenant_manager: Arc<TenantManager>,
+    tenant_manager: Arc<TenantShardManager>,
     remote_storage: GenericRemoteStorage,
     command_queue: tokio::sync::mpsc::Receiver<CommandRequest<DownloadCommand>>,
     background_jobs_can_start: Barrier,
@@ -92,7 +92,7 @@ pub(super) async fn downloader_task(
 }
 
 struct SecondaryDownloader {
-    tenant_manager: Arc<TenantManager>,
+    tenant_manager: Arc<TenantShardManager>,
     remote_storage: GenericRemoteStorage,
     root_ctx: RequestContext,
 }

pageserver/src/tenant/secondary/heatmap_uploader.rs

@@ -10,11 +10,11 @@ use crate::{
     tenant::{
         config::AttachmentMode,
         mgr::GetTenantError,
-        mgr::TenantManager,
+        mgr::TenantShardManager,
         remote_timeline_client::remote_heatmap_path,
         span::debug_assert_current_span_has_tenant_id,
         tasks::{warn_when_period_overrun, BackgroundLoopKind},
-        Tenant,
+        TenantShard,
     },
 };
 
@@ -35,7 +35,7 @@ use tracing::{info_span, instrument, Instrument};
 use utils::{backoff, completion::Barrier, yielding_loop::yielding_loop};
 
 pub(super) async fn heatmap_uploader_task(
-    tenant_manager: Arc<TenantManager>,
+    tenant_manager: Arc<TenantShardManager>,
     remote_storage: GenericRemoteStorage,
     command_queue: tokio::sync::mpsc::Receiver<CommandRequest<UploadCommand>>,
     background_jobs_can_start: Barrier,
@@ -61,7 +61,7 @@ pub(super) async fn heatmap_uploader_task(
 /// handling loop and mutates it as needed: there are no locks here, because that event loop
 /// can hold &mut references to this type throughout.
 struct HeatmapUploader {
-    tenant_manager: Arc<TenantManager>,
+    tenant_manager: Arc<TenantShardManager>,
     remote_storage: GenericRemoteStorage,
     cancel: CancellationToken,
 
@@ -79,7 +79,7 @@ impl RunningJob for WriteInProgress {
 }
 
 struct UploadPending {
-    tenant: Arc<Tenant>,
+    tenant: Arc<TenantShard>,
     last_upload: Option<LastUploadState>,
     target_time: Option<Instant>,
     period: Option<Duration>,
@@ -111,7 +111,7 @@ impl scheduler::Completion for WriteComplete {
 struct UploaderTenantState {
     // This Weak only exists to enable culling idle instances of this type
     // when the Tenant has been deallocated.
-    tenant: Weak<Tenant>,
+    tenant: Weak<TenantShard>,
 
     /// Digest of the serialized heatmap that we last successfully uploaded
     last_upload_state: Option<LastUploadState>,
@@ -362,7 +362,7 @@ struct LastUploadState {
 /// of the object we would have uploaded.
 async fn upload_tenant_heatmap(
     remote_storage: GenericRemoteStorage,
-    tenant: &Arc<Tenant>,
+    tenant: &Arc<TenantShard>,
     last_upload: Option<LastUploadState>,
 ) -> Result<UploadHeatmapOutcome, UploadHeatmapError> {
     debug_assert_current_span_has_tenant_id();

pageserver/src/tenant/secondary/scheduler.rs

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

pageserver/src/tenant/size.rs

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

pageserver/src/tenant/storage_layer/batch_split_writer.rs

@@ -404,7 +404,7 @@ mod tests {
 
     use crate::{
         tenant::{
-            harness::{TenantHarness, TIMELINE_ID},
+            harness::{TenantShardHarness, TIMELINE_ID},
             storage_layer::AsLayerDesc,
         },
         DEFAULT_PG_VERSION,
@@ -431,7 +431,7 @@ mod tests {
 
     #[tokio::test]
     async fn write_one_image() {
-        let harness = TenantHarness::create("split_writer_write_one_image")
+        let harness = TenantShardHarness::create("split_writer_write_one_image")
             .await
             .unwrap();
         let (tenant, ctx) = harness.load().await;
@@ -510,7 +510,7 @@ mod tests {
     /// Test the image+delta writer by writing a large number of images and deltas. If discard is
     /// set to true, all layers will be discarded.
     async fn write_split_helper(harness_name: &'static str, discard: bool) {
-        let harness = TenantHarness::create(harness_name).await.unwrap();
+        let harness = TenantShardHarness::create(harness_name).await.unwrap();
         let (tenant, ctx) = harness.load().await;
 
         let tline = tenant
@@ -605,7 +605,7 @@ mod tests {
 
     #[tokio::test]
     async fn write_large_img() {
-        let harness = TenantHarness::create("split_writer_write_large_img")
+        let harness = TenantShardHarness::create("split_writer_write_large_img")
             .await
             .unwrap();
         let (tenant, ctx) = harness.load().await;
@@ -692,7 +692,7 @@ mod tests {
 
     #[tokio::test]
     async fn write_split_single_key() {
-        let harness = TenantHarness::create("split_writer_write_split_single_key")
+        let harness = TenantShardHarness::create("split_writer_write_split_single_key")
             .await
             .unwrap();
         let (tenant, ctx) = harness.load().await;

pageserver/src/tenant/storage_layer/delta_layer.rs

@@ -1618,11 +1618,11 @@ pub(crate) mod test {
     use crate::tenant::harness::TIMELINE_ID;
     use crate::tenant::storage_layer::{Layer, ResidentLayer};
     use crate::tenant::vectored_blob_io::StreamingVectoredReadPlanner;
-    use crate::tenant::{Tenant, Timeline};
+    use crate::tenant::{TenantShard, Timeline};
     use crate::{
         context::DownloadBehavior,
         task_mgr::TaskKind,
-        tenant::{disk_btree::tests::TestDisk, harness::TenantHarness},
+        tenant::{disk_btree::tests::TestDisk, harness::TenantShardHarness},
         DEFAULT_PG_VERSION,
     };
     use bytes::Bytes;
@@ -1879,7 +1879,8 @@ pub(crate) mod test {
 
     #[tokio::test]
     async fn test_delta_layer_vectored_read_end_to_end() -> anyhow::Result<()> {
-        let harness = TenantHarness::create("test_delta_layer_oversized_vectored_read").await?;
+        let harness =
+            TenantShardHarness::create("test_delta_layer_oversized_vectored_read").await?;
         let (tenant, ctx) = harness.load().await;
 
         let timeline_id = TimelineId::generate();
@@ -1982,7 +1983,7 @@ pub(crate) mod test {
         use bytes::Bytes;
         use pageserver_api::record::NeonWalRecord;
 
-        let h = crate::tenant::harness::TenantHarness::create("truncate_delta_smoke")
+        let h = crate::tenant::harness::TenantShardHarness::create("truncate_delta_smoke")
             .await
             .unwrap();
         let (tenant, ctx) = h.load().await;
@@ -2214,7 +2215,7 @@ pub(crate) mod test {
     }
 
     pub(crate) async fn produce_delta_layer(
-        tenant: &Tenant,
+        tenant: &TenantShard,
         tline: &Arc<Timeline>,
         mut deltas: Vec<(Key, Lsn, Value)>,
         ctx: &RequestContext,
@@ -2268,7 +2269,9 @@ pub(crate) mod test {
 
     #[tokio::test]
     async fn delta_layer_iterator() {
-        let harness = TenantHarness::create("delta_layer_iterator").await.unwrap();
+        let harness = TenantShardHarness::create("delta_layer_iterator")
+            .await
+            .unwrap();
         let (tenant, ctx) = harness.load().await;
 
         let tline = tenant

pageserver/src/tenant/storage_layer/filter_iterator.rs

@@ -106,7 +106,7 @@ mod tests {
 
     use crate::{
         tenant::{
-            harness::{TenantHarness, TIMELINE_ID},
+            harness::{TenantShardHarness, TIMELINE_ID},
             storage_layer::delta_layer::test::produce_delta_layer,
         },
         DEFAULT_PG_VERSION,
@@ -137,7 +137,7 @@ mod tests {
         use bytes::Bytes;
         use pageserver_api::value::Value;
 
-        let harness = TenantHarness::create("filter_iterator_filter_keyspace_iterator")
+        let harness = TenantShardHarness::create("filter_iterator_filter_keyspace_iterator")
             .await
             .unwrap();
         let (tenant, ctx) = harness.load().await;

pageserver/src/tenant/storage_layer/image_layer.rs

@@ -1138,10 +1138,10 @@ mod test {
         context::RequestContext,
         tenant::{
             config::TenantConf,
-            harness::{TenantHarness, TIMELINE_ID},
+            harness::{TenantShardHarness, TIMELINE_ID},
             storage_layer::{Layer, ResidentLayer},
             vectored_blob_io::StreamingVectoredReadPlanner,
-            Tenant, Timeline,
+            TenantShard, Timeline,
         },
         DEFAULT_PG_VERSION,
     };
@@ -1170,7 +1170,7 @@ mod test {
         // Create an unsharded parent with a layer.
         //
 
-        let harness = TenantHarness::create_custom(
+        let harness = TenantShardHarness::create_custom(
             "test_image_layer_rewrite--parent",
             tenant_conf.clone(),
             tenant_id,
@@ -1233,7 +1233,7 @@ mod test {
                 ShardStripeSize(0x8000),
             )
             .unwrap();
-            let harness = TenantHarness::create_custom(
+            let harness = TenantShardHarness::create_custom(
                 Box::leak(Box::new(format!(
                     "test_image_layer_rewrite--child{}",
                     shard_identity.shard_slug()
@@ -1324,7 +1324,7 @@ mod test {
     }
 
     async fn produce_image_layer(
-        tenant: &Tenant,
+        tenant: &TenantShard,
         tline: &Arc<Timeline>,
         mut images: Vec<(Key, Bytes)>,
         lsn: Lsn,
@@ -1380,7 +1380,9 @@ mod test {
 
     #[tokio::test]
     async fn image_layer_iterator() {
-        let harness = TenantHarness::create("image_layer_iterator").await.unwrap();
+        let harness = TenantShardHarness::create("image_layer_iterator")
+            .await
+            .unwrap();
         let (tenant, ctx) = harness.load().await;
 
         let tline = tenant

pageserver/src/tenant/storage_layer/layer/tests.rs

@@ -1,6 +1,6 @@
 use std::time::UNIX_EPOCH;
 
-use pageserver_api::key::{Key, CONTROLFILE_KEY};
+use pageserver_api::key::CONTROLFILE_KEY;
 use tokio::task::JoinSet;
 use utils::{
     completion::{self, Completion},
@@ -9,11 +9,8 @@ use utils::{
 
 use super::failpoints::{Failpoint, FailpointKind};
 use super::*;
-use crate::{
-    context::DownloadBehavior,
-    tenant::{harness::test_img, storage_layer::LayerVisibilityHint},
-};
-use crate::{task_mgr::TaskKind, tenant::harness::TenantHarness};
+use crate::{context::DownloadBehavior, tenant::storage_layer::LayerVisibilityHint};
+use crate::{task_mgr::TaskKind, tenant::harness::TenantShardHarness};
 
 /// Used in tests to advance a future to wanted await point, and not futher.
 const ADVANCE: std::time::Duration = std::time::Duration::from_secs(3600);
@@ -27,58 +24,27 @@ const FOREVER: std::time::Duration = std::time::Duration::from_secs(ADVANCE.as_s
 async fn smoke_test() {
     let handle = tokio::runtime::Handle::current();
 
-    let h = TenantHarness::create("smoke_test").await.unwrap();
+    let h = TenantShardHarness::create("smoke_test").await.unwrap();
     let span = h.span();
     let download_span = span.in_scope(|| tracing::info_span!("downloading", timeline_id = 1));
     let (tenant, _) = h.load().await;
 
     let ctx = RequestContext::new(TaskKind::UnitTest, DownloadBehavior::Download);
 
-    let image_layers = vec![(
-        Lsn(0x40),
-        vec![(
-            Key::from_hex("620000000033333333444444445500000000").unwrap(),
-            test_img("foo"),
-        )],
-    )];
-
-    // Create a test timeline with one real layer, and one synthetic test layer.  The synthetic
-    // one is only there so that we can GC the real one without leaving the timeline's metadata
-    // empty, which is an illegal state (see [`IndexPart::validate`]).
     let timeline = tenant
-        .create_test_timeline_with_layers(
-            TimelineId::generate(),
-            Lsn(0x10),
-            14,
-            &ctx,
-            Default::default(),
-            image_layers,
-            Lsn(0x100),
-        )
+        .create_test_timeline(TimelineId::generate(), Lsn(0x10), 14, &ctx)
         .await
         .unwrap();
 
-    // Grab one of the timeline's layers to exercise in the test, and the other layer that is just
-    // there to avoid the timeline being illegally empty
-    let (layer, dummy_layer) = {
+    let layer = {
         let mut layers = {
             let layers = timeline.layers.read().await;
             layers.likely_resident_layers().cloned().collect::<Vec<_>>()
         };
 
-        assert_eq!(layers.len(), 2);
+        assert_eq!(layers.len(), 1);
 
-        layers.sort_by_key(|l| l.layer_desc().get_key_range().start);
-        let synthetic_layer = layers.pop().unwrap();
-        let real_layer = layers.pop().unwrap();
-        tracing::info!(
-            "real_layer={:?} ({}), synthetic_layer={:?} ({})",
-            real_layer,
-            real_layer.layer_desc().file_size,
-            synthetic_layer,
-            synthetic_layer.layer_desc().file_size
-        );
-        (real_layer, synthetic_layer)
+        layers.swap_remove(0)
     };
 
     // all layers created at pageserver are like `layer`, initialized with strong
@@ -207,13 +173,10 @@ async fn smoke_test() {
 
     let rtc = &timeline.remote_client;
 
-    // Simulate GC removing our test layer.
     {
-        let mut g = timeline.layers.write().await;
-
         let layers = &[layer];
+        let mut g = timeline.layers.write().await;
         g.open_mut().unwrap().finish_gc_timeline(layers);
-
         // this just updates the remote_physical_size for demonstration purposes
         rtc.schedule_gc_update(layers).unwrap();
     }
@@ -228,10 +191,7 @@ async fn smoke_test() {
 
     rtc.wait_completion().await.unwrap();
 
-    assert_eq!(
-        rtc.get_remote_physical_size(),
-        dummy_layer.metadata().file_size
-    );
+    assert_eq!(rtc.get_remote_physical_size(), 0);
     assert_eq!(0, LAYER_IMPL_METRICS.inits_cancelled.get())
 }
 
@@ -242,7 +202,7 @@ async fn evict_and_wait_on_wanted_deleted() {
     // this is the runtime on which Layer spawns the blocking tasks on
     let handle = tokio::runtime::Handle::current();
 
-    let h = TenantHarness::create("evict_and_wait_on_wanted_deleted")
+    let h = TenantShardHarness::create("evict_and_wait_on_wanted_deleted")
         .await
         .unwrap();
     utils::logging::replace_panic_hook_with_tracing_panic_hook().forget();
@@ -326,7 +286,7 @@ fn read_wins_pending_eviction() {
     rt.block_on(async move {
         // this is the runtime on which Layer spawns the blocking tasks on
         let handle = tokio::runtime::Handle::current();
-        let h = TenantHarness::create("read_wins_pending_eviction")
+        let h = TenantShardHarness::create("read_wins_pending_eviction")
             .await
             .unwrap();
         let (tenant, ctx) = h.load().await;
@@ -460,7 +420,7 @@ fn multiple_pending_evictions_scenario(name: &'static str, in_order: bool) {
     rt.block_on(async move {
         // this is the runtime on which Layer spawns the blocking tasks on
         let handle = tokio::runtime::Handle::current();
-        let h = TenantHarness::create(name).await.unwrap();
+        let h = TenantShardHarness::create(name).await.unwrap();
         let (tenant, ctx) = h.load().await;
         let span = h.span();
         let download_span = span.in_scope(|| tracing::info_span!("downloading", timeline_id = 1));
@@ -629,7 +589,7 @@ fn multiple_pending_evictions_scenario(name: &'static str, in_order: bool) {
 #[tokio::test(start_paused = true)]
 async fn cancelled_get_or_maybe_download_does_not_cancel_eviction() {
     let handle = tokio::runtime::Handle::current();
-    let h = TenantHarness::create("cancelled_get_or_maybe_download_does_not_cancel_eviction")
+    let h = TenantShardHarness::create("cancelled_get_or_maybe_download_does_not_cancel_eviction")
         .await
         .unwrap();
     let (tenant, ctx) = h.load().await;
@@ -707,7 +667,7 @@ async fn cancelled_get_or_maybe_download_does_not_cancel_eviction() {
 #[tokio::test(start_paused = true)]
 async fn evict_and_wait_does_not_wait_for_download() {
     // let handle = tokio::runtime::Handle::current();
-    let h = TenantHarness::create("evict_and_wait_does_not_wait_for_download")
+    let h = TenantShardHarness::create("evict_and_wait_does_not_wait_for_download")
         .await
         .unwrap();
     let (tenant, ctx) = h.load().await;
@@ -806,7 +766,7 @@ async fn eviction_cancellation_on_drop() {
     // this is the runtime on which Layer spawns the blocking tasks on
     let handle = tokio::runtime::Handle::current();
 
-    let h = TenantHarness::create("eviction_cancellation_on_drop")
+    let h = TenantShardHarness::create("eviction_cancellation_on_drop")
         .await
         .unwrap();
     utils::logging::replace_panic_hook_with_tracing_panic_hook().forget();

pageserver/src/tenant/storage_layer/merge_iterator.rs

@@ -357,7 +357,7 @@ mod tests {
 
     use crate::{
         tenant::{
-            harness::{TenantHarness, TIMELINE_ID},
+            harness::{TenantShardHarness, TIMELINE_ID},
             storage_layer::delta_layer::test::{produce_delta_layer, sort_delta},
         },
         DEFAULT_PG_VERSION,
@@ -393,7 +393,7 @@ mod tests {
         use bytes::Bytes;
         use pageserver_api::value::Value;
 
-        let harness = TenantHarness::create("merge_iterator_merge_in_between")
+        let harness = TenantShardHarness::create("merge_iterator_merge_in_between")
             .await
             .unwrap();
         let (tenant, ctx) = harness.load().await;
@@ -458,7 +458,7 @@ mod tests {
         use bytes::Bytes;
         use pageserver_api::value::Value;
 
-        let harness = TenantHarness::create("merge_iterator_delta_merge")
+        let harness = TenantShardHarness::create("merge_iterator_delta_merge")
             .await
             .unwrap();
         let (tenant, ctx) = harness.load().await;
@@ -535,7 +535,7 @@ mod tests {
         use bytes::Bytes;
         use pageserver_api::value::Value;
 
-        let harness = TenantHarness::create("merge_iterator_delta_image_mixed_merge")
+        let harness = TenantShardHarness::create("merge_iterator_delta_image_mixed_merge")
             .await
             .unwrap();
         let (tenant, ctx) = harness.load().await;

pageserver/src/tenant/tasks.rs

@@ -12,7 +12,7 @@ use crate::task_mgr;
 use crate::task_mgr::{TaskKind, BACKGROUND_RUNTIME};
 use crate::tenant::throttle::Stats;
 use crate::tenant::timeline::CompactionError;
-use crate::tenant::{Tenant, TenantState};
+use crate::tenant::{TenantShard, TenantState};
 use rand::Rng;
 use tokio_util::sync::CancellationToken;
 use tracing::*;
@@ -81,7 +81,7 @@ pub(crate) async fn concurrent_background_tasks_rate_limit_permit(
 
 /// Start per tenant background loops: compaction and gc.
 pub fn start_background_loops(
-    tenant: &Arc<Tenant>,
+    tenant: &Arc<TenantShard>,
     background_jobs_can_start: Option<&completion::Barrier>,
 ) {
     let tenant_shard_id = tenant.tenant_shard_id;
@@ -158,7 +158,7 @@ pub fn start_background_loops(
 ///
 /// Compaction task's main loop
 ///
-async fn compaction_loop(tenant: Arc<Tenant>, cancel: CancellationToken) {
+async fn compaction_loop(tenant: Arc<TenantShard>, cancel: CancellationToken) {
     const MAX_BACKOFF_SECS: f64 = 300.0;
     // How many errors we have seen consequtively
     let mut error_run_count = 0;
@@ -318,7 +318,7 @@ fn log_compaction_error(
 ///
 /// GC task's main loop
 ///
-async fn gc_loop(tenant: Arc<Tenant>, cancel: CancellationToken) {
+async fn gc_loop(tenant: Arc<TenantShard>, cancel: CancellationToken) {
     const MAX_BACKOFF_SECS: f64 = 300.0;
     // How many errors we have seen consequtively
     let mut error_run_count = 0;
@@ -418,7 +418,7 @@ async fn gc_loop(tenant: Arc<Tenant>, cancel: CancellationToken) {
     TENANT_TASK_EVENTS.with_label_values(&["stop"]).inc();
 }
 
-async fn ingest_housekeeping_loop(tenant: Arc<Tenant>, cancel: CancellationToken) {
+async fn ingest_housekeeping_loop(tenant: Arc<TenantShard>, cancel: CancellationToken) {
     TENANT_TASK_EVENTS.with_label_values(&["start"]).inc();
     async {
     let mut last_throttle_flag_reset_at = Instant::now();
@@ -496,7 +496,7 @@ async fn ingest_housekeeping_loop(tenant: Arc<Tenant>, cancel: CancellationToken
     TENANT_TASK_EVENTS.with_label_values(&["stop"]).inc();
 }
 
-async fn wait_for_active_tenant(tenant: &Arc<Tenant>) -> ControlFlow<()> {
+async fn wait_for_active_tenant(tenant: &Arc<TenantShard>) -> ControlFlow<()> {
     // if the tenant has a proper status already, no need to wait for anything
     if tenant.current_state() == TenantState::Active {
         ControlFlow::Continue(())

pageserver/src/tenant/timeline.rs

@@ -51,9 +51,7 @@ use tokio::{
 use tokio_util::sync::CancellationToken;
 use tracing::*;
 use utils::{
-    fs_ext,
-    guard_arc_swap::GuardArcSwap,
-    pausable_failpoint,
+    fs_ext, pausable_failpoint,
     postgres_client::PostgresClientProtocol,
     sync::gate::{Gate, GateGuard},
 };
@@ -76,7 +74,6 @@ use std::{pin::pin, sync::OnceLock};
 
 use crate::{
     aux_file::AuxFileSizeEstimator,
-    page_service::TenantManagerTypes,
     tenant::{
         config::AttachmentMode,
         layer_map::{LayerMap, SearchResult},
@@ -356,8 +353,8 @@ pub struct Timeline {
     // though let's keep them both for better error visibility.
     pub initdb_lsn: Lsn,
 
-    /// The repartitioning result. Allows a single writer and multiple readers.
-    pub(crate) partitioning: GuardArcSwap<((KeyPartitioning, SparseKeyPartitioning), Lsn)>,
+    /// When did we last calculate the partitioning? Make it pub to test cases.
+    pub(super) partitioning: tokio::sync::Mutex<((KeyPartitioning, SparseKeyPartitioning), Lsn)>,
 
     /// Configuration: how often should the partitioning be recalculated.
     repartition_threshold: u64,
@@ -416,7 +413,7 @@ pub struct Timeline {
     /// Timeline deletion will acquire both compaction and gc locks in whatever order.
     gc_lock: tokio::sync::Mutex<()>,
 
-    /// Cloned from [`super::Tenant::pagestream_throttle`] on construction.
+    /// Cloned from [`super::TenantShard::pagestream_throttle`] on construction.
     pub(crate) pagestream_throttle: Arc<crate::tenant::throttle::Throttle>,
 
     /// Size estimator for aux file v2
@@ -432,7 +429,7 @@ pub struct Timeline {
 
     pub(crate) l0_flush_global_state: L0FlushGlobalState,
 
-    pub(crate) handles: handle::PerTimelineState<TenantManagerTypes>,
+    pub(crate) handles: handle::PerTimelineState<crate::page_service::TenantManagerTypes>,
 
     pub(crate) attach_wal_lag_cooldown: Arc<OnceLock<WalLagCooldown>>,
 
@@ -1703,7 +1700,7 @@ impl Timeline {
 
     pub(crate) fn activate(
         self: &Arc<Self>,
-        parent: Arc<crate::tenant::Tenant>,
+        parent: Arc<crate::tenant::TenantShard>,
         broker_client: BrokerClientChannel,
         background_jobs_can_start: Option<&completion::Barrier>,
         ctx: &RequestContext,
@@ -2343,8 +2340,7 @@ impl Timeline {
                     // initial logical size is 0.
                     LogicalSize::empty_initial()
                 },
-
-                partitioning: GuardArcSwap::new((
+                partitioning: tokio::sync::Mutex::new((
                     (KeyPartitioning::new(), KeyPartitioning::new().into_sparse()),
                     Lsn(0),
                 )),
@@ -2679,7 +2675,7 @@ impl Timeline {
         //     (1) and (4)
         // TODO: this is basically a no-op now, should we remove it?
         self.remote_client.schedule_barrier()?;
-        // Tenant::create_timeline will wait for these uploads to happen before returning, or
+        // TenantShard::create_timeline will wait for these uploads to happen before returning, or
         // on retry.
 
         // Now that we have the full layer map, we may calculate the visibility of layers within it (a global scan)
@@ -4032,15 +4028,18 @@ impl Timeline {
         flags: EnumSet<CompactFlags>,
         ctx: &RequestContext,
     ) -> Result<((KeyPartitioning, SparseKeyPartitioning), Lsn), CompactionError> {
-        let Ok(mut guard) = self.partitioning.try_write_guard() else {
+        let Ok(mut partitioning_guard) = self.partitioning.try_lock() else {
             // NB: there are two callers, one is the compaction task, of which there is only one per struct Tenant and hence Timeline.
             // The other is the initdb optimization in flush_frozen_layer, used by `boostrap_timeline`, which runs before `.activate()`
             // and hence before the compaction task starts.
+            // Note that there are a third "caller" that will take the `partitioning` lock. It is `gc_compaction_split_jobs` for
+            // gc-compaction where it uses the repartition data to determine the split jobs. In the future, it might use its own
+            // heuristics, but for now, we should allow concurrent access to it and let the caller retry compaction.
             return Err(CompactionError::Other(anyhow!(
-                "repartition() called concurrently"
+                "repartition() called concurrently, this is rare and a retry should be fine"
             )));
         };
-        let ((dense_partition, sparse_partition), partition_lsn) = &*guard.read();
+        let ((dense_partition, sparse_partition), partition_lsn) = &*partitioning_guard;
         if lsn < *partition_lsn {
             return Err(CompactionError::Other(anyhow!(
                 "repartition() called with LSN going backwards, this should not happen"
@@ -4068,9 +4067,9 @@ impl Timeline {
         let sparse_partitioning = SparseKeyPartitioning {
             parts: vec![sparse_ks],
         }; // no partitioning for metadata keys for now
-        let result = ((dense_partitioning, sparse_partitioning), lsn);
-        guard.write(result.clone());
-        Ok(result)
+        *partitioning_guard = ((dense_partitioning, sparse_partitioning), lsn);
+
+        Ok((partitioning_guard.0.clone(), partitioning_guard.1))
     }
 
     // Is it time to create a new image layer for the given partition?
@@ -4576,7 +4575,7 @@ impl Timeline {
     /// from our ancestor to be branches of this timeline.
     pub(crate) async fn prepare_to_detach_from_ancestor(
         self: &Arc<Timeline>,
-        tenant: &crate::tenant::Tenant,
+        tenant: &crate::tenant::TenantShard,
         options: detach_ancestor::Options,
         ctx: &RequestContext,
     ) -> Result<detach_ancestor::Progress, detach_ancestor::Error> {
@@ -4594,7 +4593,7 @@ impl Timeline {
     /// resetting the tenant.
     pub(crate) async fn detach_from_ancestor_and_reparent(
         self: &Arc<Timeline>,
-        tenant: &crate::tenant::Tenant,
+        tenant: &crate::tenant::TenantShard,
         prepared: detach_ancestor::PreparedTimelineDetach,
         ctx: &RequestContext,
     ) -> Result<detach_ancestor::DetachingAndReparenting, detach_ancestor::Error> {
@@ -4606,7 +4605,7 @@ impl Timeline {
     /// The tenant must've been reset if ancestry was modified previously (in tenant manager).
     pub(crate) async fn complete_detaching_timeline_ancestor(
         self: &Arc<Timeline>,
-        tenant: &crate::tenant::Tenant,
+        tenant: &crate::tenant::TenantShard,
         attempt: detach_ancestor::Attempt,
         ctx: &RequestContext,
     ) -> Result<(), detach_ancestor::Error> {
@@ -4626,10 +4625,6 @@ impl Drop for Timeline {
                 }
             }
         }
-        info!(
-            "Timeline {} for tenant {} is being dropped",
-            self.timeline_id, self.tenant_shard_id.tenant_id
-        );
     }
 }
 
@@ -5614,14 +5609,14 @@ impl Timeline {
     /// Persistently blocks gc for `Manual` reason.
     ///
     /// Returns true if no such block existed before, false otherwise.
-    pub(crate) async fn block_gc(&self, tenant: &super::Tenant) -> anyhow::Result<bool> {
+    pub(crate) async fn block_gc(&self, tenant: &super::TenantShard) -> anyhow::Result<bool> {
         use crate::tenant::remote_timeline_client::index::GcBlockingReason;
         assert_eq!(self.tenant_shard_id, tenant.tenant_shard_id);
         tenant.gc_block.insert(self, GcBlockingReason::Manual).await
     }
 
     /// Persistently unblocks gc for `Manual` reason.
-    pub(crate) async fn unblock_gc(&self, tenant: &super::Tenant) -> anyhow::Result<()> {
+    pub(crate) async fn unblock_gc(&self, tenant: &super::TenantShard) -> anyhow::Result<()> {
         use crate::tenant::remote_timeline_client::index::GcBlockingReason;
         assert_eq!(self.tenant_shard_id, tenant.tenant_shard_id);
         tenant.gc_block.remove(self, GcBlockingReason::Manual).await
@@ -5639,8 +5634,8 @@ impl Timeline {
 
     /// Force create an image layer and place it into the layer map.
     ///
-    /// DO NOT use this function directly. Use [`Tenant::branch_timeline_test_with_layers`]
-    /// or [`Tenant::create_test_timeline_with_layers`] to ensure all these layers are
+    /// DO NOT use this function directly. Use [`TenantShard::branch_timeline_test_with_layers`]
+    /// or [`TenantShard::create_test_timeline_with_layers`] to ensure all these layers are
     /// placed into the layer map in one run AND be validated.
     #[cfg(test)]
     pub(super) async fn force_create_image_layer(
@@ -5678,24 +5673,16 @@ impl Timeline {
         info!("force created image layer {}", image_layer.local_path());
         {
             let mut guard = self.layers.write().await;
-            guard
-                .open_mut()
-                .unwrap()
-                .force_insert_layer(image_layer.clone());
+            guard.open_mut().unwrap().force_insert_layer(image_layer);
         }
 
-        // Update remote_timeline_client state to reflect existence of this layer
-        self.remote_client
-            .schedule_layer_file_upload(image_layer)
-            .unwrap();
-
         Ok(())
     }
 
     /// Force create a delta layer and place it into the layer map.
     ///
-    /// DO NOT use this function directly. Use [`Tenant::branch_timeline_test_with_layers`]
-    /// or [`Tenant::create_test_timeline_with_layers`] to ensure all these layers are
+    /// DO NOT use this function directly. Use [`TenantShard::branch_timeline_test_with_layers`]
+    /// or [`TenantShard::create_test_timeline_with_layers`] to ensure all these layers are
     /// placed into the layer map in one run AND be validated.
     #[cfg(test)]
     pub(super) async fn force_create_delta_layer(
@@ -5739,17 +5726,9 @@ impl Timeline {
         info!("force created delta layer {}", delta_layer.local_path());
         {
             let mut guard = self.layers.write().await;
-            guard
-                .open_mut()
-                .unwrap()
-                .force_insert_layer(delta_layer.clone());
+            guard.open_mut().unwrap().force_insert_layer(delta_layer);
         }
 
-        // Update remote_timeline_client state to reflect existence of this layer
-        self.remote_client
-            .schedule_layer_file_upload(delta_layer)
-            .unwrap();
-
         Ok(())
     }
 
@@ -6085,7 +6064,7 @@ mod tests {
     use utils::{id::TimelineId, lsn::Lsn};
 
     use crate::tenant::{
-        harness::{test_img, TenantHarness},
+        harness::{test_img, TenantShardHarness},
         layer_map::LayerMap,
         storage_layer::{Layer, LayerName},
         timeline::{DeltaLayerTestDesc, EvictionError},
@@ -6094,7 +6073,9 @@ mod tests {
 
     #[tokio::test]
     async fn test_heatmap_generation() {
-        let harness = TenantHarness::create("heatmap_generation").await.unwrap();
+        let harness = TenantShardHarness::create("heatmap_generation")
+            .await
+            .unwrap();
 
         let covered_delta = DeltaLayerTestDesc::new_with_inferred_key_range(
             Lsn(0x10)..Lsn(0x20),
@@ -6184,7 +6165,7 @@ mod tests {
 
     #[tokio::test]
     async fn two_layer_eviction_attempts_at_the_same_time() {
-        let harness = TenantHarness::create("two_layer_eviction_attempts_at_the_same_time")
+        let harness = TenantShardHarness::create("two_layer_eviction_attempts_at_the_same_time")
             .await
             .unwrap();
 

pageserver/src/tenant/timeline/compaction.rs

@@ -2146,7 +2146,12 @@ impl Timeline {
         let mut compact_jobs = Vec::new();
         // For now, we simply use the key partitioning information; we should do a more fine-grained partitioning
         // by estimating the amount of files read for a compaction job. We should also partition on LSN.
-        let ((dense_ks, sparse_ks), _) = self.partitioning.read().as_ref().clone();
+        let ((dense_ks, sparse_ks), _) = {
+            let Ok(partition) = self.partitioning.try_lock() else {
+                bail!("failed to acquire partition lock during gc-compaction");
+            };
+            partition.clone()
+        };
         // Truncate the key range to be within user specified compaction range.
         fn truncate_to(
             source_start: &Key,

pageserver/src/tenant/timeline/delete.rs

@@ -16,8 +16,8 @@ use crate::{
     tenant::{
         metadata::TimelineMetadata,
         remote_timeline_client::{PersistIndexPartWithDeletedFlagError, RemoteTimelineClient},
-        CreateTimelineCause, DeleteTimelineError, MaybeDeletedIndexPart, Tenant,
-        TenantManifestError, TimelineOrOffloaded,
+        CreateTimelineCause, DeleteTimelineError, MaybeDeletedIndexPart, TenantManifestError,
+        TenantShard, TimelineOrOffloaded,
     },
     virtual_file::MaybeFatalIo,
 };
@@ -112,9 +112,9 @@ pub(super) async fn delete_local_timeline_directory(
 }
 
 /// It is important that this gets called when DeletionGuard is being held.
-/// For more context see comments in [`make_timeline_delete_guard`]
+/// For more context see comments in [`DeleteTimelineFlow::prepare`]
 async fn remove_maybe_offloaded_timeline_from_tenant(
-    tenant: &Tenant,
+    tenant: &TenantShard,
     timeline: &TimelineOrOffloaded,
     _: &DeletionGuard, // using it as a witness
 ) -> anyhow::Result<()> {
@@ -188,13 +188,15 @@ impl DeleteTimelineFlow {
     // error out if some of the shutdown tasks have already been completed!
     #[instrument(skip_all)]
     pub async fn run(
-        tenant: &Arc<Tenant>,
+        tenant: &Arc<TenantShard>,
         timeline_id: TimelineId,
     ) -> Result<(), DeleteTimelineError> {
         super::debug_assert_current_span_has_tenant_and_timeline_id();
 
+        let allow_offloaded_children = false;
+        let set_stopping = true;
         let (timeline, mut guard) =
-            make_timeline_delete_guard(tenant, timeline_id, TimelineDeleteGuardKind::Delete)?;
+            Self::prepare(tenant, timeline_id, allow_offloaded_children, set_stopping)?;
 
         guard.mark_in_progress()?;
 
@@ -284,7 +286,7 @@ impl DeleteTimelineFlow {
     /// Shortcut to create Timeline in stopping state and spawn deletion task.
     #[instrument(skip_all, fields(%timeline_id))]
     pub(crate) async fn resume_deletion(
-        tenant: Arc<Tenant>,
+        tenant: Arc<TenantShard>,
         timeline_id: TimelineId,
         local_metadata: &TimelineMetadata,
         remote_client: RemoteTimelineClient,
@@ -331,10 +333,79 @@ impl DeleteTimelineFlow {
         Ok(())
     }
 
+    pub(super) fn prepare(
+        tenant: &TenantShard,
+        timeline_id: TimelineId,
+        allow_offloaded_children: bool,
+        set_stopping: bool,
+    ) -> Result<(TimelineOrOffloaded, DeletionGuard), DeleteTimelineError> {
+        // Note the interaction between this guard and deletion guard.
+        // Here we attempt to lock deletion guard when we're holding a lock on timelines.
+        // This is important because when you take into account `remove_timeline_from_tenant`
+        // we remove timeline from memory when we still hold the deletion guard.
+        // So here when timeline deletion is finished timeline wont be present in timelines map at all
+        // which makes the following sequence impossible:
+        // T1: get preempted right before the try_lock on `Timeline::delete_progress`
+        // T2: do a full deletion, acquire and drop `Timeline::delete_progress`
+        // T1: acquire deletion lock, do another `DeleteTimelineFlow::run`
+        // For more context see this discussion: `https://github.com/neondatabase/neon/pull/4552#discussion_r1253437346`
+        let timelines = tenant.timelines.lock().unwrap();
+        let timelines_offloaded = tenant.timelines_offloaded.lock().unwrap();
+
+        let timeline = match timelines.get(&timeline_id) {
+            Some(t) => TimelineOrOffloaded::Timeline(Arc::clone(t)),
+            None => match timelines_offloaded.get(&timeline_id) {
+                Some(t) => TimelineOrOffloaded::Offloaded(Arc::clone(t)),
+                None => return Err(DeleteTimelineError::NotFound),
+            },
+        };
+
+        // Ensure that there are no child timelines, because we are about to remove files,
+        // which will break child branches
+        let mut children = Vec::new();
+        if !allow_offloaded_children {
+            children.extend(timelines_offloaded.iter().filter_map(|(id, entry)| {
+                (entry.ancestor_timeline_id == Some(timeline_id)).then_some(*id)
+            }));
+        }
+        children.extend(timelines.iter().filter_map(|(id, entry)| {
+            (entry.get_ancestor_timeline_id() == Some(timeline_id)).then_some(*id)
+        }));
+
+        if !children.is_empty() {
+            return Err(DeleteTimelineError::HasChildren(children));
+        }
+
+        // Note that using try_lock here is important to avoid a deadlock.
+        // Here we take lock on timelines and then the deletion guard.
+        // At the end of the operation we're holding the guard and need to lock timelines map
+        // to remove the timeline from it.
+        // Always if you have two locks that are taken in different order this can result in a deadlock.
+
+        let delete_progress = Arc::clone(timeline.delete_progress());
+        let delete_lock_guard = match delete_progress.try_lock_owned() {
+            Ok(guard) => DeletionGuard(guard),
+            Err(_) => {
+                // Unfortunately if lock fails arc is consumed.
+                return Err(DeleteTimelineError::AlreadyInProgress(Arc::clone(
+                    timeline.delete_progress(),
+                )));
+            }
+        };
+
+        if set_stopping {
+            if let TimelineOrOffloaded::Timeline(timeline) = &timeline {
+                timeline.set_state(TimelineState::Stopping);
+            }
+        }
+
+        Ok((timeline, delete_lock_guard))
+    }
+
     fn schedule_background(
         guard: DeletionGuard,
         conf: &'static PageServerConf,
-        tenant: Arc<Tenant>,
+        tenant: Arc<TenantShard>,
         timeline: TimelineOrOffloaded,
         remote_client: Arc<RemoteTimelineClient>,
     ) {
@@ -368,7 +439,7 @@ impl DeleteTimelineFlow {
     async fn background(
         mut guard: DeletionGuard,
         conf: &PageServerConf,
-        tenant: &Tenant,
+        tenant: &TenantShard,
         timeline: &TimelineOrOffloaded,
         remote_client: Arc<RemoteTimelineClient>,
     ) -> Result<(), DeleteTimelineError> {
@@ -412,80 +483,6 @@ impl DeleteTimelineFlow {
     }
 }
 
-#[derive(Copy, Clone, PartialEq, Eq)]
-pub(super) enum TimelineDeleteGuardKind {
-    Offload,
-    Delete,
-}
-
-pub(super) fn make_timeline_delete_guard(
-    tenant: &Tenant,
-    timeline_id: TimelineId,
-    guard_kind: TimelineDeleteGuardKind,
-) -> Result<(TimelineOrOffloaded, DeletionGuard), DeleteTimelineError> {
-    // Note the interaction between this guard and deletion guard.
-    // Here we attempt to lock deletion guard when we're holding a lock on timelines.
-    // This is important because when you take into account `remove_timeline_from_tenant`
-    // we remove timeline from memory when we still hold the deletion guard.
-    // So here when timeline deletion is finished timeline wont be present in timelines map at all
-    // which makes the following sequence impossible:
-    // T1: get preempted right before the try_lock on `Timeline::delete_progress`
-    // T2: do a full deletion, acquire and drop `Timeline::delete_progress`
-    // T1: acquire deletion lock, do another `DeleteTimelineFlow::run`
-    // For more context see this discussion: `https://github.com/neondatabase/neon/pull/4552#discussion_r1253437346`
-    let timelines = tenant.timelines.lock().unwrap();
-    let timelines_offloaded = tenant.timelines_offloaded.lock().unwrap();
-
-    let timeline = match timelines.get(&timeline_id) {
-        Some(t) => TimelineOrOffloaded::Timeline(Arc::clone(t)),
-        None => match timelines_offloaded.get(&timeline_id) {
-            Some(t) => TimelineOrOffloaded::Offloaded(Arc::clone(t)),
-            None => return Err(DeleteTimelineError::NotFound),
-        },
-    };
-
-    // Ensure that there are no child timelines, because we are about to remove files,
-    // which will break child branches
-    let mut children = Vec::new();
-    if guard_kind == TimelineDeleteGuardKind::Delete {
-        children.extend(timelines_offloaded.iter().filter_map(|(id, entry)| {
-            (entry.ancestor_timeline_id == Some(timeline_id)).then_some(*id)
-        }));
-    }
-    children.extend(timelines.iter().filter_map(|(id, entry)| {
-        (entry.get_ancestor_timeline_id() == Some(timeline_id)).then_some(*id)
-    }));
-
-    if !children.is_empty() {
-        return Err(DeleteTimelineError::HasChildren(children));
-    }
-
-    // Note that using try_lock here is important to avoid a deadlock.
-    // Here we take lock on timelines and then the deletion guard.
-    // At the end of the operation we're holding the guard and need to lock timelines map
-    // to remove the timeline from it.
-    // Always if you have two locks that are taken in different order this can result in a deadlock.
-
-    let delete_progress = Arc::clone(timeline.delete_progress());
-    let delete_lock_guard = match delete_progress.try_lock_owned() {
-        Ok(guard) => DeletionGuard(guard),
-        Err(_) => {
-            // Unfortunately if lock fails arc is consumed.
-            return Err(DeleteTimelineError::AlreadyInProgress(Arc::clone(
-                timeline.delete_progress(),
-            )));
-        }
-    };
-
-    if guard_kind == TimelineDeleteGuardKind::Delete {
-        if let TimelineOrOffloaded::Timeline(timeline) = &timeline {
-            timeline.set_state(TimelineState::Stopping);
-        }
-    }
-
-    Ok((timeline, delete_lock_guard))
-}
-
 pub(super) struct DeletionGuard(OwnedMutexGuard<DeleteTimelineFlow>);
 
 impl Deref for DeletionGuard {

pageserver/src/tenant/timeline/detach_ancestor.rs

@@ -7,7 +7,7 @@ use crate::{
     tenant::{
         remote_timeline_client::index::GcBlockingReason::DetachAncestor,
         storage_layer::{AsLayerDesc as _, DeltaLayerWriter, Layer, ResidentLayer},
-        Tenant,
+        TenantShard,
     },
     virtual_file::{MaybeFatalIo, VirtualFile},
 };
@@ -159,7 +159,7 @@ impl Attempt {
 /// See [`Timeline::prepare_to_detach_from_ancestor`]
 pub(super) async fn prepare(
     detached: &Arc<Timeline>,
-    tenant: &Tenant,
+    tenant: &TenantShard,
     options: Options,
     ctx: &RequestContext,
 ) -> Result<Progress, Error> {
@@ -410,7 +410,7 @@ pub(super) async fn prepare(
     Ok(Progress::Prepared(attempt, prepared))
 }
 
-async fn start_new_attempt(detached: &Timeline, tenant: &Tenant) -> Result<Attempt, Error> {
+async fn start_new_attempt(detached: &Timeline, tenant: &TenantShard) -> Result<Attempt, Error> {
     let attempt = obtain_exclusive_attempt(detached, tenant)?;
 
     // insert the block in the index_part.json, if not already there.
@@ -426,13 +426,16 @@ async fn start_new_attempt(detached: &Timeline, tenant: &Tenant) -> Result<Attem
     Ok(attempt)
 }
 
-async fn continue_with_blocked_gc(detached: &Timeline, tenant: &Tenant) -> Result<Attempt, Error> {
+async fn continue_with_blocked_gc(
+    detached: &Timeline,
+    tenant: &TenantShard,
+) -> Result<Attempt, Error> {
     // FIXME: it would be nice to confirm that there is an in-memory version, since we've just
     // verified there is a persistent one?
     obtain_exclusive_attempt(detached, tenant)
 }
 
-fn obtain_exclusive_attempt(detached: &Timeline, tenant: &Tenant) -> Result<Attempt, Error> {
+fn obtain_exclusive_attempt(detached: &Timeline, tenant: &TenantShard) -> Result<Attempt, Error> {
     use Error::{OtherTimelineDetachOngoing, ShuttingDown};
 
     // ensure we are the only active attempt for this tenant
@@ -460,7 +463,7 @@ fn obtain_exclusive_attempt(detached: &Timeline, tenant: &Tenant) -> Result<Atte
 
 fn reparented_direct_children(
     detached: &Arc<Timeline>,
-    tenant: &Tenant,
+    tenant: &TenantShard,
 ) -> Result<HashSet<TimelineId>, Error> {
     let mut all_direct_children = tenant
         .timelines
@@ -698,7 +701,7 @@ impl DetachingAndReparenting {
 /// See [`Timeline::detach_from_ancestor_and_reparent`].
 pub(super) async fn detach_and_reparent(
     detached: &Arc<Timeline>,
-    tenant: &Tenant,
+    tenant: &TenantShard,
     prepared: PreparedTimelineDetach,
     _ctx: &RequestContext,
 ) -> Result<DetachingAndReparenting, Error> {
@@ -901,7 +904,7 @@ pub(super) async fn detach_and_reparent(
 
 pub(super) async fn complete(
     detached: &Arc<Timeline>,
-    tenant: &Tenant,
+    tenant: &TenantShard,
     mut attempt: Attempt,
     _ctx: &RequestContext,
 ) -> Result<(), Error> {
@@ -970,7 +973,7 @@ where
 }
 
 fn check_no_archived_children_of_ancestor(
-    tenant: &Tenant,
+    tenant: &TenantShard,
     detached: &Arc<Timeline>,
     ancestor: &Arc<Timeline>,
     ancestor_lsn: Lsn,

pageserver/src/tenant/timeline/eviction_task.rs

@@ -31,7 +31,8 @@ use crate::{
     task_mgr::{self, TaskKind, BACKGROUND_RUNTIME},
     tenant::{
         size::CalculateSyntheticSizeError, storage_layer::LayerVisibilityHint,
-        tasks::BackgroundLoopKind, timeline::EvictionError, LogicalSizeCalculationCause, Tenant,
+        tasks::BackgroundLoopKind, timeline::EvictionError, LogicalSizeCalculationCause,
+        TenantShard,
     },
 };
 
@@ -52,7 +53,7 @@ pub struct EvictionTaskTenantState {
 impl Timeline {
     pub(super) fn launch_eviction_task(
         self: &Arc<Self>,
-        parent: Arc<Tenant>,
+        parent: Arc<TenantShard>,
         background_tasks_can_start: Option<&completion::Barrier>,
     ) {
         let self_clone = Arc::clone(self);
@@ -79,7 +80,7 @@ impl Timeline {
     }
 
     #[instrument(skip_all, fields(tenant_id = %self.tenant_shard_id.tenant_id, shard_id = %self.tenant_shard_id.shard_slug(), timeline_id = %self.timeline_id))]
-    async fn eviction_task(self: Arc<Self>, tenant: Arc<Tenant>) {
+    async fn eviction_task(self: Arc<Self>, tenant: Arc<TenantShard>) {
         use crate::tenant::tasks::random_init_delay;
 
         // acquire the gate guard only once within a useful span
@@ -123,7 +124,7 @@ impl Timeline {
     #[instrument(skip_all, fields(policy_kind = policy.discriminant_str()))]
     async fn eviction_iteration(
         self: &Arc<Self>,
-        tenant: &Tenant,
+        tenant: &TenantShard,
         policy: &EvictionPolicy,
         cancel: &CancellationToken,
         gate: &GateGuard,
@@ -180,7 +181,7 @@ impl Timeline {
 
     async fn eviction_iteration_threshold(
         self: &Arc<Self>,
-        tenant: &Tenant,
+        tenant: &TenantShard,
         p: &EvictionPolicyLayerAccessThreshold,
         cancel: &CancellationToken,
         gate: &GateGuard,
@@ -314,7 +315,7 @@ impl Timeline {
     /// disk usage based eviction task.
     async fn imitiate_only(
         self: &Arc<Self>,
-        tenant: &Tenant,
+        tenant: &TenantShard,
         p: &EvictionPolicyLayerAccessThreshold,
         cancel: &CancellationToken,
         gate: &GateGuard,
@@ -370,7 +371,7 @@ impl Timeline {
     #[instrument(skip_all)]
     async fn imitate_layer_accesses(
         &self,
-        tenant: &Tenant,
+        tenant: &TenantShard,
         p: &EvictionPolicyLayerAccessThreshold,
         cancel: &CancellationToken,
         gate: &GateGuard,
@@ -506,7 +507,7 @@ impl Timeline {
     #[instrument(skip_all)]
     async fn imitate_synthetic_size_calculation_worker(
         &self,
-        tenant: &Tenant,
+        tenant: &TenantShard,
         cancel: &CancellationToken,
         ctx: &RequestContext,
     ) {

pageserver/src/tenant/timeline/handle.rs

@@ -32,151 +32,54 @@
 //!
 //! # Design
 //!
-//! ## Data Structures
-//!
 //! There are three user-facing data structures:
 //! - `PerTimelineState`: a struct embedded into each Timeline struct. Lifetime == Timeline lifetime.
 //! - `Cache`: a struct private to each connection handler; Lifetime == connection lifetime.
 //! - `Handle`: a smart pointer that holds the Timeline gate open and derefs to `&Timeline`.
-//! - `WeakHandle`: downgrade of a `Handle` that does not keep the gate open, but allows
-//!   trying to ugprade back to a `Handle`, guaranteeing it's the same `Timeline` *object*.
+//!   Lifetime: for a single request dispatch on the Timeline (i.e., one getpage request)
 //!
-//! Internally, there is 0 or 1 `HandleInner` per `(Cache,Timeline)`.
-//! Since Cache:Connection is 1:1, there is 0 or 1 `HandleInner` per `(Connection,Timeline)`.
+//! The `Handle` is just a wrapper around an `Arc<HandleInner>`.
 //!
-//! The `HandleInner`  is allocated as a `Arc<Mutex<HandleInner>>` and
-//! referenced weakly and strongly from various places which we are now illustrating.
-//! For brevity, we will omit the `Arc<Mutex<>>` part in the following and instead
-//! use `strong ref` and `weak ref` when referring to the `Arc<Mutex<HandleInner>>`
-//! or `Weak<Mutex<HandleInner>>`, respectively.
-//!
-//! - The `Handle` is a strong ref.
-//! - The `WeakHandle` is a weak ref.
-//! - The `PerTimelineState` contains a `HashMap<CacheId, strong ref>`.
-//! - The `Cache` is a `HashMap<unique identifier for the shard, weak ref>`.
-//!
-//! Lifetimes:
-//! - `WeakHandle` and `Handle`: single pagestream request.
-//! - `Cache`: single page service connection.
-//! - `PerTimelineState`:  lifetime of the Timeline object (i.e., i.e., till `Timeline::shutdown`).
-//!
-//! ## Request Handling Flow (= filling and using the `Cache``)
+//! There is one long-lived `Arc<HandleInner>`, which is stored in the `PerTimelineState`.
+//! The `Cache` stores a `Weak<HandleInner>` for each cached Timeline.
 //!
 //! To dispatch a request, the page service connection calls `Cache::get`.
 //!
 //! A cache miss means we consult the tenant manager for shard routing,
-//! resulting in an `Arc<Timeline>`. We enter its gate _once_ and store it in the the
-//! `Arc<Mutex<HandleInner>>>`. A weak ref is stored in the `Cache`
-//! and a strong ref in the `PerTimelineState`.
-//! A strong ref is returned wrapped in a `Handle`.
+//! resulting in an `Arc<Timeline>`. We enter its gate _once_ and construct an
+//! `Arc<HandleInner>`. We store a `Weak<HandleInner>` in the cache
+//! and the `Arc<HandleInner>` in the `PerTimelineState`.
 //!
 //! For subsequent requests, `Cache::get` will perform a "fast path" shard routing
-//! and find the weak ref in the cache.
-//! We upgrade the weak ref to a strong ref and return it wrapped in a `Handle`.
+//! and find the `Weak<HandleInner>` in the cache.
+//! We upgrade the `Weak<HandleInner>` to an `Arc<HandleInner>` and wrap it in the user-facing `Handle` type.
 //!
-//! The pagestream processing is pipelined and involves a batching step.
-//! While a request is batching, the `Handle` is downgraded to a `WeakHandle`.
-//! When the batch is ready to be executed, the `WeakHandle` is upgraded back to a `Handle`
-//! and the request handler dispatches the request to the right `<Handle as Deref<Target = Timeline>>::$request_method`.
+//! The request handler dispatches the request to the right `<Handle as Deref<Target = Timeline>>::$request_method`.
 //! It then drops the `Handle`, which drops the `Arc<HandleInner>`.
 //!
-//! # Performance
+//! # Memory Management / How The Reference Cycle Is Broken
 //!
-//! Remember from the introductory section:
+//! The attentive reader may have noticed the strong reference cycle
+//! from `Arc<HandleInner>` to `PerTimelineState` to `Arc<Timeline>`.
 //!
-//! > However, we want to avoid the overhead of entering the gate for every
-//! > method invocation.
-//!
-//! Why do we want to avoid that?
-//! Because the gate is a shared location in memory and entering it involves
-//! bumping refcounts, which leads to cache contention if done frequently
-//! from multiple cores in parallel.
-//!
-//! So, we only acquire the `GateGuard` once on `Cache` miss, and wrap it in an `Arc`.
-//! That `Arc` is private to the `HandleInner` and hence to the connection.
-//! (Review the "Data Structures" section if that is unclear to you.)
-//!
-//! A `WeakHandle` is a weak ref to the `HandleInner`.
-//! When upgrading a `WeakHandle`, we upgrade to a strong ref to the `HandleInner` and
-//! further acquire an additional strong ref to the `Arc<GateGuard>` inside it.
-//! Again, this manipulation of ref counts is is cheap because `Arc` is private to the connection.
-//!
-//! When downgrading a `Handle` to a `WeakHandle`, we drop the `Arc<GateGuard>`.
-//! Again, this is cheap because the `Arc` is private to the connection.
-//!
-//! In addition to the GateGuard, we need to provide `Deref<Target=Timeline>` impl.
-//! For this, both `Handle` need infallible access to an `Arc<Timeline>`.
-//! We could clone the `Arc<Timeline>` when upgrading a `WeakHandle`, but that would cause contention
-//! on the shared memory location that trakcs the refcount of the `Arc<Timeline>`.
-//! Instead, we wrap the `Arc<Timeline>` into another `Arc`.
-//! so that we can clone it cheaply when upgrading a `WeakHandle`.
-//!
-//! # Shutdown
-//!
-//! The attentive reader may have noticed the following reference cycle around the `Arc<Timeline>`:
-//!
-//! ```text
-//! Timeline --owns--> PerTimelineState --strong--> HandleInner --strong--> Timeline
-//! ```
-//!
-//! Further, there is this cycle:
-//!
-//! ```text
-//! Timeline --owns--> PerTimelineState --strong--> HandleInner --strong--> GateGuard --keepalive--> Timeline
-//! ```
-//!
-//! The former cycle is a memory leak if not broken.
-//! The latter cycle further prevents the Timeline from shutting down
-//! because we certainly won't drop the Timeline while the GateGuard is alive.
-//! Preventing shutdown is the whole point of this handle/cache system,
-//! but when the Timeline needs to shut down, we need to break the cycle.
+//! This cycle is intentional: while it exists, the `Cache` can upgrade its
+//! `Weak<HandleInner>` to an `Arc<HandleInner>` in a single atomic operation.
 //!
 //! The cycle is broken by either
-//! - Timeline shutdown (=> `PerTimelineState::shutdown`)
-//! - Connection shutdown (=> dropping the `Cache`).
+//! - `PerTimelineState::shutdown` or
+//! - dropping the `Cache`.
 //!
-//! Both transition the `HandleInner` from [`HandleInner::KeepingTimelineGateOpen`] to
-//! [`HandleInner::ShutDown`], which drops the only long-lived strong ref to the
-//! `Arc<GateGuard>`.
-//!
-//! `PerTimelineState::shutdown` drops all the `HandleInners` it contains,
-//! thereby breaking the cycle.
-//! It also initiates draining of already existing `Handle`s by
-//! poisoning things so that no new `HandleInner`'s can be added
-//! to the `PerTimelineState`, which will make subsequent `Cache::get` fail.
-//!
-//! Concurrently existing / already upgraded `Handle`s will extend the
-//! lifetime of the `Arc<Mutex<HandleInner>>` and hence cycles.
+//! Concurrently existing `Handle`s will extend the existence of the cycle.
 //! However, since `Handle`s are short-lived and new `Handle`s are not
-//! handed out from `Cache::get` or `WeakHandle::upgrade` after
-//! `PerTimelineState::shutdown`, that extension of the cycle is bounded.
-//!
-//! Concurrently existing `WeakHandle`s will fail to `upgrade()`:
-//! while they will succeed in upgrading `Weak<Mutex<HandleInner>>`,
-//! they will find the inner in state `HandleInner::ShutDown` state where the
-//! `Arc<GateGuard>` and Timeline has already been dropped.
-//!
-//! Dropping the `Cache` undoes the registration of this `Cache`'s
-//! `HandleInner`s from all the `PerTimelineState`s, i.e., it
-//! removes the strong ref to each of its `HandleInner`s
-//! from all the `PerTimelineState`.
-//!
-//! # Locking Rules
-//!
-//! To prevent deadlocks we:
-//!
-//! 1. Only ever hold one of the locks at a time.
-//! 2. Don't add more than one Drop impl that locks on the
-//!    cycles above.
-//!
-//! As per (2), that impl is in `Drop for Cache`.
+//! handed out after either `PerTimelineState::shutdown` or `Cache` drop,
+//! that extension of the cycle is bounded.
 //!
 //! # Fast Path for Shard Routing
 //!
 //! The `Cache` has a fast path for shard routing to avoid calling into
 //! the tenant manager for every request.
 //!
-//! The `Cache` maintains a hash map of `ShardTimelineId` to `WeakHandle`s.
+//! The `Cache` maintains a hash map of `ShardTimelineId` to `Weak<HandleInner>`.
 //!
 //! The current implementation uses the first entry in the hash map
 //! to determine the `ShardParameters` and derive the correct
@@ -184,18 +87,18 @@
 //!
 //! It then looks up the hash map for that `ShardTimelineId := {ShardIndex,TimelineId}`.
 //!
-//! If the lookup is successful and the `WeakHandle` can be upgraded,
+//! If the lookup is successful and the `Weak<HandleInner>` can be upgraded,
 //! it's a hit.
 //!
 //! ## Cache invalidation
 //!
-//! The insight is that cache invalidation is sufficient and most efficiently if done lazily.
+//! The insight is that cache invalidation is sufficient and most efficiently done lazily.
 //! The only reasons why an entry in the cache can become stale are:
 //! 1. The `PerTimelineState` / Timeline is shutting down e.g. because the shard is
 //!    being detached, timeline or shard deleted, or pageserver is shutting down.
 //! 2. We're doing a shard split and new traffic should be routed to the child shards.
 //!
-//! Regarding (1), we will eventually fail to upgrade the `WeakHandle` once the
+//! Regarding (1), we will eventually fail to upgrade the `Weak<HandleInner>` once the
 //! timeline has shut down, and when that happens, we remove the entry from the cache.
 //!
 //! Regarding (2), the insight is that it is toally fine to keep dispatching requests
@@ -204,6 +107,8 @@
 
 use std::collections::hash_map;
 use std::collections::HashMap;
+use std::sync::atomic::AtomicBool;
+use std::sync::atomic::Ordering;
 use std::sync::Arc;
 use std::sync::Mutex;
 use std::sync::Weak;
@@ -247,7 +152,7 @@ pub(crate) struct Cache<T: Types> {
     map: Map<T>,
 }
 
-type Map<T> = HashMap<ShardTimelineId, WeakHandle<T>>;
+type Map<T> = HashMap<ShardTimelineId, Weak<HandleInner<T>>>;
 
 impl<T: Types> Default for Cache<T> {
     fn default() -> Self {
@@ -265,22 +170,12 @@ pub(crate) struct ShardTimelineId {
 }
 
 /// See module-level comment.
-pub(crate) struct Handle<T: Types> {
-    timeline: Arc<T::Timeline>,
-    #[allow(dead_code)] // the field exists to keep the gate open
-    gate_guard: Arc<utils::sync::gate::GateGuard>,
-    inner: Arc<Mutex<HandleInner<T>>>,
-}
-pub(crate) struct WeakHandle<T: Types> {
-    inner: Weak<Mutex<HandleInner<T>>>,
-}
-enum HandleInner<T: Types> {
-    KeepingTimelineGateOpen {
-        #[allow(dead_code)]
-        gate_guard: Arc<utils::sync::gate::GateGuard>,
-        timeline: Arc<T::Timeline>,
-    },
-    ShutDown,
+pub(crate) struct Handle<T: Types>(Arc<HandleInner<T>>);
+struct HandleInner<T: Types> {
+    shut_down: AtomicBool,
+    timeline: T::Timeline,
+    // The timeline's gate held open.
+    _gate_guard: utils::sync::gate::GateGuard,
 }
 
 /// Embedded in each [`Types::Timeline`] as the anchor for the only long-lived strong ref to `HandleInner`.
@@ -288,8 +183,7 @@ enum HandleInner<T: Types> {
 /// See module-level comment for details.
 pub struct PerTimelineState<T: Types> {
     // None = shutting down
-    #[allow(clippy::type_complexity)]
-    handles: Mutex<Option<HashMap<CacheId, Arc<Mutex<HandleInner<T>>>>>>,
+    handles: Mutex<Option<HashMap<CacheId, Arc<HandleInner<T>>>>>,
 }
 
 impl<T: Types> Default for PerTimelineState<T> {
@@ -349,24 +243,49 @@ impl<T: Types> Cache<T> {
         shard_selector: ShardSelector,
         tenant_manager: &T::TenantManager,
     ) -> Result<Handle<T>, GetError<T>> {
-        // terminates because when every iteration we remove an element from the map
-        let miss: ShardSelector = loop {
+        // terminates because each iteration removes an element from the map
+        loop {
+            let handle = self
+                .get_impl(timeline_id, shard_selector, tenant_manager)
+                .await?;
+            if handle.0.shut_down.load(Ordering::Relaxed) {
+                let removed = self
+                    .map
+                    .remove(&handle.0.timeline.shard_timeline_id())
+                    .expect("invariant of get_impl is that the returned handle is in the map");
+                assert!(
+                    Weak::ptr_eq(&removed, &Arc::downgrade(&handle.0)),
+                    "shard_timeline_id() incorrect?"
+                );
+            } else {
+                return Ok(handle);
+            }
+        }
+    }
+
+    #[instrument(level = "trace", skip_all)]
+    async fn get_impl(
+        &mut self,
+        timeline_id: TimelineId,
+        shard_selector: ShardSelector,
+        tenant_manager: &T::TenantManager,
+    ) -> Result<Handle<T>, GetError<T>> {
+        let miss: ShardSelector = {
             let routing_state = self.shard_routing(timeline_id, shard_selector);
             match routing_state {
                 RoutingResult::FastPath(handle) => return Ok(handle),
                 RoutingResult::SlowPath(key) => match self.map.get(&key) {
                     Some(cached) => match cached.upgrade() {
-                        Ok(upgraded) => return Ok(upgraded),
-                        Err(HandleUpgradeError::ShutDown) => {
-                            // TODO: dedup with shard_routing()
+                        Some(upgraded) => return Ok(Handle(upgraded)),
+                        None => {
                             trace!("handle cache stale");
                             self.map.remove(&key).unwrap();
-                            continue;
+                            ShardSelector::Known(key.shard_index)
                         }
                     },
-                    None => break ShardSelector::Known(key.shard_index),
+                    None => ShardSelector::Known(key.shard_index),
                 },
-                RoutingResult::NeedConsultTenantManager => break shard_selector,
+                RoutingResult::NeedConsultTenantManager => shard_selector,
             }
         };
         self.get_miss(timeline_id, miss, tenant_manager).await
@@ -383,7 +302,7 @@ impl<T: Types> Cache<T> {
             let Some((first_key, first_handle)) = self.map.iter().next() else {
                 return RoutingResult::NeedConsultTenantManager;
             };
-            let Ok(first_handle) = first_handle.upgrade() else {
+            let Some(first_handle) = first_handle.upgrade() else {
                 // TODO: dedup with get()
                 trace!("handle cache stale");
                 let first_key_owned = *first_key;
@@ -391,7 +310,7 @@ impl<T: Types> Cache<T> {
                 continue;
             };
 
-            let first_handle_shard_identity = first_handle.get_shard_identity();
+            let first_handle_shard_identity = first_handle.timeline.get_shard_identity();
             let make_shard_index = |shard_num: ShardNumber| ShardIndex {
                 shard_number: shard_num,
                 shard_count: first_handle_shard_identity.count,
@@ -410,11 +329,11 @@ impl<T: Types> Cache<T> {
             };
             let first_handle_shard_timeline_id = ShardTimelineId {
                 shard_index: first_handle_shard_identity.shard_index(),
-                timeline_id: first_handle.shard_timeline_id().timeline_id,
+                timeline_id: first_handle.timeline.shard_timeline_id().timeline_id,
             };
 
             if need_shard_timeline_id == first_handle_shard_timeline_id {
-                return RoutingResult::FastPath(first_handle);
+                return RoutingResult::FastPath(Handle(first_handle));
             } else {
                 return RoutingResult::SlowPath(need_shard_timeline_id);
             }
@@ -438,30 +357,23 @@ impl<T: Types> Cache<T> {
                     ShardSelector::Known(idx) => assert_eq!(idx, &key.shard_index),
                 }
 
-                trace!("creating new HandleInner");
-                let handle_inner_arc = Arc::new(Mutex::new(HandleInner::KeepingTimelineGateOpen {
-                    gate_guard: Arc::new(
-                        // this enter() is expensive in production code because
-                        // it hits the global Arc<Timeline>::gate refcounts
-                        match timeline.gate().enter() {
-                            Ok(guard) => guard,
-                            Err(_) => {
-                                return Err(GetError::TimelineGateClosed);
-                            }
-                        },
-                    ),
-                    // this clone is expensive in production code because
-                    // it hits the global Arc<Timeline>::clone refcounts
-                    timeline: Arc::new(timeline.clone()),
-                }));
-                let handle_weak = WeakHandle {
-                    inner: Arc::downgrade(&handle_inner_arc),
+                let gate_guard = match timeline.gate().enter() {
+                    Ok(guard) => guard,
+                    Err(_) => {
+                        return Err(GetError::TimelineGateClosed);
+                    }
                 };
-                let handle = handle_weak
-                    .upgrade()
-                    .ok()
-                    .expect("we just created it and it's not linked anywhere yet");
-                {
+                trace!("creating new HandleInner");
+                let handle = Arc::new(
+                    // TODO: global metric that keeps track of the number of live HandlerTimeline instances
+                    // so we can identify reference cycle bugs.
+                    HandleInner {
+                        shut_down: AtomicBool::new(false),
+                        _gate_guard: gate_guard,
+                        timeline: timeline.clone(),
+                    },
+                );
+                let handle = {
                     let mut lock_guard = timeline
                         .per_timeline_state()
                         .handles
@@ -469,8 +381,7 @@ impl<T: Types> Cache<T> {
                         .expect("mutex poisoned");
                     match &mut *lock_guard {
                         Some(per_timeline_state) => {
-                            let replaced =
-                                per_timeline_state.insert(self.id, Arc::clone(&handle_inner_arc));
+                            let replaced = per_timeline_state.insert(self.id, Arc::clone(&handle));
                             assert!(replaced.is_none(), "some earlier code left a stale handle");
                             match self.map.entry(key) {
                                 hash_map::Entry::Occupied(_o) => {
@@ -481,7 +392,8 @@ impl<T: Types> Cache<T> {
                                     unreachable!()
                                 }
                                 hash_map::Entry::Vacant(v) => {
-                                    v.insert(handle_weak);
+                                    v.insert(Arc::downgrade(&handle));
+                                    handle
                                 }
                             }
                         }
@@ -489,62 +401,14 @@ impl<T: Types> Cache<T> {
                             return Err(GetError::PerTimelineStateShutDown);
                         }
                     }
-                }
-                Ok(handle)
+                };
+                Ok(Handle(handle))
             }
             Err(e) => Err(GetError::TenantManager(e)),
         }
     }
 }
 
-pub(crate) enum HandleUpgradeError {
-    ShutDown,
-}
-
-impl<T: Types> WeakHandle<T> {
-    pub(crate) fn upgrade(&self) -> Result<Handle<T>, HandleUpgradeError> {
-        let Some(inner) = Weak::upgrade(&self.inner) else {
-            return Err(HandleUpgradeError::ShutDown);
-        };
-        let lock_guard = inner.lock().expect("poisoned");
-        match &*lock_guard {
-            HandleInner::KeepingTimelineGateOpen {
-                timeline,
-                gate_guard,
-            } => {
-                let gate_guard = Arc::clone(gate_guard);
-                let timeline = Arc::clone(timeline);
-                drop(lock_guard);
-                Ok(Handle {
-                    timeline,
-                    gate_guard,
-                    inner,
-                })
-            }
-            HandleInner::ShutDown => Err(HandleUpgradeError::ShutDown),
-        }
-    }
-
-    pub(crate) fn is_same_handle_as(&self, other: &WeakHandle<T>) -> bool {
-        Weak::ptr_eq(&self.inner, &other.inner)
-    }
-}
-
-impl<T: Types> std::ops::Deref for Handle<T> {
-    type Target = T::Timeline;
-    fn deref(&self) -> &Self::Target {
-        &self.timeline
-    }
-}
-
-impl<T: Types> Handle<T> {
-    pub(crate) fn downgrade(&self) -> WeakHandle<T> {
-        WeakHandle {
-            inner: Arc::downgrade(&self.inner),
-        }
-    }
-}
-
 impl<T: Types> PerTimelineState<T> {
     /// After this method returns, [`Cache::get`] will never again return a [`Handle`]
     /// to the [`Types::Timeline`] that embeds this per-timeline state.
@@ -566,62 +430,43 @@ impl<T: Types> PerTimelineState<T> {
             trace!("already shut down");
             return;
         };
-        for handle_inner_arc in handles.values() {
+        for handle in handles.values() {
             // Make hits fail.
-            let mut lock_guard = handle_inner_arc.lock().expect("poisoned");
-            lock_guard.shutdown();
+            handle.shut_down.store(true, Ordering::Relaxed);
         }
         drop(handles);
     }
 }
 
+impl<T: Types> std::ops::Deref for Handle<T> {
+    type Target = T::Timeline;
+    fn deref(&self) -> &Self::Target {
+        &self.0.timeline
+    }
+}
+
+#[cfg(test)]
+impl<T: Types> Drop for HandleInner<T> {
+    fn drop(&mut self) {
+        trace!("HandleInner dropped");
+    }
+}
+
 // When dropping a [`Cache`], prune its handles in the [`PerTimelineState`] to break the reference cycle.
 impl<T: Types> Drop for Cache<T> {
     fn drop(&mut self) {
-        for (
-            _,
-            WeakHandle {
-                inner: handle_inner_weak,
-            },
-        ) in self.map.drain()
-        {
-            let Some(handle_inner_arc) = handle_inner_weak.upgrade() else {
-                continue;
-            };
-            let Some(handle_timeline) = handle_inner_arc
-                // locking rules: drop lock before acquiring other lock below
-                .lock()
-                .expect("poisoned")
-                .shutdown()
-            else {
-                // Concurrent PerTimelineState::shutdown.
-                continue;
-            };
-            // Clean up per_timeline_state so the HandleInner allocation can be dropped.
-            let per_timeline_state = handle_timeline.per_timeline_state();
-            let mut handles_lock_guard = per_timeline_state.handles.lock().expect("mutex poisoned");
-            let Some(handles) = &mut *handles_lock_guard else {
-                continue;
-            };
-            let Some(removed_handle_inner_arc) = handles.remove(&self.id) else {
-                // Concurrent PerTimelineState::shutdown.
-                continue;
-            };
-            drop(handles_lock_guard); // locking rules!
-            assert!(Arc::ptr_eq(&removed_handle_inner_arc, &handle_inner_arc));
-        }
-    }
-}
-
-impl<T: Types> HandleInner<T> {
-    fn shutdown(&mut self) -> Option<Arc<T::Timeline>> {
-        match std::mem::replace(self, HandleInner::ShutDown) {
-            HandleInner::KeepingTimelineGateOpen { timeline, .. } => Some(timeline),
-            HandleInner::ShutDown => {
-                // Duplicate shutdowns are possible because both Cache::drop and PerTimelineState::shutdown
-                // may do it concurrently, but locking rules disallow holding per-timeline-state lock and
-                // the handle lock at the same time.
-                None
+        for (_, weak) in self.map.drain() {
+            if let Some(strong) = weak.upgrade() {
+                // handle is still being kept alive in PerTimelineState
+                let timeline = strong.timeline.per_timeline_state();
+                let mut handles = timeline.handles.lock().expect("mutex poisoned");
+                if let Some(handles) = &mut *handles {
+                    let Some(removed) = handles.remove(&self.id) else {
+                        // There could have been a shutdown inbetween us upgrading the weak and locking the mutex.
+                        continue;
+                    };
+                    assert!(Arc::ptr_eq(&removed, &strong));
+                }
             }
         }
     }
@@ -629,8 +474,6 @@ impl<T: Types> HandleInner<T> {
 
 #[cfg(test)]
 mod tests {
-    use std::sync::Weak;
-
     use pageserver_api::{
         key::{rel_block_to_key, Key, DBDIR_KEY},
         models::ShardParameters,
@@ -740,13 +583,39 @@ mod tests {
         //
         // fill the cache
         //
+        assert_eq!(
+            (Arc::strong_count(&shard0), Arc::weak_count(&shard0)),
+            (2, 1),
+            "strong: shard0, mgr; weak: myself"
+        );
+
         let handle: Handle<_> = cache
             .get(timeline_id, ShardSelector::Page(key), &mgr)
             .await
             .expect("we have the timeline");
+        let handle_inner_weak = Arc::downgrade(&handle.0);
         assert!(Weak::ptr_eq(&handle.myself, &shard0.myself));
+        assert_eq!(
+            (
+                Weak::strong_count(&handle_inner_weak),
+                Weak::weak_count(&handle_inner_weak)
+            ),
+            (2, 2),
+            "strong: handle, per_timeline_state, weak: handle_inner_weak, cache"
+        );
         assert_eq!(cache.map.len(), 1);
+
+        assert_eq!(
+            (Arc::strong_count(&shard0), Arc::weak_count(&shard0)),
+            (3, 1),
+            "strong: handleinner(per_timeline_state), shard0, mgr; weak: myself"
+        );
         drop(handle);
+        assert_eq!(
+            (Arc::strong_count(&shard0), Arc::weak_count(&shard0)),
+            (3, 1),
+            "strong: handleinner(per_timeline_state), shard0, mgr; weak: myself"
+        );
 
         //
         // demonstrate that Handle holds up gate closure
@@ -771,11 +640,21 @@ mod tests {
         // SHUTDOWN
         shard0.per_timeline_state.shutdown(); // keeping handle alive across shutdown
 
+        assert_eq!(
+            1,
+            Weak::strong_count(&handle_inner_weak),
+            "through local var handle"
+        );
         assert_eq!(
             cache.map.len(),
             1,
             "this is an implementation detail but worth pointing out: we can't clear the cache from shutdown(), it's cleared on first access after"
         );
+        assert_eq!(
+            (Arc::strong_count(&shard0), Arc::weak_count(&shard0)),
+            (3, 1),
+            "strong: handleinner(via handle), shard0, mgr; weak: myself"
+        );
 
         // this handle is perfectly usable
         handle.getpage();
@@ -799,6 +678,16 @@ mod tests {
         }
 
         drop(handle);
+        assert_eq!(
+            0,
+            Weak::strong_count(&handle_inner_weak),
+            "the HandleInner destructor already ran"
+        );
+        assert_eq!(
+            (Arc::strong_count(&shard0), Arc::weak_count(&shard0)),
+            (2, 1),
+            "strong: shard0, mgr; weak: myself"
+        );
 
         // closing gate succeeds after dropping handle
         tokio::select! {
@@ -817,8 +706,10 @@ mod tests {
         assert_eq!(cache.map.len(), 0);
 
         // ensure all refs to shard0 are gone and we're not leaking anything
+        let myself = Weak::clone(&shard0.myself);
         drop(shard0);
         drop(mgr);
+        assert_eq!(Weak::strong_count(&myself), 0);
     }
 
     #[tokio::test]
@@ -1057,11 +948,15 @@ mod tests {
                 handle
             };
             handle.getpage();
-            used_handles.push(Arc::downgrade(&handle.timeline));
+            used_handles.push(Arc::downgrade(&handle.0));
         }
 
-        // No handles exist, thus gates are closed and don't require shutdown.
-        // Thus the gate should close immediately, even without shutdown.
+        // No handles exist, thus gates are closed and don't require shutdown
+        assert!(used_handles
+            .iter()
+            .all(|weak| Weak::strong_count(weak) == 0));
+
+        // ... thus the gate should close immediately, even without shutdown
         tokio::select! {
             _ = shard0.gate.close() => { }
             _ = tokio::time::sleep(FOREVER) => {
@@ -1069,172 +964,4 @@ mod tests {
             }
         }
     }
-
-    #[tokio::test(start_paused = true)]
-    async fn test_weak_handles() {
-        crate::tenant::harness::setup_logging();
-        let timeline_id = TimelineId::generate();
-        let shard0 = Arc::new_cyclic(|myself| StubTimeline {
-            gate: Default::default(),
-            id: timeline_id,
-            shard: ShardIdentity::unsharded(),
-            per_timeline_state: PerTimelineState::default(),
-            myself: myself.clone(),
-        });
-        let mgr = StubManager {
-            shards: vec![shard0.clone()],
-        };
-
-        let refcount_start = Arc::strong_count(&shard0);
-
-        let key = DBDIR_KEY;
-
-        let mut cache = Cache::<TestTypes>::default();
-
-        let handle = cache
-            .get(timeline_id, ShardSelector::Page(key), &mgr)
-            .await
-            .expect("we have the timeline");
-        assert!(Weak::ptr_eq(&handle.myself, &shard0.myself));
-
-        let weak_handle = handle.downgrade();
-
-        drop(handle);
-
-        let upgraded_handle = weak_handle.upgrade().ok().expect("we can upgrade it");
-
-        // Start shutdown
-        shard0.per_timeline_state.shutdown();
-
-        // Upgrades during shutdown don't work, even if upgraded_handle exists.
-        weak_handle
-            .upgrade()
-            .err()
-            .expect("can't upgrade weak handle as soon as shutdown started");
-
-        // But upgraded_handle is still alive, so the gate won't close.
-        tokio::select! {
-            _ = shard0.gate.close() => {
-                panic!("handle is keeping gate open");
-            }
-            _ = tokio::time::sleep(FOREVER) => { }
-        }
-
-        // Drop the last handle.
-        drop(upgraded_handle);
-
-        // The gate should close now, despite there still being a weak_handle.
-        tokio::select! {
-            _ = shard0.gate.close() => { }
-            _ = tokio::time::sleep(FOREVER) => {
-                panic!("only strong handle is dropped and we shut down per-timeline-state")
-            }
-        }
-
-        // The weak handle still can't be upgraded.
-        weak_handle
-            .upgrade()
-            .err()
-            .expect("still shouldn't be able to upgrade the weak handle");
-
-        // There should be no strong references to the timeline object except the one on "stack".
-        assert_eq!(Arc::strong_count(&shard0), refcount_start);
-    }
-
-    #[tokio::test(start_paused = true)]
-    async fn test_reference_cycle_broken_when_cache_is_dropped() {
-        crate::tenant::harness::setup_logging();
-        let timeline_id = TimelineId::generate();
-        let shard0 = Arc::new_cyclic(|myself| StubTimeline {
-            gate: Default::default(),
-            id: timeline_id,
-            shard: ShardIdentity::unsharded(),
-            per_timeline_state: PerTimelineState::default(),
-            myself: myself.clone(),
-        });
-        let mgr = StubManager {
-            shards: vec![shard0.clone()],
-        };
-        let key = DBDIR_KEY;
-
-        let mut cache = Cache::<TestTypes>::default();
-
-        // helper to check if a handle is referenced by per_timeline_state
-        let per_timeline_state_refs_handle = |handle_weak: &Weak<Mutex<HandleInner<_>>>| {
-            let per_timeline_state = shard0.per_timeline_state.handles.lock().unwrap();
-            let per_timeline_state = per_timeline_state.as_ref().unwrap();
-            per_timeline_state
-                .values()
-                .any(|v| Weak::ptr_eq(&Arc::downgrade(v), handle_weak))
-        };
-
-        // Fill the cache.
-        let handle = cache
-            .get(timeline_id, ShardSelector::Page(key), &mgr)
-            .await
-            .expect("we have the timeline");
-        assert!(Weak::ptr_eq(&handle.myself, &shard0.myself));
-        let handle_inner_weak = Arc::downgrade(&handle.inner);
-        assert!(
-            per_timeline_state_refs_handle(&handle_inner_weak),
-            "we still hold `handle` _and_ haven't dropped `cache` yet"
-        );
-
-        // Drop the cache.
-        drop(cache);
-
-        assert!(
-            !(per_timeline_state_refs_handle(&handle_inner_weak)),
-            "nothing should reference the handle allocation anymore"
-        );
-        assert!(
-            Weak::upgrade(&handle_inner_weak).is_some(),
-            "the local `handle` still keeps the allocation alive"
-        );
-        // but obviously the cache is gone so no new allocations can be handed out.
-
-        // Drop handle.
-        drop(handle);
-        assert!(
-            Weak::upgrade(&handle_inner_weak).is_none(),
-            "the local `handle` is dropped, so the allocation should be dropped by now"
-        );
-    }
-
-    #[tokio::test(start_paused = true)]
-    async fn test_reference_cycle_broken_when_per_timeline_state_shutdown() {
-        crate::tenant::harness::setup_logging();
-        let timeline_id = TimelineId::generate();
-        let shard0 = Arc::new_cyclic(|myself| StubTimeline {
-            gate: Default::default(),
-            id: timeline_id,
-            shard: ShardIdentity::unsharded(),
-            per_timeline_state: PerTimelineState::default(),
-            myself: myself.clone(),
-        });
-        let mgr = StubManager {
-            shards: vec![shard0.clone()],
-        };
-        let key = DBDIR_KEY;
-
-        let mut cache = Cache::<TestTypes>::default();
-        let handle = cache
-            .get(timeline_id, ShardSelector::Page(key), &mgr)
-            .await
-            .expect("we have the timeline");
-        // grab a weak reference to the inner so can later try to Weak::upgrade it and assert that fails
-        let handle_inner_weak = Arc::downgrade(&handle.inner);
-
-        // drop the handle, obviously the lifetime of `inner` is at least as long as each strong reference to it
-        drop(handle);
-        assert!(Weak::upgrade(&handle_inner_weak).is_some(), "can still");
-
-        // Shutdown the per_timeline_state.
-        shard0.per_timeline_state.shutdown();
-        assert!(Weak::upgrade(&handle_inner_weak).is_none(), "can no longer");
-
-        // cache only contains Weak's, so, it can outlive the per_timeline_state without
-        // Drop explicitly solely to make this point.
-        drop(cache);
-    }
 }

pageserver/src/tenant/timeline/offload.rs

@@ -2,12 +2,11 @@ use std::sync::Arc;
 
 use pageserver_api::models::{TenantState, TimelineState};
 
-use super::delete::{delete_local_timeline_directory, DeletionGuard};
+use super::delete::{delete_local_timeline_directory, DeleteTimelineFlow, DeletionGuard};
 use super::Timeline;
 use crate::span::debug_assert_current_span_has_tenant_and_timeline_id;
 use crate::tenant::remote_timeline_client::ShutdownIfArchivedError;
-use crate::tenant::timeline::delete::{make_timeline_delete_guard, TimelineDeleteGuardKind};
-use crate::tenant::{OffloadedTimeline, Tenant, TenantManifestError, TimelineOrOffloaded};
+use crate::tenant::{OffloadedTimeline, TenantManifestError, TenantShard, TimelineOrOffloaded};
 
 #[derive(thiserror::Error, Debug)]
 pub(crate) enum OffloadError {
@@ -31,16 +30,19 @@ impl From<TenantManifestError> for OffloadError {
 }
 
 pub(crate) async fn offload_timeline(
-    tenant: &Tenant,
+    tenant: &TenantShard,
     timeline: &Arc<Timeline>,
 ) -> Result<(), OffloadError> {
     debug_assert_current_span_has_tenant_and_timeline_id();
     tracing::info!("offloading archived timeline");
 
-    let (timeline, guard) = make_timeline_delete_guard(
+    let allow_offloaded_children = true;
+    let set_stopping = false;
+    let (timeline, guard) = DeleteTimelineFlow::prepare(
         tenant,
         timeline.timeline_id,
-        TimelineDeleteGuardKind::Offload,
+        allow_offloaded_children,
+        set_stopping,
     )
     .map_err(|e| OffloadError::Other(anyhow::anyhow!(e)))?;
 
@@ -104,11 +106,11 @@ pub(crate) async fn offload_timeline(
 }
 
 /// It is important that this gets called when DeletionGuard is being held.
-/// For more context see comments in [`make_timeline_delete_guard`]
+/// For more context see comments in [`DeleteTimelineFlow::prepare`]
 ///
 /// Returns the strong count of the timeline `Arc`
 fn remove_timeline_from_tenant(
-    tenant: &Tenant,
+    tenant: &TenantShard,
     timeline: &Timeline,
     _: &DeletionGuard, // using it as a witness
 ) -> usize {

pageserver/src/tenant/timeline/uninit.rs

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

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

@@ -1117,7 +1117,7 @@ impl ReconnectReason {
 #[cfg(test)]
 mod tests {
     use super::*;
-    use crate::tenant::harness::{TenantHarness, TIMELINE_ID};
+    use crate::tenant::harness::{TenantShardHarness, TIMELINE_ID};
     use pageserver_api::config::defaults::DEFAULT_WAL_RECEIVER_PROTOCOL;
     use url::Host;
 
@@ -1141,7 +1141,7 @@ mod tests {
 
     #[tokio::test]
     async fn no_connection_no_candidate() -> anyhow::Result<()> {
-        let harness = TenantHarness::create("no_connection_no_candidate").await?;
+        let harness = TenantShardHarness::create("no_connection_no_candidate").await?;
         let mut state = dummy_state(&harness).await;
         let now = Utc::now().naive_utc();
 
@@ -1174,7 +1174,7 @@ mod tests {
 
     #[tokio::test]
     async fn connection_no_candidate() -> anyhow::Result<()> {
-        let harness = TenantHarness::create("connection_no_candidate").await?;
+        let harness = TenantShardHarness::create("connection_no_candidate").await?;
         let mut state = dummy_state(&harness).await;
         let now = Utc::now().naive_utc();
 
@@ -1239,7 +1239,7 @@ mod tests {
 
     #[tokio::test]
     async fn no_connection_candidate() -> anyhow::Result<()> {
-        let harness = TenantHarness::create("no_connection_candidate").await?;
+        let harness = TenantShardHarness::create("no_connection_candidate").await?;
         let mut state = dummy_state(&harness).await;
         let now = Utc::now().naive_utc();
 
@@ -1302,7 +1302,7 @@ mod tests {
 
     #[tokio::test]
     async fn candidate_with_many_connection_failures() -> anyhow::Result<()> {
-        let harness = TenantHarness::create("candidate_with_many_connection_failures").await?;
+        let harness = TenantShardHarness::create("candidate_with_many_connection_failures").await?;
         let mut state = dummy_state(&harness).await;
         let now = Utc::now().naive_utc();
 
@@ -1342,7 +1342,7 @@ mod tests {
 
     #[tokio::test]
     async fn lsn_wal_over_threshold_current_candidate() -> anyhow::Result<()> {
-        let harness = TenantHarness::create("lsn_wal_over_threshcurrent_candidate").await?;
+        let harness = TenantShardHarness::create("lsn_wal_over_threshcurrent_candidate").await?;
         let mut state = dummy_state(&harness).await;
         let current_lsn = Lsn(100_000).align();
         let now = Utc::now().naive_utc();
@@ -1409,7 +1409,7 @@ mod tests {
     #[tokio::test]
     async fn timeout_connection_threshold_current_candidate() -> anyhow::Result<()> {
         let harness =
-            TenantHarness::create("timeout_connection_threshold_current_candidate").await?;
+            TenantShardHarness::create("timeout_connection_threshold_current_candidate").await?;
         let mut state = dummy_state(&harness).await;
         let current_lsn = Lsn(100_000).align();
         let now = Utc::now().naive_utc();
@@ -1472,7 +1472,8 @@ mod tests {
 
     #[tokio::test]
     async fn timeout_wal_over_threshold_current_candidate() -> anyhow::Result<()> {
-        let harness = TenantHarness::create("timeout_wal_over_threshold_current_candidate").await?;
+        let harness =
+            TenantShardHarness::create("timeout_wal_over_threshold_current_candidate").await?;
         let mut state = dummy_state(&harness).await;
         let current_lsn = Lsn(100_000).align();
         let new_lsn = Lsn(100_100).align();
@@ -1540,7 +1541,7 @@ mod tests {
 
     const DUMMY_SAFEKEEPER_HOST: &str = "safekeeper_connstr";
 
-    async fn dummy_state(harness: &TenantHarness) -> ConnectionManagerState {
+    async fn dummy_state(harness: &TenantShardHarness) -> ConnectionManagerState {
         let (tenant, ctx) = harness.load().await;
         let timeline = tenant
             .create_test_timeline(TIMELINE_ID, Lsn(0x8), crate::DEFAULT_PG_VERSION, &ctx)
@@ -1575,7 +1576,7 @@ mod tests {
         // and pageserver should prefer to connect to it.
         let test_az = Some("test_az".to_owned());
 
-        let harness = TenantHarness::create("switch_to_same_availability_zone").await?;
+        let harness = TenantShardHarness::create("switch_to_same_availability_zone").await?;
         let mut state = dummy_state(&harness).await;
         state.conf.availability_zone.clone_from(&test_az);
         let current_lsn = Lsn(100_000).align();

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

@@ -140,7 +140,7 @@ pub(super) async fn handle_walreceiver_connection(
 
     let (replication_client, connection) = {
         let mut config = wal_source_connconf.to_tokio_postgres_config();
-        config.application_name(format!("pageserver-{}", node.0).as_str());
+        config.application_name("pageserver");
         config.replication_mode(tokio_postgres::config::ReplicationMode::Physical);
         match time::timeout(connect_timeout, config.connect(postgres::NoTls)).await {
             Ok(client_and_conn) => client_and_conn?,

pageserver/src/tenant/upload_queue.rs

@@ -563,7 +563,7 @@ impl UploadOp {
 #[cfg(test)]
 mod tests {
     use super::*;
-    use crate::tenant::harness::{TenantHarness, TIMELINE_ID};
+    use crate::tenant::harness::{TenantShardHarness, TIMELINE_ID};
     use crate::tenant::storage_layer::layer::local_layer_path;
     use crate::tenant::storage_layer::Layer;
     use crate::tenant::Timeline;
@@ -621,7 +621,7 @@ mod tests {
 
         runtime
             .block_on(async {
-                let harness = TenantHarness::create(test_name).await?;
+                let harness = TenantShardHarness::create(test_name).await?;
                 let (tenant, ctx) = harness.load().await;
                 tenant
                     .create_test_timeline(TIMELINE_ID, Lsn(8), DEFAULT_PG_VERSION, &ctx)

pageserver/src/utilization.rs

@@ -9,12 +9,14 @@ use utils::serde_percent::Percent;
 
 use pageserver_api::models::PageserverUtilization;
 
-use crate::{config::PageServerConf, metrics::NODE_UTILIZATION_SCORE, tenant::mgr::TenantManager};
+use crate::{
+    config::PageServerConf, metrics::NODE_UTILIZATION_SCORE, tenant::mgr::TenantShardManager,
+};
 
 pub(crate) fn regenerate(
     conf: &PageServerConf,
     tenants_path: &Path,
-    tenant_manager: &TenantManager,
+    tenant_manager: &TenantShardManager,
 ) -> anyhow::Result<PageserverUtilization> {
     let statvfs = nix::sys::statvfs::statvfs(tenants_path)
         .map_err(std::io::Error::from)

pageserver/src/walingest.rs

@@ -1531,7 +1531,10 @@ mod tests {
 
     #[tokio::test]
     async fn test_relsize() -> Result<()> {
-        let (tenant, ctx) = TenantHarness::create("test_relsize").await?.load().await;
+        let (tenant, ctx) = TenantShardHarness::create("test_relsize")
+            .await?
+            .load()
+            .await;
         let tline = tenant
             .create_test_timeline(TIMELINE_ID, Lsn(8), DEFAULT_PG_VERSION, &ctx)
             .await?;
@@ -1771,7 +1774,7 @@ mod tests {
     // and then created it again within the same layer.
     #[tokio::test]
     async fn test_drop_extend() -> Result<()> {
-        let (tenant, ctx) = TenantHarness::create("test_drop_extend")
+        let (tenant, ctx) = TenantShardHarness::create("test_drop_extend")
             .await?
             .load()
             .await;
@@ -1847,7 +1850,7 @@ mod tests {
     // and then extended it again within the same layer.
     #[tokio::test]
     async fn test_truncate_extend() -> Result<()> {
-        let (tenant, ctx) = TenantHarness::create("test_truncate_extend")
+        let (tenant, ctx) = TenantShardHarness::create("test_truncate_extend")
             .await?
             .load()
             .await;
@@ -2001,7 +2004,10 @@ mod tests {
     /// split into multiple 1 GB segments in Postgres.
     #[tokio::test]
     async fn test_large_rel() -> Result<()> {
-        let (tenant, ctx) = TenantHarness::create("test_large_rel").await?.load().await;
+        let (tenant, ctx) = TenantShardHarness::create("test_large_rel")
+            .await?
+            .load()
+            .await;
         let tline = tenant
             .create_test_timeline(TIMELINE_ID, Lsn(8), DEFAULT_PG_VERSION, &ctx)
             .await?;
@@ -2109,7 +2115,9 @@ mod tests {
         let startpoint = Lsn::from_hex("14AEC08").unwrap();
         let _endpoint = Lsn::from_hex("1FFFF98").unwrap();
 
-        let harness = TenantHarness::create("test_ingest_real_wal").await.unwrap();
+        let harness = TenantShardHarness::create("test_ingest_real_wal")
+            .await
+            .unwrap();
         let span = harness
             .span()
             .in_scope(|| info_span!("timeline_span", timeline_id=%TIMELINE_ID));

pgxn/neon/file_cache.c

@@ -911,85 +911,57 @@ lfc_writev(NRelFileInfo rinfo, ForkNumber forkNum, BlockNumber blkno,
 			if (entry->access_count++ == 0)
 				dlist_delete(&entry->list_node);
 		}
-		/*-----------
-		 * If the chunk wasn't already in the LFC then we have these
-		 * options, in order of preference:
-		 *
-		 * Unless there is no space available, we can:
-		 *  1. Use an entry from the `holes` list, and
-		 *  2. Create a new entry.
-		 * We can always, regardless of space in the LFC:
-		 *  3. evict an entry from LRU, and
-		 *  4. ignore the write operation (the least favorite option)
-		 */
-		else if (lfc_ctl->used < lfc_ctl->limit)
+		else
 		{
-			if (!dlist_is_empty(&lfc_ctl->holes))
+			/*
+			 * We have two choices if all cache pages are pinned (i.e. used in IO
+			 * operations):
+			 *
+			 * 1) Wait until some of this operation is completed and pages is
+			 * unpinned.
+			 *
+			 * 2) Allocate one more chunk, so that specified cache size is more
+			 * recommendation than hard limit.
+			 *
+			 * As far as probability of such event (that all pages are pinned) is
+			 * considered to be very very small: there are should be very large
+			 * number of concurrent IO operations and them are limited by
+			 * max_connections, we prefer not to complicate code and use second
+			 * approach.
+			 */
+			if (lfc_ctl->used >= lfc_ctl->limit && !dlist_is_empty(&lfc_ctl->lru))
+			{
+				/* Cache overflow: evict least recently used chunk */
+				FileCacheEntry *victim = dlist_container(FileCacheEntry, list_node, dlist_pop_head_node(&lfc_ctl->lru));
+	
+				for (int i = 0; i < BLOCKS_PER_CHUNK; i++)
+				{
+					lfc_ctl->used_pages -= (victim->bitmap[i >> 5] >> (i & 31)) & 1;
+				}
+				CriticalAssert(victim->access_count == 0);
+				entry->offset = victim->offset; /* grab victim's chunk */
+				hash_search_with_hash_value(lfc_hash, &victim->key, victim->hash, HASH_REMOVE, NULL);
+				neon_log(DEBUG2, "Swap file cache page");
+			}
+			else if (!dlist_is_empty(&lfc_ctl->holes))
 			{
 				/* We can reuse a hole that was left behind when the LFC was shrunk previously */
-				FileCacheEntry *hole = dlist_container(FileCacheEntry, list_node,
-													   dlist_pop_head_node(&lfc_ctl->holes));
-				uint32 offset = hole->offset;
-				bool hole_found;
-
-				hash_search_with_hash_value(lfc_hash, &hole->key,
-											hole->hash, HASH_REMOVE, &hole_found);
+				FileCacheEntry *hole = dlist_container(FileCacheEntry, list_node, dlist_pop_head_node(&lfc_ctl->holes));
+				uint32		offset = hole->offset;
+				bool		hole_found;
+	
+				hash_search_with_hash_value(lfc_hash, &hole->key, hole->hash, HASH_REMOVE, &hole_found);
 				CriticalAssert(hole_found);
-
+	
 				lfc_ctl->used += 1;
-				entry->offset = offset;			/* reuse the hole */
+				entry->offset = offset;	/* reuse the hole */
 			}
 			else
 			{
 				lfc_ctl->used += 1;
-				entry->offset = lfc_ctl->size++;/* allocate new chunk at end
-												 * of file */
+				entry->offset = lfc_ctl->size++;	/* allocate new chunk at end
+													 * of file */
 			}
-		}
-		/*
-		 * We've already used up all allocated LFC entries.
-		 *
-		 * If we can clear an entry from the LRU, do that.
-		 * If we can't (e.g. because all other slots are being accessed)
-		 * then we will remove this entry from the hash and continue
-		 * on to the next chunk, as we may not exceed the limit.
-		 */
-		else if (!dlist_is_empty(&lfc_ctl->lru))
-		{
-			/* Cache overflow: evict least recently used chunk */
-			FileCacheEntry *victim = dlist_container(FileCacheEntry, list_node,
-													 dlist_pop_head_node(&lfc_ctl->lru));
-
-			for (int i = 0; i < BLOCKS_PER_CHUNK; i++)
-			{
-				lfc_ctl->used_pages -= (victim->bitmap[i >> 5] >> (i & 31)) & 1;
-			}
-
-			CriticalAssert(victim->access_count == 0);
-			entry->offset = victim->offset; /* grab victim's chunk */
-			hash_search_with_hash_value(lfc_hash, &victim->key,
-										victim->hash, HASH_REMOVE, NULL);
-			neon_log(DEBUG2, "Swap file cache page");
-		}
-		else
-		{
-			/* Can't add this chunk - we don't have the space for it */
-			hash_search_with_hash_value(lfc_hash, &entry->key, hash,
-										HASH_REMOVE, NULL);
-
-			/*
-			 * We can't process this chunk due to lack of space in LFC,
-			 * so skip to the next one
-			 */
-			LWLockRelease(lfc_lock);
-			blkno += blocks_in_chunk;
-			buf_offset += blocks_in_chunk;
-			nblocks -= blocks_in_chunk;
-			continue;
-		}
-
-		if (!found)
-		{
 			entry->access_count = 1;
 			entry->hash = hash;
 			memset(entry->bitmap, 0, sizeof entry->bitmap);

pgxn/neon/pagestore_client.h

@@ -34,8 +34,6 @@ typedef enum
 	T_NeonGetPageRequest,
 	T_NeonDbSizeRequest,
 	T_NeonGetSlruSegmentRequest,
-	/* future tags above this line */
-	T_NeonTestRequest = 99, /* only in cfg(feature = "testing") */
 
 	/* pagestore -> pagestore_client */
 	T_NeonExistsResponse = 100,
@@ -44,8 +42,6 @@ typedef enum
 	T_NeonErrorResponse,
 	T_NeonDbSizeResponse,
 	T_NeonGetSlruSegmentResponse,
-	/* future tags above this line */
-	T_NeonTestResponse = 199, /* only in cfg(feature = "testing") */
 } NeonMessageTag;
 
 typedef uint64 NeonRequestId;

safekeeper/client/src/mgmt_api.rs

@@ -102,11 +102,6 @@ impl Client {
         self.get(&uri).await
     }
 
-    pub async fn utilization(&self) -> Result<reqwest::Response> {
-        let uri = format!("{}/v1/utilization/", self.mgmt_api_endpoint);
-        self.get(&uri).await
-    }
-
     async fn get<U: IntoUrl>(&self, uri: U) -> Result<reqwest::Response> {
         self.request(Method::GET, uri, ()).await
     }

safekeeper/src/http/routes.rs

@@ -127,13 +127,6 @@ async fn timeline_create_handler(mut request: Request<Body>) -> Result<Response<
     json_response(StatusCode::OK, ())
 }
 
-async fn utilization_handler(request: Request<Body>) -> Result<Response<Body>, ApiError> {
-    check_permission(&request, None)?;
-    let global_timelines = get_global_timelines(&request);
-    let utilization = global_timelines.get_timeline_counts();
-    json_response(StatusCode::OK, utilization)
-}
-
 /// List all (not deleted) timelines.
 /// Note: it is possible to do the same with debug_dump.
 async fn timeline_list_handler(request: Request<Body>) -> Result<Response<Body>, ApiError> {
@@ -627,7 +620,6 @@ pub fn make_router(
                 failpoints_handler(r, cancel).await
             })
         })
-        .get("/v1/uzilization", |r| request_span(r, utilization_handler))
         .delete("/v1/tenant/:tenant_id", |r| {
             request_span(r, tenant_delete_handler)
         })

safekeeper/src/timelines_global_map.rs

@@ -13,7 +13,6 @@ use anyhow::{bail, Context, Result};
 use camino::Utf8PathBuf;
 use camino_tempfile::Utf8TempDir;
 use safekeeper_api::membership::Configuration;
-use safekeeper_api::models::SafekeeperUtilization;
 use safekeeper_api::ServerInfo;
 use serde::Serialize;
 use std::collections::HashMap;
@@ -417,20 +416,6 @@ impl GlobalTimelines {
             .collect()
     }
 
-    /// Returns statistics about timeline counts
-    pub fn get_timeline_counts(&self) -> SafekeeperUtilization {
-        let global_lock = self.state.lock().unwrap();
-        let timeline_count = global_lock
-            .timelines
-            .values()
-            .filter(|t| match t {
-                GlobalMapTimeline::CreationInProgress => false,
-                GlobalMapTimeline::Timeline(t) => !t.is_cancelled(),
-            })
-            .count() as u64;
-        SafekeeperUtilization { timeline_count }
-    }
-
     /// Returns all timelines belonging to a given tenant. Used for deleting all timelines of a tenant,
     /// and that's why it can return cancelled timelines, to retry deleting them.
     fn get_all_for_tenant(&self, tenant_id: TenantId) -> Vec<Arc<Timeline>> {

storage_controller/migrations/2025-01-15-181207_safekeepers_disabled_to_pause/down.sql

@@ -1,2 +0,0 @@
-ALTER TABLE safekeepers ALTER COLUMN scheduling_policy SET DEFAULT 'disabled';
-UPDATE safekeepers SET scheduling_policy = 'disabled' WHERE scheduling_policy = 'pause';

storage_controller/migrations/2025-01-15-181207_safekeepers_disabled_to_pause/up.sql

@@ -1,2 +0,0 @@
-ALTER TABLE safekeepers ALTER COLUMN scheduling_policy SET DEFAULT 'pause';
-UPDATE safekeepers SET scheduling_policy = 'pause' WHERE scheduling_policy = 'disabled';

storage_controller/src/service.rs

@@ -5411,15 +5411,6 @@ impl Service {
 
         expect_shards.sort_by_key(|tsp| (tsp.tenant_id.clone(), tsp.shard_number, tsp.shard_count));
 
-        // Because JSON contents of persistent tenants might disagree with the fields in current `TenantConfig`
-        // definition, we will do an encode/decode cycle to ensure any legacy fields are dropped and any new
-        // fields are added, before doing a comparison.
-        for tsp in &mut persistent_shards {
-            let config: TenantConfig = serde_json::from_str(&tsp.config)
-                .map_err(|e| ApiError::InternalServerError(e.into()))?;
-            tsp.config = serde_json::to_string(&config).expect("Encoding config is infallible");
-        }
-
         if persistent_shards != expect_shards {
             tracing::error!("Consistency check failed on shards.");
 
@@ -7279,14 +7270,19 @@ impl Service {
         Ok(())
     }
 
-    /// Create a node fill plan (pick secondaries to promote), based on:
-    /// 1. Shards which have a secondary on this node, and this node is in their home AZ, and are currently attached to a node
-    ///    outside their home AZ, should be migrated back here.
-    /// 2. If after step 1 we have not migrated enough shards for this node to have its fair share of
-    ///    attached shards, we will promote more shards from the nodes with the most attached shards, unless
-    ///    those shards have a home AZ that doesn't match the node we're filling.
+    /// Create a node fill plan (pick secondaries to promote) that meets the following requirements:
+    /// 1. The node should be filled until it reaches the expected cluster average of
+    ///    attached shards. If there are not enough secondaries on the node, the plan stops early.
+    /// 2. Select tenant shards to promote such that the number of attached shards is balanced
+    ///    throughout the cluster. We achieve this by picking tenant shards from each node,
+    ///    starting from the ones with the largest number of attached shards, until the node
+    ///    reaches the expected cluster average.
+    /// 3. Avoid promoting more shards of the same tenant than required. The upper bound
+    ///    for the number of tenants from the same shard promoted to the node being filled is:
+    ///    shard count for the tenant divided by the number of nodes in the cluster.
     fn fill_node_plan(&self, node_id: NodeId) -> Vec<TenantShardId> {
         let mut locked = self.inner.write().unwrap();
+        let fill_requirement = locked.scheduler.compute_fill_requirement(node_id);
         let (nodes, tenants, _scheduler) = locked.parts_mut();
 
         let node_az = nodes
@@ -7295,79 +7291,53 @@ impl Service {
             .get_availability_zone_id()
             .clone();
 
-        // The tenant shard IDs that we plan to promote from secondary to attached on this node
-        let mut plan = Vec::new();
+        let mut tids_by_node = tenants
+            .iter_mut()
+            .filter_map(|(tid, tenant_shard)| {
+                if !matches!(
+                    tenant_shard.get_scheduling_policy(),
+                    ShardSchedulingPolicy::Active
+                ) {
+                    // Only include tenants in fills if they have a normal (Active) scheduling policy.  We
+                    // even exclude Essential, because moving to fill a node is not essential to keeping this
+                    // tenant available.
+                    return None;
+                }
 
-        // Collect shards which do not have a preferred AZ & are elegible for moving in stage 2
-        let mut free_tids_by_node: HashMap<NodeId, Vec<TenantShardId>> = HashMap::new();
-
-        // Don't respect AZ preferences if there is only one AZ.  This comes up in tests, but it could
-        // conceivably come up in real life if deploying a single-AZ region intentionally.
-        let respect_azs = nodes
-            .values()
-            .map(|n| n.get_availability_zone_id())
-            .unique()
-            .count()
-            > 1;
-
-        // Step 1: collect all shards that we are required to migrate back to this node because their AZ preference
-        // requires it.
-        for (tsid, tenant_shard) in tenants {
-            if !tenant_shard.intent.get_secondary().contains(&node_id) {
-                // Shard doesn't have a secondary on this node, ignore it.
-                continue;
-            }
-
-            // AZ check: when filling nodes after a restart, our intent is to move _back_ the
-            // shards which belong on this node, not to promote shards whose scheduling preference
-            // would be on their currently attached node.  So will avoid promoting shards whose
-            // home AZ doesn't match the AZ of the node we're filling.
-            match tenant_shard.preferred_az() {
-                _ if !respect_azs => {
-                    if let Some(primary) = tenant_shard.intent.get_attached() {
-                        free_tids_by_node.entry(*primary).or_default().push(*tsid);
+                // AZ check: when filling nodes after a restart, our intent is to move _back_ the
+                // shards which belong on this node, not to promote shards whose scheduling preference
+                // would be on their currently attached node.  So will avoid promoting shards whose
+                // home AZ doesn't match the AZ of the node we're filling.
+                match tenant_shard.preferred_az() {
+                    None => {
+                        // Shard doesn't have an AZ preference: it is elegible to be moved.
+                    }
+                    Some(az) if az == &node_az => {
+                        // This shard's home AZ is equal to the node we're filling: it is
+                        // elegible to be moved: fall through;
+                    }
+                    Some(_) => {
+                        // This shard's home AZ is somewhere other than the node we're filling:
+                        // do not include it in the fill plan.
+                        return None;
                     }
                 }
-                None => {
-                    // Shard doesn't have an AZ preference: it is elegible to be moved, but we
-                    // will only do so if our target shard count requires it.
-                    if let Some(primary) = tenant_shard.intent.get_attached() {
-                        free_tids_by_node.entry(*primary).or_default().push(*tsid);
-                    }
-                }
-                Some(az) if az == &node_az => {
-                    // This shard's home AZ is equal to the node we're filling: it should
-                    // be moved back to this node as part of filling, unless its currently
-                    // attached location is also in its home AZ.
-                    if let Some(primary) = tenant_shard.intent.get_attached() {
-                        if nodes
-                            .get(primary)
-                            .expect("referenced node must exist")
-                            .get_availability_zone_id()
-                            != tenant_shard
-                                .preferred_az()
-                                .expect("tenant must have an AZ preference")
-                        {
-                            plan.push(*tsid)
-                        }
-                    } else {
-                        plan.push(*tsid)
-                    }
-                }
-                Some(_) => {
-                    // This shard's home AZ is somewhere other than the node we're filling,
-                    // it may not be moved back to this node as part of filling.  Ignore it
-                }
-            }
-        }
 
-        // Step 2: also promote any AZ-agnostic shards as required to achieve the target number of attachments
-        let fill_requirement = locked.scheduler.compute_fill_requirement(node_id);
+                if tenant_shard.intent.get_secondary().contains(&node_id) {
+                    if let Some(primary) = tenant_shard.intent.get_attached() {
+                        return Some((*primary, *tid));
+                    }
+                }
+
+                None
+            })
+            .into_group_map();
 
         let expected_attached = locked.scheduler.expected_attached_shard_count();
         let nodes_by_load = locked.scheduler.nodes_by_attached_shard_count();
 
         let mut promoted_per_tenant: HashMap<TenantId, usize> = HashMap::new();
+        let mut plan = Vec::new();
 
         for (node_id, attached) in nodes_by_load {
             let available = locked.nodes.get(&node_id).is_some_and(|n| n.is_available());
@@ -7376,7 +7346,7 @@ impl Service {
             }
 
             if plan.len() >= fill_requirement
-                || free_tids_by_node.is_empty()
+                || tids_by_node.is_empty()
                 || attached <= expected_attached
             {
                 break;
@@ -7388,7 +7358,7 @@ impl Service {
 
             let mut remove_node = false;
             while take > 0 {
-                match free_tids_by_node.get_mut(&node_id) {
+                match tids_by_node.get_mut(&node_id) {
                     Some(tids) => match tids.pop() {
                         Some(tid) => {
                             let max_promote_for_tenant = std::cmp::max(
@@ -7414,7 +7384,7 @@ impl Service {
             }
 
             if remove_node {
-                free_tids_by_node.remove(&node_id);
+                tids_by_node.remove(&node_id);
             }
         }
 

storage_controller/src/service/chaos_injector.rs

@@ -1,17 +1,11 @@
-use std::{
-    collections::{BTreeMap, HashMap},
-    sync::Arc,
-    time::Duration,
-};
+use std::{sync::Arc, time::Duration};
 
 use pageserver_api::controller_api::ShardSchedulingPolicy;
 use rand::seq::SliceRandom;
 use rand::thread_rng;
 use tokio_util::sync::CancellationToken;
-use utils::id::NodeId;
-use utils::shard::TenantShardId;
 
-use super::{Node, Scheduler, Service, TenantShard};
+use super::Service;
 
 pub struct ChaosInjector {
     service: Arc<Service>,
@@ -41,86 +35,50 @@ impl ChaosInjector {
         }
     }
 
-    /// If a shard has a secondary and attached location, then re-assign the secondary to be
-    /// attached and the attached to be secondary.
-    ///
-    /// Only modifies tenants if they're in Active scheduling policy.
-    fn maybe_migrate_to_secondary(
-        &self,
-        tenant_shard_id: TenantShardId,
-        nodes: &Arc<HashMap<NodeId, Node>>,
-        tenants: &mut BTreeMap<TenantShardId, TenantShard>,
-        scheduler: &mut Scheduler,
-    ) {
-        let shard = tenants
-            .get_mut(&tenant_shard_id)
-            .expect("Held lock between choosing ID and this get");
-
-        if !matches!(shard.get_scheduling_policy(), ShardSchedulingPolicy::Active) {
-            // Skip non-active scheduling policies, so that a shard with a policy like Pause can
-            // be pinned without being disrupted by us.
-            tracing::info!(
-                "Skipping shard {tenant_shard_id}: scheduling policy is {:?}",
-                shard.get_scheduling_policy()
-            );
-            return;
-        }
-
-        // Pick a secondary to promote
-        let Some(new_location) = shard
-            .intent
-            .get_secondary()
-            .choose(&mut thread_rng())
-            .cloned()
-        else {
-            tracing::info!(
-                "Skipping shard {tenant_shard_id}: no secondary location, can't migrate"
-            );
-            return;
-        };
-
-        let Some(old_location) = *shard.intent.get_attached() else {
-            tracing::info!("Skipping shard {tenant_shard_id}: currently has no attached location");
-            return;
-        };
-
-        tracing::info!("Injecting chaos: migrate {tenant_shard_id} {old_location}->{new_location}");
-
-        shard.intent.demote_attached(scheduler, old_location);
-        shard.intent.promote_attached(scheduler, new_location);
-        self.service.maybe_reconcile_shard(shard, nodes);
-    }
-
     async fn inject_chaos(&mut self) {
         // Pick some shards to interfere with
         let batch_size = 128;
         let mut inner = self.service.inner.write().unwrap();
         let (nodes, tenants, scheduler) = inner.parts_mut();
         let tenant_ids = tenants.keys().cloned().collect::<Vec<_>>();
-
-        // Prefer to migrate tenants that are currently outside their home AZ.  This avoids the chaos injector
-        // continuously pushing tenants outside their home AZ: instead, we'll tend to cycle between picking some
-        // random tenants to move, and then on next chaos iteration moving them back, then picking some new
-        // random tenants on the next iteration.
-        let mut victims = Vec::with_capacity(batch_size);
-        for shard in tenants.values() {
-            if shard.is_attached_outside_preferred_az(nodes) {
-                victims.push(shard.tenant_shard_id);
-            }
-
-            if victims.len() >= batch_size {
-                break;
-            }
-        }
-
-        let choose_random = batch_size.saturating_sub(victims.len());
-        tracing::info!("Injecting chaos: found {} shards to migrate back to home AZ, picking {choose_random} random shards to migrate", victims.len());
-
-        let random_victims = tenant_ids.choose_multiple(&mut thread_rng(), choose_random);
-        victims.extend(random_victims);
+        let victims = tenant_ids.choose_multiple(&mut thread_rng(), batch_size);
 
         for victim in victims {
-            self.maybe_migrate_to_secondary(victim, nodes, tenants, scheduler);
+            let shard = tenants
+                .get_mut(victim)
+                .expect("Held lock between choosing ID and this get");
+
+            if !matches!(shard.get_scheduling_policy(), ShardSchedulingPolicy::Active) {
+                // Skip non-active scheduling policies, so that a shard with a policy like Pause can
+                // be pinned without being disrupted by us.
+                tracing::info!(
+                    "Skipping shard {victim}: scheduling policy is {:?}",
+                    shard.get_scheduling_policy()
+                );
+                continue;
+            }
+
+            // Pick a secondary to promote
+            let Some(new_location) = shard
+                .intent
+                .get_secondary()
+                .choose(&mut thread_rng())
+                .cloned()
+            else {
+                tracing::info!("Skipping shard {victim}: no secondary location, can't migrate");
+                continue;
+            };
+
+            let Some(old_location) = *shard.intent.get_attached() else {
+                tracing::info!("Skipping shard {victim}: currently has no attached location");
+                continue;
+            };
+
+            tracing::info!("Injecting chaos: migrate {victim} {old_location}->{new_location}");
+
+            shard.intent.demote_attached(scheduler, old_location);
+            shard.intent.promote_attached(scheduler, new_location);
+            self.service.maybe_reconcile_shard(shard, nodes);
         }
     }
 }

storage_controller/src/tenant_shard.rs

@@ -1793,23 +1793,6 @@ impl TenantShard {
             }
         }
     }
-
-    /// Returns true if the tenant shard is attached to a node that is outside the preferred AZ.
-    ///
-    /// If the shard does not have a preferred AZ, returns false.
-    pub(crate) fn is_attached_outside_preferred_az(&self, nodes: &HashMap<NodeId, Node>) -> bool {
-        self.intent
-            .get_attached()
-            .map(|node_id| {
-                Some(
-                    nodes
-                        .get(&node_id)
-                        .expect("referenced node exists")
-                        .get_availability_zone_id(),
-                ) == self.intent.preferred_az_id.as_ref()
-            })
-            .unwrap_or(false)
-    }
 }
 
 impl Drop for TenantShard {

test_runner/fixtures/neon_fixtures.py

@@ -370,7 +370,6 @@ class NeonEnvBuilder:
         pageserver_config_override: str | Callable[[dict[str, Any]], None] | None = None,
         num_safekeepers: int = 1,
         num_pageservers: int = 1,
-        num_azs: int = 1,
         # Use non-standard SK ids to check for various parsing bugs
         safekeepers_id_start: int = 0,
         # fsync is disabled by default to make the tests go faster
@@ -402,7 +401,6 @@ class NeonEnvBuilder:
         self.pageserver_config_override = pageserver_config_override
         self.num_safekeepers = num_safekeepers
         self.num_pageservers = num_pageservers
-        self.num_azs = num_azs
         self.safekeepers_id_start = safekeepers_id_start
         self.safekeepers_enable_fsync = safekeepers_enable_fsync
         self.auth_enabled = auth_enabled
@@ -992,7 +990,6 @@ class NeonEnv:
         self.endpoints = EndpointFactory(self)
         self.safekeepers: list[Safekeeper] = []
         self.pageservers: list[NeonPageserver] = []
-        self.num_azs = config.num_azs
         self.broker = NeonBroker(self)
         self.pageserver_remote_storage = config.pageserver_remote_storage
         self.safekeepers_remote_storage = config.safekeepers_remote_storage
@@ -1093,21 +1090,14 @@ class NeonEnv:
                 http=self.port_distributor.get_port(),
             )
 
-            # Availabilty zones may also be configured manually with `NeonEnvBuilder.pageserver_config_override`
-            if self.num_azs > 1:
-                # Round-robin assignment of AZ names like us-east-2a, us-east-2b, etc.
-                az_prefix = DEFAULT_AZ_ID[:-1]
-                availability_zone = f"{az_prefix}{chr(ord('a') + (ps_id - 1) % self.num_azs)}"
-            else:
-                availability_zone = DEFAULT_AZ_ID
-
             ps_cfg: dict[str, Any] = {
                 "id": ps_id,
                 "listen_pg_addr": f"localhost:{pageserver_port.pg}",
                 "listen_http_addr": f"localhost:{pageserver_port.http}",
                 "pg_auth_type": pg_auth_type,
                 "http_auth_type": http_auth_type,
-                "availability_zone": availability_zone,
+                # Default which can be overriden with `NeonEnvBuilder.pageserver_config_override`
+                "availability_zone": DEFAULT_AZ_ID,
                 # Disable pageserver disk syncs in tests: when running tests concurrently, this avoids
                 # the pageserver taking a long time to start up due to syncfs flushing other tests' data
                 "no_sync": True,

test_runner/regress/test_local_file_cache.py

@@ -7,78 +7,9 @@ import threading
 import time
 
 import pytest
-from fixtures.neon_fixtures import NeonEnv, NeonEnvBuilder
+from fixtures.neon_fixtures import NeonEnvBuilder
 from fixtures.utils import USE_LFC, query_scalar
 
-"""
-Test whether LFC doesn't error out when the LRU is empty, but the LFC is
-already at its maximum size.
-
-If we don't handle this safely, we might allocate more hash entries than
-otherwise considered safe, thus causing ERRORs in hash_search(HASH_ENTER) once
-we hit lfc->used >= lfc->limit.
-"""
-
-
-@pytest.mark.skipif(not USE_LFC, reason="LFC is disabled, skipping")
-def test_local_file_cache_all_pinned(neon_simple_env: NeonEnv):
-    env = neon_simple_env
-    endpoint = env.endpoints.create_start(
-        "main",
-        config_lines=[
-            "neon.max_file_cache_size='1MB'",
-            "neon.file_cache_size_limit='1MB'",
-        ],
-    )
-    top_cur = endpoint.connect().cursor()
-
-    stop = threading.Event()
-    n_rows = 10000
-    n_threads = 5
-    n_updates_per_connection = 1000
-
-    top_cur.execute("CREATE TABLE lfctest (id int4 PRIMARY KEY, n int) WITH (fillfactor=10)")
-    top_cur.execute(f"INSERT INTO lfctest SELECT g, 1 FROM generate_series(1, {n_rows}) g")
-
-    # Start threads that will perform random UPDATEs. Each UPDATE
-    # increments the counter on the row, so that we can check at the
-    # end that the sum of all the counters match the number of updates
-    # performed (plus the initial 1 on each row).
-    #
-    # Furthermore, each thread will reconnect between every 1000 updates.
-    def run_updates(n_updates_performed_q: queue.Queue[int]):
-        n_updates_performed = 0
-        conn = endpoint.connect()
-        cur = conn.cursor()
-        while not stop.is_set():
-            id = random.randint(1, n_rows)
-            cur.execute(f"UPDATE lfctest SET n = n + 1 WHERE id = {id}")
-            n_updates_performed += 1
-            if n_updates_performed % n_updates_per_connection == 0:
-                cur.close()
-                conn.close()
-                conn = endpoint.connect()
-                cur = conn.cursor()
-        n_updates_performed_q.put(n_updates_performed)
-
-    n_updates_performed_q: queue.Queue[int] = queue.Queue()
-    threads: list[threading.Thread] = []
-    for _i in range(n_threads):
-        thread = threading.Thread(target=run_updates, args=(n_updates_performed_q,), daemon=True)
-        thread.start()
-        threads.append(thread)
-
-    time.sleep(15)
-
-    stop.set()
-
-    n_updates_performed = 0
-    for thread in threads:
-        thread.join()
-        n_updates_performed += n_updates_performed_q.get()
-
-    assert query_scalar(top_cur, "SELECT SUM(n) FROM lfctest") == n_rows + n_updates_performed
-
 
 @pytest.mark.skipif(not USE_LFC, reason="LFC is disabled, skipping")
 def test_local_file_cache_unlink(neon_env_builder: NeonEnvBuilder):

test_runner/regress/test_page_service_batching_regressions.py

@@ -1,60 +0,0 @@
-# NB: there are benchmarks that double-serve as tests inside the `performance` directory.
-
-import subprocess
-from pathlib import Path
-
-import pytest
-from fixtures.log_helper import log
-from fixtures.neon_fixtures import NeonEnvBuilder
-
-
-@pytest.mark.timeout(30)  # test takes <20s if pageserver impl is correct
-@pytest.mark.parametrize("kind", ["pageserver-stop", "tenant-detach"])
-def test_slow_flush(neon_env_builder: NeonEnvBuilder, neon_binpath: Path, kind: str):
-    def patch_pageserver_toml(config):
-        config["page_service_pipelining"] = {
-            "mode": "pipelined",
-            "max_batch_size": 32,
-            "execution": "concurrent-futures",
-        }
-
-    neon_env_builder.pageserver_config_override = patch_pageserver_toml
-    env = neon_env_builder.init_start()
-
-    log.info("make flush appear slow")
-
-    log.info("sending requests until pageserver accepts no more")
-    # TODO: extract this into a helper, like subprocess_capture,
-    # so that we capture the stderr from the helper somewhere.
-    child = subprocess.Popen(
-        [
-            neon_binpath / "test_helper_slow_client_reads",
-            env.pageserver.connstr(),
-            str(env.initial_tenant),
-            str(env.initial_timeline),
-        ],
-        bufsize=0,  # unbuffered
-        stdin=subprocess.PIPE,
-        stdout=subprocess.PIPE,
-    )
-    assert child.stdout is not None
-    buf = child.stdout.read(1)
-    if len(buf) != 1:
-        raise Exception("unexpected EOF")
-    if buf != b"R":
-        raise Exception(f"unexpected data: {buf!r}")
-    log.info("helper reports pageserver accepts no more requests")
-    log.info(
-        "assuming pageserver connection handle is in a state where TCP has backpressured pageserver=>client response flush() into userspace"
-    )
-
-    if kind == "pageserver-stop":
-        log.info("try to shut down the pageserver cleanly")
-        env.pageserver.stop()
-    elif kind == "tenant-detach":
-        log.info("try to shut down the tenant")
-        env.pageserver.tenant_detach(env.initial_tenant)
-    else:
-        raise ValueError(f"unexpected kind: {kind}")
-
-    log.info("shutdown did not time out, test passed")

test_runner/regress/test_storage_controller.py

@@ -2394,7 +2394,6 @@ def test_storage_controller_node_deletion(
     Test that deleting a node works & properly reschedules everything that was on the node.
     """
     neon_env_builder.num_pageservers = 3
-    neon_env_builder.num_azs = 3
     env = neon_env_builder.init_configs()
     env.start()
 
@@ -2408,9 +2407,6 @@ def test_storage_controller_node_deletion(
             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
-
     victim = env.pageservers[-1]
 
     # The procedure a human would follow is:
@@ -3215,7 +3211,7 @@ def test_safekeeper_deployment_time_update(neon_env_builder: NeonEnvBuilder):
     # some small tests for the scheduling policy querying and returning APIs
     newest_info = target.get_safekeeper(inserted["id"])
     assert newest_info
-    assert newest_info["scheduling_policy"] == "Pause"
+    assert newest_info["scheduling_policy"] == "Disabled"
     target.safekeeper_scheduling_policy(inserted["id"], "Decomissioned")
     newest_info = target.get_safekeeper(inserted["id"])
     assert newest_info

test_runner/regress/test_tenants.py

@@ -194,7 +194,7 @@ def test_metrics_normal_work(neon_env_builder: NeonEnvBuilder):
         io_metrics = query_all_safekeepers(
             "safekeeper_pg_io_bytes_total",
             {
-                "app_name": f"pageserver-{env.pageserver.id}",
+                "app_name": "pageserver",
                 "client_az": "test_ps_az",
                 "dir": io_direction,
                 "same_az": "false",