make Tenant::timelines a tokio::sync::RwLock

This is preliminary work for/from #4220 (async `Layer::get_value_reconstruct_data`). The patch converts `Tenant::timelines` from `std::sync::Mutex` to `tokio::sync::Mutex`. We need this change because we want to switch `Timeline::layers` to an async RwLock. We need that because we hold `Timeline::layers` while calling `Layer::get_value_reconstruct_data`. So, if we want to make get_value_reconstruct_data async, we need to make `Timeline::layers` async first.
fix merge fallout (?)
2026-05-16 20:50:37 +00:00 · 2023-05-24 17:53:12 +02:00 · 2023-05-24 17:42:51 +02:00 · 2023-05-24 17:25:35 +02:00 · 2023-05-24 17:20:03 +02:00 · 2023-05-24 17:19:22 +02:00
17 changed files with 432 additions and 325 deletions
--- a/libs/pageserver_api/src/models.rs
+++ b/libs/pageserver_api/src/models.rs
@@ -37,6 +37,8 @@ pub enum TenantState {
    Loading,
    /// This tenant is being downloaded from cloud storage.
    Attaching,
+    /// The tenant is transitioning from Loading/Attaching to Active.
+    Activating,
    /// Tenant is fully operational
    Active,
    /// A tenant is recognized by pageserver, but it is being detached or the
@@ -60,6 +62,7 @@ impl TenantState {
            // tenant mgr startup distinguishes attaching from loading via marker file.
            // If it's loading, there is no attach marker file, i.e., attach had finished in the past.
            Self::Loading => Attached,
+            Self::Activating => todo!(),
            // We only reach Active after successful load / attach.
            // So, call atttachment status Attached.
            Self::Active => Attached,
--- a/pageserver/src/bin/pageserver.rs
+++ b/pageserver/src/bin/pageserver.rs
@@ -9,6 +9,7 @@ use clap::{Arg, ArgAction, Command};
 use fail::FailScenario;
 use metrics::launch_timestamp::{set_launch_timestamp_metric, LaunchTimestamp};
 use pageserver::disk_usage_eviction_task::{self, launch_disk_usage_global_eviction_task};
+use pageserver::task_mgr::WALRECEIVER_RUNTIME;
 use remote_storage::GenericRemoteStorage;
 use tracing::*;

@@ -18,9 +19,7 @@ use pageserver::{
    context::{DownloadBehavior, RequestContext},
    http, page_cache, page_service, task_mgr,
    task_mgr::TaskKind,
-    task_mgr::{
-        BACKGROUND_RUNTIME, COMPUTE_REQUEST_RUNTIME, MGMT_REQUEST_RUNTIME, WALRECEIVER_RUNTIME,
-    },
+    task_mgr::{BACKGROUND_RUNTIME, COMPUTE_REQUEST_RUNTIME, MGMT_REQUEST_RUNTIME},
    tenant::mgr,
    virtual_file,
 };
@@ -276,7 +275,17 @@ fn start_pageserver(
    let pageserver_listener = tcp_listener::bind(pg_addr)?;

    // Launch broker client
-    WALRECEIVER_RUNTIME.block_on(pageserver::broker_client::init_broker_client(conf))?;
+    let broker_client = WALRECEIVER_RUNTIME
+        .block_on(async {
+            // Note: we do not attempt connecting here (but validate endpoints sanity).
+            storage_broker::connect(conf.broker_endpoint.clone(), conf.broker_keepalive_interval)
+        })
+        .with_context(|| {
+            format!(
+                "create broker client for uri={:?} keepalive_interval={:?}",
+                &conf.broker_endpoint, conf.broker_keepalive_interval,
+            )
+        })?;

    // Initialize authentication for incoming connections
    let http_auth;
@@ -326,7 +335,11 @@ fn start_pageserver(
    let remote_storage = create_remote_storage_client(conf)?;

    // Scan the local 'tenants/' directory and start loading the tenants
-    BACKGROUND_RUNTIME.block_on(mgr::init_tenant_mgr(conf, remote_storage.clone()))?;
+    BACKGROUND_RUNTIME.block_on(mgr::init_tenant_mgr(
+        conf,
+        broker_client.clone(),
+        remote_storage.clone(),
+    ))?;

    // shared state between the disk-usage backed eviction background task and the http endpoint
    // that allows triggering disk-usage based eviction manually. note that the http endpoint
@@ -351,6 +364,7 @@ fn start_pageserver(
            conf,
            launch_ts,
            http_auth,
+            broker_client.clone(),
            remote_storage,
            disk_usage_eviction_state,
        )?
@@ -427,6 +441,7 @@ fn start_pageserver(
            async move {
                page_service::libpq_listener_main(
                    conf,
+                    broker_client,
                    pg_auth,
                    pageserver_listener,
                    conf.pg_auth_type,
--- a/pageserver/src/broker_client.rs
+++ b/pageserver/src/broker_client.rs
@@ -1,48 +0,0 @@
-//! The broker client instance of the pageserver, created during pageserver startup.
-//! Used by each timelines' [`walreceiver`].
-
-use crate::config::PageServerConf;
-
-use anyhow::Context;
-use once_cell::sync::OnceCell;
-use storage_broker::BrokerClientChannel;
-use tracing::*;
-
-static BROKER_CLIENT: OnceCell<BrokerClientChannel> = OnceCell::new();
-
-///
-/// Initialize the broker client. This must be called once at page server startup.
-///
-pub async fn init_broker_client(conf: &'static PageServerConf) -> anyhow::Result<()> {
-    let broker_endpoint = conf.broker_endpoint.clone();
-
-    // Note: we do not attempt connecting here (but validate endpoints sanity).
-    let broker_client =
-        storage_broker::connect(broker_endpoint.clone(), conf.broker_keepalive_interval).context(
-            format!(
-                "Failed to create broker client to {}",
-                &conf.broker_endpoint
-            ),
-        )?;
-
-    if BROKER_CLIENT.set(broker_client).is_err() {
-        panic!("broker already initialized");
-    }
-
-    info!(
-        "Initialized broker client with endpoints: {}",
-        broker_endpoint
-    );
-    Ok(())
-}
-
-///
-/// Get a handle to the broker client
-///
-pub fn get_broker_client() -> &'static BrokerClientChannel {
-    BROKER_CLIENT.get().expect("broker client not initialized")
-}
-
-pub fn is_broker_client_initialized() -> bool {
-    BROKER_CLIENT.get().is_some()
-}
--- a/pageserver/src/consumption_metrics.rs
+++ b/pageserver/src/consumption_metrics.rs
@@ -150,7 +150,7 @@ pub async fn collect_metrics_iteration(
        let mut tenant_resident_size = 0;

        // iterate through list of timelines in tenant
-        for timeline in tenant.list_timelines().iter() {
+        for timeline in tenant.list_timelines().await.iter() {
            // collect per-timeline metrics only for active timelines
            if timeline.is_active() {
                let timeline_written_size = u64::from(timeline.get_last_record_lsn());
--- a/pageserver/src/disk_usage_eviction_task.rs
+++ b/pageserver/src/disk_usage_eviction_task.rs
@@ -508,7 +508,7 @@ async fn collect_eviction_candidates(
        // a little unfair to tenants during shutdown in such a situation is tolerable.
        let mut tenant_candidates = Vec::new();
        let mut max_layer_size = 0;
-        for tl in tenant.list_timelines() {
+        for tl in tenant.list_timelines().await {
            if !tl.is_active() {
                continue;
            }
--- a/pageserver/src/http/routes.rs
+++ b/pageserver/src/http/routes.rs
@@ -7,6 +7,7 @@ use hyper::{Body, Request, Response, Uri};
 use metrics::launch_timestamp::LaunchTimestamp;
 use pageserver_api::models::{DownloadRemoteLayersTaskSpawnRequest, TenantAttachRequest};
 use remote_storage::GenericRemoteStorage;
+use storage_broker::BrokerClientChannel;
 use tenant_size_model::{SizeResult, StorageModel};
 use tokio_util::sync::CancellationToken;
 use tracing::*;
@@ -51,6 +52,7 @@ struct State {
    auth: Option<Arc<JwtAuth>>,
    allowlist_routes: Vec<Uri>,
    remote_storage: Option<GenericRemoteStorage>,
+    broker_client: storage_broker::BrokerClientChannel,
    disk_usage_eviction_state: Arc<disk_usage_eviction_task::State>,
 }

@@ -59,6 +61,7 @@ impl State {
        conf: &'static PageServerConf,
        auth: Option<Arc<JwtAuth>>,
        remote_storage: Option<GenericRemoteStorage>,
+        broker_client: storage_broker::BrokerClientChannel,
        disk_usage_eviction_state: Arc<disk_usage_eviction_task::State>,
    ) -> anyhow::Result<Self> {
        let allowlist_routes = ["/v1/status", "/v1/doc", "/swagger.yml"]
@@ -70,6 +73,7 @@ impl State {
            auth,
            allowlist_routes,
            remote_storage,
+            broker_client,
            disk_usage_eviction_state,
        })
    }
@@ -271,6 +275,8 @@ async fn timeline_create_handler(mut request: Request<Body>) -> Result<Response<

    let ctx = RequestContext::new(TaskKind::MgmtRequest, DownloadBehavior::Error);

+    let state = get_state(&request);
+
    async {
        let tenant = mgr::get_tenant(tenant_id, true).await?;
        match tenant.create_timeline(
@@ -278,6 +284,7 @@ async fn timeline_create_handler(mut request: Request<Body>) -> Result<Response<
            request_data.ancestor_timeline_id.map(TimelineId::from),
            request_data.ancestor_start_lsn,
            request_data.pg_version.unwrap_or(crate::DEFAULT_PG_VERSION),
+            state.broker_client.clone(),
            &ctx,
        )
        .await {
@@ -305,7 +312,7 @@ async fn timeline_list_handler(request: Request<Body>) -> Result<Response<Body>,

    let response_data = async {
        let tenant = mgr::get_tenant(tenant_id, true).await?;
-        let timelines = tenant.list_timelines();
+        let timelines = tenant.list_timelines().await;

        let mut response_data = Vec::with_capacity(timelines.len());
        for timeline in timelines {
@@ -344,6 +351,7 @@ async fn timeline_detail_handler(request: Request<Body>) -> Result<Response<Body

        let timeline = tenant
            .get_timeline(timeline_id, false)
+            .await
            .map_err(ApiError::NotFound)?;

        let timeline_info = build_timeline_info(
@@ -408,6 +416,7 @@ async fn tenant_attach_handler(mut request: Request<Body>) -> Result<Response<Bo
            state.conf,
            tenant_id,
            tenant_conf,
+            state.broker_client.clone(),
            remote_storage.clone(),
            &ctx,
        )
@@ -457,9 +466,15 @@ async fn tenant_load_handler(request: Request<Body>) -> Result<Response<Body>, A
    let ctx = RequestContext::new(TaskKind::MgmtRequest, DownloadBehavior::Warn);

    let state = get_state(&request);
-    mgr::load_tenant(state.conf, tenant_id, state.remote_storage.clone(), &ctx)
-        .instrument(info_span!("load", tenant = %tenant_id))
-        .await?;
+    mgr::load_tenant(
+        state.conf,
+        tenant_id,
+        state.broker_client.clone(),
+        state.remote_storage.clone(),
+        &ctx,
+    )
+    .instrument(info_span!("load", tenant = %tenant_id))
+    .await?;

    json_response(StatusCode::ACCEPTED, ())
 }
@@ -506,7 +521,7 @@ async fn tenant_status(request: Request<Body>) -> Result<Response<Body>, ApiErro

        // Calculate total physical size of all timelines
        let mut current_physical_size = 0;
-        for timeline in tenant.list_timelines().iter() {
+        for timeline in tenant.list_timelines().await.iter() {
            current_physical_size += timeline.layer_size_sum();
        }

@@ -744,6 +759,7 @@ async fn tenant_create_handler(mut request: Request<Body>) -> Result<Response<Bo
        state.conf,
        tenant_conf,
        target_tenant_id,
+        state.broker_client.clone(),
        state.remote_storage.clone(),
        &ctx,
    )
@@ -820,7 +836,7 @@ async fn handle_tenant_break(r: Request<Body>) -> Result<Response<Body>, ApiErro
        .await
        .map_err(|_| ApiError::Conflict(String::from("no active tenant found")))?;

-    tenant.set_broken("broken from test".to_owned());
+    tenant.set_broken("broken from test".to_owned()).await;

    json_response(StatusCode::OK, ())
 }
@@ -960,6 +976,7 @@ async fn active_timeline_of_active_tenant(
    let tenant = mgr::get_tenant(tenant_id, true).await?;
    tenant
        .get_timeline(timeline_id, true)
+        .await
        .map_err(ApiError::NotFound)
 }

@@ -1099,6 +1116,7 @@ pub fn make_router(
    conf: &'static PageServerConf,
    launch_ts: &'static LaunchTimestamp,
    auth: Option<Arc<JwtAuth>>,
+    broker_client: BrokerClientChannel,
    remote_storage: Option<GenericRemoteStorage>,
    disk_usage_eviction_state: Arc<disk_usage_eviction_task::State>,
 ) -> anyhow::Result<RouterBuilder<hyper::Body, ApiError>> {
@@ -1145,8 +1163,14 @@ pub fn make_router(

    Ok(router
        .data(Arc::new(
-            State::new(conf, auth, remote_storage, disk_usage_eviction_state)
-                .context("Failed to initialize router state")?,
+            State::new(
+                conf,
+                auth,
+                remote_storage,
+                broker_client,
+                disk_usage_eviction_state,
+            )
+            .context("Failed to initialize router state")?,
        ))
        .get("/v1/status", |r| RequestSpan(status_handler).handle(r))
        .put(
--- a/pageserver/src/lib.rs
+++ b/pageserver/src/lib.rs
@@ -1,6 +1,5 @@
 mod auth;
 pub mod basebackup;
-pub mod broker_client;
 pub mod config;
 pub mod consumption_metrics;
 pub mod context;
--- a/pageserver/src/page_service.rs
+++ b/pageserver/src/page_service.rs
@@ -172,6 +172,7 @@ async fn read_tar_eof(mut reader: (impl AsyncRead + Unpin)) -> anyhow::Result<()
 ///
 pub async fn libpq_listener_main(
    conf: &'static PageServerConf,
+    broker_client: storage_broker::BrokerClientChannel,
    auth: Option<Arc<JwtAuth>>,
    listener: TcpListener,
    auth_type: AuthType,
@@ -213,7 +214,14 @@ pub async fn libpq_listener_main(
                    None,
                    "serving compute connection task",
                    false,
-                    page_service_conn_main(conf, local_auth, socket, auth_type, connection_ctx),
+                    page_service_conn_main(
+                        conf,
+                        broker_client.clone(),
+                        local_auth,
+                        socket,
+                        auth_type,
+                        connection_ctx,
+                    ),
                );
            }
            Err(err) => {
@@ -230,6 +238,7 @@ pub async fn libpq_listener_main(

 async fn page_service_conn_main(
    conf: &'static PageServerConf,
+    broker_client: storage_broker::BrokerClientChannel,
    auth: Option<Arc<JwtAuth>>,
    socket: tokio::net::TcpStream,
    auth_type: AuthType,
@@ -266,7 +275,7 @@ async fn page_service_conn_main(
    // and create a child per-query context when it invokes process_query.
    // But it's in a shared crate, so, we store connection_ctx inside PageServerHandler
    // and create the per-query context in process_query ourselves.
-    let mut conn_handler = PageServerHandler::new(conf, auth, connection_ctx);
+    let mut conn_handler = PageServerHandler::new(conf, broker_client, auth, connection_ctx);
    let pgbackend = PostgresBackend::new_from_io(socket, peer_addr, auth_type, None)?;

    match pgbackend
@@ -324,6 +333,7 @@ impl PageRequestMetrics {

 struct PageServerHandler {
    _conf: &'static PageServerConf,
+    broker_client: storage_broker::BrokerClientChannel,
    auth: Option<Arc<JwtAuth>>,
    claims: Option<Claims>,

@@ -337,11 +347,13 @@ struct PageServerHandler {
 impl PageServerHandler {
    pub fn new(
        conf: &'static PageServerConf,
+        broker_client: storage_broker::BrokerClientChannel,
        auth: Option<Arc<JwtAuth>>,
        connection_ctx: RequestContext,
    ) -> Self {
        PageServerHandler {
            _conf: conf,
+            broker_client,
            auth,
            claims: None,
            connection_ctx,
@@ -376,7 +388,7 @@ impl PageServerHandler {
        };

        // Check that the timeline exists
-        let timeline = tenant.get_timeline(timeline_id, true)?;
+        let timeline = tenant.get_timeline(timeline_id, true).await?;

        // switch client to COPYBOTH
        pgb.write_message_noflush(&BeMessage::CopyBothResponse)?;
@@ -475,7 +487,9 @@ impl PageServerHandler {
        // Create empty timeline
        info!("creating new timeline");
        let tenant = get_active_tenant_with_timeout(tenant_id, &ctx).await?;
-        let timeline = tenant.create_empty_timeline(timeline_id, base_lsn, pg_version, &ctx)?;
+        let timeline = tenant
+            .create_empty_timeline(timeline_id, base_lsn, pg_version, &ctx)
+            .await?;

        // TODO mark timeline as not ready until it reaches end_lsn.
        // We might have some wal to import as well, and we should prevent compute
@@ -494,7 +508,12 @@ impl PageServerHandler {

        let mut copyin_reader = pin!(StreamReader::new(copyin_stream(pgb)));
        timeline
-            .import_basebackup_from_tar(&mut copyin_reader, base_lsn, &ctx)
+            .import_basebackup_from_tar(
+                &mut copyin_reader,
+                base_lsn,
+                self.broker_client.clone(),
+                &ctx,
+            )
            .await?;

        // Read the end of the tar archive.
@@ -1184,6 +1203,6 @@ async fn get_active_tenant_timeline(
    ctx: &RequestContext,
 ) -> Result<Arc<Timeline>, GetActiveTenantError> {
    let tenant = get_active_tenant_with_timeout(tenant_id, ctx).await?;
-    let timeline = tenant.get_timeline(timeline_id, true)?;
+    let timeline = tenant.get_timeline(timeline_id, true).await?;
    Ok(timeline)
 }
--- a/pageserver/src/pgdatadir_mapping.rs
+++ b/pageserver/src/pgdatadir_mapping.rs
@@ -1594,13 +1594,15 @@ fn is_slru_block_key(key: Key) -> bool {
 }

 #[cfg(test)]
-pub fn create_test_timeline(
+pub async fn create_test_timeline(
    tenant: &crate::tenant::Tenant,
    timeline_id: utils::id::TimelineId,
    pg_version: u32,
    ctx: &RequestContext,
 ) -> anyhow::Result<std::sync::Arc<Timeline>> {
-    let tline = tenant.create_test_timeline(timeline_id, Lsn(8), pg_version, ctx)?;
+    let tline = tenant
+        .create_test_timeline(timeline_id, Lsn(8), pg_version, ctx)
+        .await?;
    let mut m = tline.begin_modification(Lsn(8));
    m.init_empty()?;
    m.commit()?;
@@ -1630,7 +1632,7 @@ mod tests {
    #[test]
    fn test_list_rels_drop() -> Result<()> {
        let repo = RepoHarness::create("test_list_rels_drop")?.load();
-        let tline = create_empty_timeline(repo, TIMELINE_ID)?;
+        let tline = create_empty_timeline(repo, TIMELINE_ID).await?;
        const TESTDB: u32 = 111;

        // Import initial dummy checkpoint record, otherwise the get_timeline() call
--- a/pageserver/src/tenant.rs
+++ b/pageserver/src/tenant.rs
@@ -16,6 +16,7 @@ use futures::FutureExt;
 use pageserver_api::models::TimelineState;
 use remote_storage::DownloadError;
 use remote_storage::GenericRemoteStorage;
+use storage_broker::BrokerClientChannel;
 use tokio::sync::watch;
 use tokio::task::JoinSet;
 use tracing::*;
@@ -39,8 +40,7 @@ use std::process::Stdio;
 use std::sync::atomic::AtomicU64;
 use std::sync::atomic::Ordering;
 use std::sync::Arc;
-use std::sync::MutexGuard;
-use std::sync::{Mutex, RwLock};
+use std::sync::RwLock;
 use std::time::{Duration, Instant};

 use self::config::TenantConf;
@@ -136,7 +136,7 @@ pub struct Tenant {
    tenant_conf: Arc<RwLock<TenantConfOpt>>,

    tenant_id: TenantId,
-    timelines: Mutex<HashMap<TimelineId, Arc<Timeline>>>,
+    timelines: tokio::sync::Mutex<HashMap<TimelineId, Arc<Timeline>>>,
    // This mutex prevents creation of new timelines during GC.
    // Adding yet another mutex (in addition to `timelines`) is needed because holding
    // `timelines` mutex during all GC iteration
@@ -187,7 +187,7 @@ impl UninitializedTimeline<'_> {
    /// This function launches the flush loop if not already done.
    ///
    /// The caller is responsible for activating the timeline (function `.activate()`).
-    fn initialize_with_lock(
+    async fn initialize_with_lock(
        mut self,
        _ctx: &RequestContext,
        timelines: &mut HashMap<TimelineId, Arc<Timeline>>,
@@ -238,6 +238,7 @@ impl UninitializedTimeline<'_> {
        self,
        copyin_read: &mut (impl tokio::io::AsyncRead + Send + Sync + Unpin),
        base_lsn: Lsn,
+        broker_client: storage_broker::BrokerClientChannel,
        ctx: &RequestContext,
    ) -> anyhow::Result<Arc<Timeline>> {
        let raw_timeline = self.raw_timeline()?;
@@ -262,9 +263,11 @@ impl UninitializedTimeline<'_> {

        // Initialize without loading the layer map. We started with an empty layer map, and already
        // updated it for the layers that we created during the import.
-        let mut timelines = self.owning_tenant.timelines.lock().unwrap();
-        let tl = self.initialize_with_lock(ctx, &mut timelines, false)?;
-        tl.activate(ctx)?;
+        let mut timelines = self.owning_tenant.timelines.lock().await;
+        let tl = self
+            .initialize_with_lock(ctx, &mut timelines, false)
+            .await?;
+        tl.activate(broker_client, ctx);
        Ok(tl)
    }

@@ -483,7 +486,7 @@ impl Tenant {

        let timeline = {
            // avoiding holding it across awaits
-            let mut timelines_accessor = self.timelines.lock().unwrap();
+            let mut timelines_accessor = self.timelines.lock().await;
            if timelines_accessor.contains_key(&timeline_id) {
                anyhow::bail!(
                    "Timeline {tenant_id}/{timeline_id} already exists in the tenant map"
@@ -505,7 +508,10 @@ impl Tenant {
            // Do not start walreceiver here. We do need loaded layer map for reconcile_with_remote
            // But we shouldnt start walreceiver before we have all the data locally, because working walreceiver
            // will ingest data which may require looking at the layers which are not yet available locally
-            match timeline.initialize_with_lock(ctx, &mut timelines_accessor, true) {
+            match timeline
+                .initialize_with_lock(ctx, &mut timelines_accessor, true)
+                .await
+            {
                Ok(new_timeline) => new_timeline,
                Err(e) => {
                    error!("Failed to initialize timeline {tenant_id}/{timeline_id}: {e:?}");
@@ -585,6 +591,7 @@ impl Tenant {
    pub(crate) fn spawn_attach(
        conf: &'static PageServerConf,
        tenant_id: TenantId,
+        broker_client: storage_broker::BrokerClientChannel,
        remote_storage: GenericRemoteStorage,
        ctx: &RequestContext,
    ) -> anyhow::Result<Arc<Tenant>> {
@@ -616,13 +623,13 @@ impl Tenant {
            async move {
                let doit = async {
                    tenant_clone.attach(&ctx).await?;
-                    tenant_clone.activate(&ctx)?;
+                    tenant_clone.activate(broker_client, &ctx).await?;
                    anyhow::Ok(())
                };
                match doit.await {
                    Ok(_) => {}
                    Err(e) => {
-                        tenant_clone.set_broken(e.to_string());
+                        tenant_clone.set_broken(e.to_string()).await;
                        error!("error attaching tenant: {:?}", e);
                    }
                }
@@ -761,7 +768,7 @@ impl Tenant {
    pub async fn get_remote_size(&self) -> anyhow::Result<u64> {
        let mut size = 0;

-        for timeline in self.list_timelines().iter() {
+        for timeline in self.list_timelines().await.iter() {
            if let Some(remote_client) = &timeline.remote_client {
                size += remote_client.get_remote_physical_size();
            }
@@ -787,7 +794,7 @@ impl Tenant {
            .context("Failed to create new timeline directory")?;

        let ancestor = if let Some(ancestor_id) = remote_metadata.ancestor_timeline() {
-            let timelines = self.timelines.lock().unwrap();
+            let timelines = self.timelines.lock().await;
            Some(Arc::clone(timelines.get(&ancestor_id).ok_or_else(
                || {
                    anyhow::anyhow!(
@@ -854,6 +861,7 @@ impl Tenant {
    pub fn spawn_load(
        conf: &'static PageServerConf,
        tenant_id: TenantId,
+        broker_client: storage_broker::BrokerClientChannel,
        remote_storage: Option<GenericRemoteStorage>,
        ctx: &RequestContext,
    ) -> Arc<Tenant> {
@@ -890,13 +898,13 @@ impl Tenant {
            async move {
                let doit = async {
                    tenant_clone.load(&ctx).await?;
-                    tenant_clone.activate(&ctx)?;
+                    tenant_clone.activate(broker_client, &ctx).await?;
                    anyhow::Ok(())
                };
                match doit.await {
                    Ok(()) => {}
                    Err(err) => {
-                        tenant_clone.set_broken(err.to_string());
+                        tenant_clone.set_broken(err.to_string()).await;
                        error!("could not load tenant {tenant_id}: {err:?}");
                    }
                }
@@ -1105,7 +1113,7 @@ impl Tenant {
        };

        let ancestor = if let Some(ancestor_timeline_id) = local_metadata.ancestor_timeline() {
-            let ancestor_timeline = self.get_timeline(ancestor_timeline_id, false)
+            let ancestor_timeline = self.get_timeline(ancestor_timeline_id, false).await
            .with_context(|| anyhow::anyhow!("cannot find ancestor timeline {ancestor_timeline_id} for timeline {timeline_id}"))?;
            Some(ancestor_timeline)
        } else {
@@ -1130,12 +1138,12 @@ impl Tenant {

    /// Get Timeline handle for given Neon timeline ID.
    /// This function is idempotent. It doesn't change internal state in any way.
-    pub fn get_timeline(
+    pub async fn get_timeline(
        &self,
        timeline_id: TimelineId,
        active_only: bool,
    ) -> anyhow::Result<Arc<Timeline>> {
-        let timelines_accessor = self.timelines.lock().unwrap();
+        let timelines_accessor = self.timelines.lock().await;
        let timeline = timelines_accessor.get(&timeline_id).with_context(|| {
            format!("Timeline {}/{} was not found", self.tenant_id, timeline_id)
        })?;
@@ -1154,10 +1162,10 @@ impl Tenant {

    /// Lists timelines the tenant contains.
    /// Up to tenant's implementation to omit certain timelines that ar not considered ready for use.
-    pub fn list_timelines(&self) -> Vec<Arc<Timeline>> {
+    pub async fn list_timelines(&self) -> Vec<Arc<Timeline>> {
        self.timelines
            .lock()
-            .unwrap()
+            .await
            .values()
            .map(Arc::clone)
            .collect()
@@ -1166,7 +1174,7 @@ impl Tenant {
    /// This is used to create the initial 'main' timeline during bootstrapping,
    /// or when importing a new base backup. The caller is expected to load an
    /// initial image of the datadir to the new timeline after this.
-    pub fn create_empty_timeline(
+    pub async fn create_empty_timeline(
        &self,
        new_timeline_id: TimelineId,
        initdb_lsn: Lsn,
@@ -1178,7 +1186,7 @@ impl Tenant {
            "Cannot create empty timelines on inactive tenant"
        );

-        let timelines = self.timelines.lock().unwrap();
+        let timelines = self.timelines.lock().await;
        let timeline_uninit_mark = self.create_timeline_uninit_mark(new_timeline_id, &timelines)?;
        drop(timelines);

@@ -1206,16 +1214,20 @@ impl Tenant {
    // This makes the various functions which anyhow::ensure! for Active state work in tests.
    // Our current tests don't need the background loops.
    #[cfg(test)]
-    pub fn create_test_timeline(
+    pub async fn create_test_timeline(
        &self,
        new_timeline_id: TimelineId,
        initdb_lsn: Lsn,
        pg_version: u32,
        ctx: &RequestContext,
    ) -> anyhow::Result<Arc<Timeline>> {
-        let uninit_tl = self.create_empty_timeline(new_timeline_id, initdb_lsn, pg_version, ctx)?;
-        let mut timelines = self.timelines.lock().unwrap();
-        let tl = uninit_tl.initialize_with_lock(ctx, &mut timelines, true)?;
+        let uninit_tl = self
+            .create_empty_timeline(new_timeline_id, initdb_lsn, pg_version, ctx)
+            .await?;
+        let mut timelines = self.timelines.lock().await;
+        let tl = uninit_tl
+            .initialize_with_lock(ctx, &mut timelines, true)
+            .await?;
        // The non-test code would call tl.activate() here.
        tl.set_state(TimelineState::Active);
        Ok(tl)
@@ -1234,6 +1246,7 @@ impl Tenant {
        ancestor_timeline_id: Option<TimelineId>,
        mut ancestor_start_lsn: Option<Lsn>,
        pg_version: u32,
+        broker_client: storage_broker::BrokerClientChannel,
        ctx: &RequestContext,
    ) -> anyhow::Result<Option<Arc<Timeline>>> {
        anyhow::ensure!(
@@ -1241,7 +1254,7 @@ impl Tenant {
            "Cannot create timelines on inactive tenant"
        );

-        if let Ok(existing) = self.get_timeline(new_timeline_id, false) {
+        if let Ok(existing) = self.get_timeline(new_timeline_id, false).await {
            debug!("timeline {new_timeline_id} already exists");

            if let Some(remote_client) = existing.remote_client.as_ref() {
@@ -1266,6 +1279,7 @@ impl Tenant {
            Some(ancestor_timeline_id) => {
                let ancestor_timeline = self
                    .get_timeline(ancestor_timeline_id, false)
+                    .await
                    .context("Cannot branch off the timeline that's not present in pageserver")?;

                if let Some(lsn) = ancestor_start_lsn.as_mut() {
@@ -1300,7 +1314,7 @@ impl Tenant {
            }
        };

-        loaded_timeline.activate(ctx).context("activate timeline")?;
+        loaded_timeline.activate(broker_client, ctx);

        if let Some(remote_client) = loaded_timeline.remote_client.as_ref() {
            // Wait for the upload of the 'index_part.json` file to finish, so that when we return
@@ -1360,7 +1374,7 @@ impl Tenant {
        // compactions.  We don't want to block everything else while the
        // compaction runs.
        let timelines_to_compact = {
-            let timelines = self.timelines.lock().unwrap();
+            let timelines = self.timelines.lock().await;
            let timelines_to_compact = timelines
                .iter()
                .map(|(timeline_id, timeline)| (*timeline_id, timeline.clone()))
@@ -1389,7 +1403,7 @@ impl Tenant {
        // flushing. We don't want to block everything else while the
        // flushing is performed.
        let timelines_to_flush = {
-            let timelines = self.timelines.lock().unwrap();
+            let timelines = self.timelines.lock().await;
            timelines
                .iter()
                .map(|(_id, timeline)| Arc::clone(timeline))
@@ -1414,7 +1428,7 @@ impl Tenant {
        // Transition the timeline into TimelineState::Stopping.
        // This should prevent new operations from starting.
        let timeline = {
-            let mut timelines = self.timelines.lock().unwrap();
+            let mut timelines = self.timelines.lock().await;

            // Ensure that there are no child timelines **attached to that pageserver**,
            // because detach removes files, which will break child branches
@@ -1448,7 +1462,10 @@ impl Tenant {

        // Stop the walreceiver first.
        debug!("waiting for wal receiver to shutdown");
-        timeline.walreceiver.stop().await;
+        let maybe_started_walreceiver = { timeline.walreceiver.lock().unwrap().take() };
+        if let Some(walreceiver) = maybe_started_walreceiver {
+            walreceiver.stop().await;
+        }
        debug!("wal receiver shutdown confirmed");

        // Prevent new uploads from starting.
@@ -1561,7 +1578,7 @@ impl Tenant {
        });

        // Remove the timeline from the map.
-        let mut timelines = self.timelines.lock().unwrap();
+        let mut timelines = self.timelines.lock().await;
        let children_exist = timelines
            .iter()
            .any(|(_, entry)| entry.get_ancestor_timeline_id() == Some(timeline_id));
@@ -1605,12 +1622,21 @@ impl Tenant {
    }

    /// Changes tenant status to active, unless shutdown was already requested.
-    fn activate(&self, ctx: &RequestContext) -> anyhow::Result<()> {
+    async fn activate(
+        &self,
+        broker_client: BrokerClientChannel,
+        ctx: &RequestContext,
+    ) -> anyhow::Result<()> {
        debug_assert_current_span_has_tenant_id();

        let mut result = Ok(());
+        let mut activating = false;
        self.state.send_modify(|current_state| {
            match &*current_state {
+                TenantState::Activating => {
+                    // activate() was called on an already Activating tenant. Shouldn't happen.
+                    result = Err(anyhow::anyhow!("Tenant is already activating"));
+                }
                TenantState::Active => {
                    // activate() was called on an already Active tenant. Shouldn't happen.
                    result = Err(anyhow::anyhow!("Tenant is already active"));
@@ -1627,85 +1653,88 @@ impl Tenant {
                    info!("Tenant is already in Stopping state, skipping activation");
                }
                TenantState::Loading | TenantState::Attaching => {
-                    *current_state = TenantState::Active;
-
+                    *current_state = TenantState::Activating;
                    debug!(tenant_id = %self.tenant_id, "Activating tenant");
-
-                    let timelines_accessor = self.timelines.lock().unwrap();
-                    let not_broken_timelines = timelines_accessor
-                        .values()
-                        .filter(|timeline| timeline.current_state() != TimelineState::Broken);
-
-                    // Spawn gc and compaction loops. The loops will shut themselves
-                    // down when they notice that the tenant is inactive.
-                    tasks::start_background_loops(self.tenant_id);
-
-                    let mut activated_timelines = 0;
-                    let mut timelines_broken_during_activation = 0;
-
-                    for timeline in not_broken_timelines {
-                        match timeline
-                            .activate(ctx)
-                            .context("timeline activation for activating tenant")
-                        {
-                            Ok(()) => {
-                                activated_timelines += 1;
-                            }
-                            Err(e) => {
-                                error!(
-                                    "Failed to activate timeline {}: {:#}",
-                                    timeline.timeline_id, e
-                                );
-                                timeline.set_state(TimelineState::Broken);
-                                *current_state = TenantState::broken_from_reason(format!(
-                                    "failed to activate timeline {}: {}",
-                                    timeline.timeline_id, e
-                                ));
-
-                                timelines_broken_during_activation += 1;
-                            }
-                        }
-                    }
-
-                    let elapsed = self.loading_started_at.elapsed();
-                    let total_timelines = timelines_accessor.len();
-
-                    // log a lot of stuff, because some tenants sometimes suffer from user-visible
-                    // times to activate. see https://github.com/neondatabase/neon/issues/4025
-                    info!(
-                        since_creation_millis = elapsed.as_millis(),
-                        tenant_id = %self.tenant_id,
-                        activated_timelines,
-                        timelines_broken_during_activation,
-                        total_timelines,
-                        post_state = <&'static str>::from(&*current_state),
-                        "activation attempt finished"
-                    );
+                    activating = true;
+                    // Continue outside the closure. We need to grab timelines.lock()
+                    // and we plan to turn it into a tokio::sync::Mutex in a future patch.
                }
            }
        });
-        result
+        if let Err(e) = result {
+            assert!(!activating, "transition into Activating is infallible");
+            return Err(e);
+        }
+
+        if activating {
+            let timelines_accessor = self.timelines.lock().await;
+            let not_broken_timelines = timelines_accessor
+                .values()
+                .filter(|timeline| timeline.current_state() != TimelineState::Broken);
+
+            // Spawn gc and compaction loops. The loops will shut themselves
+            // down when they notice that the tenant is inactive.
+            tasks::start_background_loops(self.tenant_id);
+
+            let mut activated_timelines = 0;
+
+            for timeline in not_broken_timelines {
+                timeline.activate(broker_client.clone(), ctx);
+                activated_timelines += 1;
+            }
+
+            self.state.send_modify(move |current_state| {
+                assert!(
+                    *current_state == TenantState::Activating,
+                    "set_stopping and set_broken wait for us to leave Activating state",
+                );
+                *current_state = TenantState::Active;
+
+                let elapsed = self.loading_started_at.elapsed();
+                let total_timelines = timelines_accessor.len();
+
+                // log a lot of stuff, because some tenants sometimes suffer from user-visible
+                // times to activate. see https://github.com/neondatabase/neon/issues/4025
+                info!(
+                    since_creation_millis = elapsed.as_millis(),
+                    tenant_id = %self.tenant_id,
+                    activated_timelines,
+                    total_timelines,
+                    post_state = <&'static str>::from(&*current_state),
+                    "activation attempt finished"
+                );
+            });
+        };
+        Ok(())
    }

-    /// Change tenant status to Stopping, to mark that it is being shut down
-    pub fn set_stopping(&self) {
+    /// Change tenant status to Stopping, to mark that it is being shut down.
+    ///
+    /// This function is not cancel-safe!
+    pub async fn set_stopping(&self) {
+        // Get the rx before checking state inside send_if_modified.
+        // This way, when we later rx.changed().await, we won't have missed
+        // any state changes.
+        let mut rx = self.state.subscribe();
+        while *rx.borrow() == TenantState::Activating {
+            rx.changed()
+                .await
+                .expect("we're a method on Tenant, so, we're keeping self.state alive here");
+        }
+        let mut stopping = false;
        self.state.send_modify(|current_state| {
            match current_state {
+                TenantState::Activating => unreachable!("we checked above and never transition back into Activating state"),
+                // FIXME: If the tenant is still Loading or Attaching, new timelines
+                // might be created after this. That's harmless, as the Timelines
+                // won't be accessible to anyone, when the Tenant is in Stopping
+                // state.
                TenantState::Active | TenantState::Loading | TenantState::Attaching => {
                    *current_state = TenantState::Stopping;
-
-                    // FIXME: If the tenant is still Loading or Attaching, new timelines
-                    // might be created after this. That's harmless, as the Timelines
-                    // won't be accessible to anyone, when the Tenant is in Stopping
-                    // state.
-                    let timelines_accessor = self.timelines.lock().unwrap();
-                    let not_broken_timelines = timelines_accessor
-                        .values()
-                        .filter(|timeline| timeline.current_state() != TimelineState::Broken);
-                    for timeline in not_broken_timelines {
-                        timeline.set_state(TimelineState::Stopping);
-                    }
-                }
+                    stopping = true;
+                    // Continue outside the closure. We need to grab timelines.lock()
+                    // and we plan to turn it into a tokio::sync::Mutex in a future patch.
+              }
                TenantState::Broken { reason, .. } => {
                    info!("Cannot set tenant to Stopping state, it is in Broken state due to: {reason}");
                }
@@ -1716,11 +1745,30 @@ impl Tenant {
                }
            }
        });
+
+        if stopping {
+            let timelines_accessor = self.timelines.lock().await;
+            let not_broken_timelines = timelines_accessor
+                .values()
+                .filter(|timeline| timeline.current_state() != TimelineState::Broken);
+            for timeline in not_broken_timelines {
+                timeline.set_state(TimelineState::Stopping);
+            }
+        }
    }

-    pub fn set_broken(&self, reason: String) {
+    pub async fn set_broken(&self, reason: String) {
+        let mut rx = self.state.subscribe();
+        while *rx.borrow() == TenantState::Activating {
+            rx.changed()
+                .await
+                .expect("we're a method on Tenant, so, we're keeping self.state alive here");
+        }
        self.state.send_modify(|current_state| {
            match *current_state {
+                TenantState::Activating => {
+                    unreachable!("we checked above and never transition back into Activating state")
+                }
                TenantState::Active => {
                    // Broken tenants can currently only used for fatal errors that happen
                    // while loading or attaching a tenant. A tenant that has already been
@@ -1758,7 +1806,7 @@ impl Tenant {
        loop {
            let current_state = receiver.borrow_and_update().clone();
            match current_state {
-                TenantState::Loading | TenantState::Attaching => {
+                TenantState::Loading | TenantState::Attaching | TenantState::Activating => {
                    // in these states, there's a chance that we can reach ::Active
                    receiver.changed().await?;
                }
@@ -1908,12 +1956,12 @@ impl Tenant {
            .or(self.conf.default_tenant_conf.min_resident_size_override)
    }

-    pub fn set_new_tenant_config(&self, new_tenant_conf: TenantConfOpt) {
+    pub async fn set_new_tenant_config(&self, new_tenant_conf: TenantConfOpt) {
        *self.tenant_conf.write().unwrap() = new_tenant_conf;
        // Don't hold self.timelines.lock() during the notifies.
        // There's no risk of deadlock right now, but there could be if we consolidate
        // mutexes in struct Timeline in the future.
-        let timelines = self.list_timelines();
+        let timelines = self.list_timelines().await;
        for timeline in timelines {
            timeline.tenant_conf_updated();
        }
@@ -1991,7 +2039,7 @@ impl Tenant {
            // activation times.
            loading_started_at: Instant::now(),
            tenant_conf: Arc::new(RwLock::new(tenant_conf)),
-            timelines: Mutex::new(HashMap::new()),
+            timelines: tokio::sync::Mutex::new(HashMap::new()),
            gc_cs: tokio::sync::Mutex::new(()),
            walredo_mgr,
            remote_storage,
@@ -2220,7 +2268,7 @@ impl Tenant {
        // Scan all timelines. For each timeline, remember the timeline ID and
        // the branch point where it was created.
        let (all_branchpoints, timeline_ids): (BTreeSet<(TimelineId, Lsn)>, _) = {
-            let timelines = self.timelines.lock().unwrap();
+            let timelines = self.timelines.lock().await;
            let mut all_branchpoints = BTreeSet::new();
            let timeline_ids = {
                if let Some(target_timeline_id) = target_timeline_id.as_ref() {
@@ -2267,6 +2315,7 @@ impl Tenant {
            // Timeline is known to be local and loaded.
            let timeline = self
                .get_timeline(timeline_id, false)
+                .await
                .with_context(|| format!("Timeline {timeline_id} was not found"))?;

            // If target_timeline is specified, ignore all other timelines
@@ -2352,7 +2401,7 @@ impl Tenant {
        // Create a placeholder for the new branch. This will error
        // out if the new timeline ID is already in use.
        let timeline_uninit_mark = {
-            let timelines = self.timelines.lock().unwrap();
+            let timelines = self.timelines.lock().await;
            self.create_timeline_uninit_mark(dst_id, &timelines)?
        };

@@ -2419,7 +2468,7 @@ impl Tenant {
        );

        let new_timeline = {
-            let mut timelines = self.timelines.lock().unwrap();
+            let mut timelines = self.timelines.lock().await;
            self.prepare_timeline(
                dst_id,
                &metadata,
@@ -2427,7 +2476,8 @@ impl Tenant {
                false,
                Some(Arc::clone(src_timeline)),
            )?
-            .initialize_with_lock(ctx, &mut timelines, true)?
+            .initialize_with_lock(ctx, &mut timelines, true)
+            .await?
        };

        // Root timeline gets its layers during creation and uploads them along with the metadata.
@@ -2457,7 +2507,7 @@ impl Tenant {
        ctx: &RequestContext,
    ) -> anyhow::Result<Arc<Timeline>> {
        let timeline_uninit_mark = {
-            let timelines = self.timelines.lock().unwrap();
+            let timelines = self.timelines.lock().await;
            self.create_timeline_uninit_mark(timeline_id, &timelines)?
        };
        // create a `tenant/{tenant_id}/timelines/basebackup-{timeline_id}.{TEMP_FILE_SUFFIX}/`
@@ -2543,8 +2593,10 @@ impl Tenant {
        // Initialize the timeline without loading the layer map, because we already updated the layer
        // map above, when we imported the datadir.
        let timeline = {
-            let mut timelines = self.timelines.lock().unwrap();
-            raw_timeline.initialize_with_lock(ctx, &mut timelines, false)?
+            let mut timelines = self.timelines.lock().await;
+            raw_timeline
+                .initialize_with_lock(ctx, &mut timelines, false)
+                .await?
        };

        info!(
@@ -2646,7 +2698,7 @@ impl Tenant {
    fn create_timeline_uninit_mark(
        &self,
        timeline_id: TimelineId,
-        timelines: &MutexGuard<HashMap<TimelineId, Arc<Timeline>>>,
+        timelines: &tokio::sync::MutexGuard<HashMap<TimelineId, Arc<Timeline>>>,
    ) -> anyhow::Result<TimelineUninitMark> {
        let tenant_id = self.tenant_id;

@@ -3190,7 +3242,7 @@ pub mod harness {
                .instrument(info_span!("try_load", tenant_id=%self.tenant_id))
                .await?;
            tenant.state.send_replace(TenantState::Active);
-            for timeline in tenant.timelines.lock().unwrap().values() {
+            for timeline in tenant.timelines.lock().await.values() {
                timeline.set_state(TimelineState::Active);
            }
            Ok(tenant)
@@ -3250,7 +3302,9 @@ mod tests {
    #[tokio::test]
    async fn test_basic() -> anyhow::Result<()> {
        let (tenant, ctx) = TenantHarness::create("test_basic")?.load().await;
-        let tline = tenant.create_test_timeline(TIMELINE_ID, Lsn(0), DEFAULT_PG_VERSION, &ctx)?;
+        let tline = tenant
+            .create_test_timeline(TIMELINE_ID, Lsn(0), DEFAULT_PG_VERSION, &ctx)
+            .await?;

        let writer = tline.writer();
        writer.put(*TEST_KEY, Lsn(0x10), &Value::Image(TEST_IMG("foo at 0x10")))?;
@@ -3283,9 +3337,14 @@ mod tests {
        let (tenant, ctx) = TenantHarness::create("no_duplicate_timelines")?
            .load()
            .await;
-        let _ = tenant.create_test_timeline(TIMELINE_ID, Lsn(0), DEFAULT_PG_VERSION, &ctx)?;
+        let _ = tenant
+            .create_test_timeline(TIMELINE_ID, Lsn(0), DEFAULT_PG_VERSION, &ctx)
+            .await?;

-        match tenant.create_empty_timeline(TIMELINE_ID, Lsn(0), DEFAULT_PG_VERSION, &ctx) {
+        match tenant
+            .create_empty_timeline(TIMELINE_ID, Lsn(0), DEFAULT_PG_VERSION, &ctx)
+            .await
+        {
            Ok(_) => panic!("duplicate timeline creation should fail"),
            Err(e) => assert_eq!(
                e.to_string(),
@@ -3314,7 +3373,9 @@ mod tests {
        use std::str::from_utf8;

        let (tenant, ctx) = TenantHarness::create("test_branch")?.load().await;
-        let tline = tenant.create_test_timeline(TIMELINE_ID, Lsn(0), DEFAULT_PG_VERSION, &ctx)?;
+        let tline = tenant
+            .create_test_timeline(TIMELINE_ID, Lsn(0), DEFAULT_PG_VERSION, &ctx)
+            .await?;
        let writer = tline.writer();

        #[allow(non_snake_case)]
@@ -3340,6 +3401,7 @@ mod tests {
            .await?;
        let newtline = tenant
            .get_timeline(NEW_TIMELINE_ID, true)
+            .await
            .expect("Should have a local timeline");
        let new_writer = newtline.writer();
        new_writer.put(TEST_KEY_A, Lsn(0x40), &test_value("bar at 0x40"))?;
@@ -3411,7 +3473,9 @@ mod tests {
            TenantHarness::create("test_prohibit_branch_creation_on_garbage_collected_data")?
                .load()
                .await;
-        let tline = tenant.create_test_timeline(TIMELINE_ID, Lsn(0), DEFAULT_PG_VERSION, &ctx)?;
+        let tline = tenant
+            .create_test_timeline(TIMELINE_ID, Lsn(0), DEFAULT_PG_VERSION, &ctx)
+            .await?;
        make_some_layers(tline.as_ref(), Lsn(0x20)).await?;

        // this removes layers before lsn 40 (50 minus 10), so there are two remaining layers, image and delta for 31-50
@@ -3448,8 +3512,9 @@ mod tests {
                .load()
                .await;

-        let tline =
-            tenant.create_test_timeline(TIMELINE_ID, Lsn(0x50), DEFAULT_PG_VERSION, &ctx)?;
+        let tline = tenant
+            .create_test_timeline(TIMELINE_ID, Lsn(0x50), DEFAULT_PG_VERSION, &ctx)
+            .await?;
        // try to branch at lsn 0x25, should fail because initdb lsn is 0x50
        match tenant
            .branch_timeline_test(&tline, NEW_TIMELINE_ID, Some(Lsn(0x25)), &ctx)
@@ -3478,7 +3543,7 @@ mod tests {
            RepoHarness::create("test_prohibit_get_for_garbage_collected_data")?
            .load();

-        let tline = repo.create_empty_timeline(TIMELINE_ID, Lsn(0), DEFAULT_PG_VERSION)?;
+        let tline = repo.create_empty_timeline(TIMELINE_ID, Lsn(0), DEFAULT_PG_VERSION).await?;
        make_some_layers(tline.as_ref(), Lsn(0x20)).await?;

        repo.gc_iteration(Some(TIMELINE_ID), 0x10, Duration::ZERO)?;
@@ -3498,7 +3563,9 @@ mod tests {
            TenantHarness::create("test_get_branchpoints_from_an_inactive_timeline")?
                .load()
                .await;
-        let tline = tenant.create_test_timeline(TIMELINE_ID, Lsn(0), DEFAULT_PG_VERSION, &ctx)?;
+        let tline = tenant
+            .create_test_timeline(TIMELINE_ID, Lsn(0), DEFAULT_PG_VERSION, &ctx)
+            .await?;
        make_some_layers(tline.as_ref(), Lsn(0x20)).await?;

        tenant
@@ -3506,6 +3573,7 @@ mod tests {
            .await?;
        let newtline = tenant
            .get_timeline(NEW_TIMELINE_ID, true)
+            .await
            .expect("Should have a local timeline");

        make_some_layers(newtline.as_ref(), Lsn(0x60)).await?;
@@ -3546,7 +3614,9 @@ mod tests {
            TenantHarness::create("test_retain_data_in_parent_which_is_needed_for_child")?
                .load()
                .await;
-        let tline = tenant.create_test_timeline(TIMELINE_ID, Lsn(0), DEFAULT_PG_VERSION, &ctx)?;
+        let tline = tenant
+            .create_test_timeline(TIMELINE_ID, Lsn(0), DEFAULT_PG_VERSION, &ctx)
+            .await?;
        make_some_layers(tline.as_ref(), Lsn(0x20)).await?;

        tenant
@@ -3554,6 +3624,7 @@ mod tests {
            .await?;
        let newtline = tenant
            .get_timeline(NEW_TIMELINE_ID, true)
+            .await
            .expect("Should have a local timeline");
        // this removes layers before lsn 40 (50 minus 10), so there are two remaining layers, image and delta for 31-50
        tenant
@@ -3569,7 +3640,9 @@ mod tests {
            TenantHarness::create("test_parent_keeps_data_forever_after_branching")?
                .load()
                .await;
-        let tline = tenant.create_test_timeline(TIMELINE_ID, Lsn(0), DEFAULT_PG_VERSION, &ctx)?;
+        let tline = tenant
+            .create_test_timeline(TIMELINE_ID, Lsn(0), DEFAULT_PG_VERSION, &ctx)
+            .await?;
        make_some_layers(tline.as_ref(), Lsn(0x20)).await?;

        tenant
@@ -3577,6 +3650,7 @@ mod tests {
            .await?;
        let newtline = tenant
            .get_timeline(NEW_TIMELINE_ID, true)
+            .await
            .expect("Should have a local timeline");

        make_some_layers(newtline.as_ref(), Lsn(0x60)).await?;
@@ -3601,14 +3675,16 @@ mod tests {
        let harness = TenantHarness::create(TEST_NAME)?;
        {
            let (tenant, ctx) = harness.load().await;
-            let tline =
-                tenant.create_test_timeline(TIMELINE_ID, Lsn(0x8000), DEFAULT_PG_VERSION, &ctx)?;
+            let tline = tenant
+                .create_test_timeline(TIMELINE_ID, Lsn(0x8000), DEFAULT_PG_VERSION, &ctx)
+                .await?;
            make_some_layers(tline.as_ref(), Lsn(0x8000)).await?;
        }

        let (tenant, _ctx) = harness.load().await;
        tenant
            .get_timeline(TIMELINE_ID, true)
+            .await
            .expect("cannot load timeline");

        Ok(())
@@ -3621,8 +3697,9 @@ mod tests {
        // create two timelines
        {
            let (tenant, ctx) = harness.load().await;
-            let tline =
-                tenant.create_test_timeline(TIMELINE_ID, Lsn(0), DEFAULT_PG_VERSION, &ctx)?;
+            let tline = tenant
+                .create_test_timeline(TIMELINE_ID, Lsn(0), DEFAULT_PG_VERSION, &ctx)
+                .await?;

            make_some_layers(tline.as_ref(), Lsn(0x20)).await?;

@@ -3633,6 +3710,7 @@ mod tests {

            let newtline = tenant
                .get_timeline(NEW_TIMELINE_ID, true)
+                .await
                .expect("Should have a local timeline");

            make_some_layers(newtline.as_ref(), Lsn(0x60)).await?;
@@ -3644,10 +3722,12 @@ mod tests {
        // check that both, child and ancestor are loaded
        let _child_tline = tenant
            .get_timeline(NEW_TIMELINE_ID, true)
+            .await
            .expect("cannot get child timeline loaded");

        let _ancestor_tline = tenant
            .get_timeline(TIMELINE_ID, true)
+            .await
            .expect("cannot get ancestor timeline loaded");

        Ok(())
@@ -3659,7 +3739,9 @@ mod tests {
        let harness = TenantHarness::create(TEST_NAME)?;
        let (tenant, ctx) = harness.load().await;

-        let tline = tenant.create_test_timeline(TIMELINE_ID, Lsn(0), DEFAULT_PG_VERSION, &ctx)?;
+        let tline = tenant
+            .create_test_timeline(TIMELINE_ID, Lsn(0), DEFAULT_PG_VERSION, &ctx)
+            .await?;
        drop(tline);
        drop(tenant);

@@ -3697,7 +3779,9 @@ mod tests {
    #[tokio::test]
    async fn test_images() -> anyhow::Result<()> {
        let (tenant, ctx) = TenantHarness::create("test_images")?.load().await;
-        let tline = tenant.create_test_timeline(TIMELINE_ID, Lsn(0), DEFAULT_PG_VERSION, &ctx)?;
+        let tline = tenant
+            .create_test_timeline(TIMELINE_ID, Lsn(0), DEFAULT_PG_VERSION, &ctx)
+            .await?;

        let writer = tline.writer();
        writer.put(*TEST_KEY, Lsn(0x10), &Value::Image(TEST_IMG("foo at 0x10")))?;
@@ -3762,7 +3846,9 @@ mod tests {
    #[tokio::test]
    async fn test_bulk_insert() -> anyhow::Result<()> {
        let (tenant, ctx) = TenantHarness::create("test_bulk_insert")?.load().await;
-        let tline = tenant.create_test_timeline(TIMELINE_ID, Lsn(0), DEFAULT_PG_VERSION, &ctx)?;
+        let tline = tenant
+            .create_test_timeline(TIMELINE_ID, Lsn(0), DEFAULT_PG_VERSION, &ctx)
+            .await?;

        let mut lsn = Lsn(0x10);

@@ -3804,7 +3890,9 @@ mod tests {
    #[tokio::test]
    async fn test_random_updates() -> anyhow::Result<()> {
        let (tenant, ctx) = TenantHarness::create("test_random_updates")?.load().await;
-        let tline = tenant.create_test_timeline(TIMELINE_ID, Lsn(0), DEFAULT_PG_VERSION, &ctx)?;
+        let tline = tenant
+            .create_test_timeline(TIMELINE_ID, Lsn(0), DEFAULT_PG_VERSION, &ctx)
+            .await?;

        const NUM_KEYS: usize = 1000;

@@ -3877,8 +3965,9 @@ mod tests {
        let (tenant, ctx) = TenantHarness::create("test_traverse_branches")?
            .load()
            .await;
-        let mut tline =
-            tenant.create_test_timeline(TIMELINE_ID, Lsn(0), DEFAULT_PG_VERSION, &ctx)?;
+        let mut tline = tenant
+            .create_test_timeline(TIMELINE_ID, Lsn(0), DEFAULT_PG_VERSION, &ctx)
+            .await?;

        const NUM_KEYS: usize = 1000;

@@ -3915,6 +4004,7 @@ mod tests {
                .await?;
            tline = tenant
                .get_timeline(new_tline_id, true)
+                .await
                .expect("Should have the branched timeline");

            for _ in 0..NUM_KEYS {
@@ -3960,8 +4050,9 @@ mod tests {
        let (tenant, ctx) = TenantHarness::create("test_traverse_ancestors")?
            .load()
            .await;
-        let mut tline =
-            tenant.create_test_timeline(TIMELINE_ID, Lsn(0), DEFAULT_PG_VERSION, &ctx)?;
+        let mut tline = tenant
+            .create_test_timeline(TIMELINE_ID, Lsn(0), DEFAULT_PG_VERSION, &ctx)
+            .await?;

        const NUM_KEYS: usize = 100;
        const NUM_TLINES: usize = 50;
@@ -3980,6 +4071,7 @@ mod tests {
                .await?;
            tline = tenant
                .get_timeline(new_tline_id, true)
+                .await
                .expect("Should have the branched timeline");

            for _ in 0..NUM_KEYS {
--- a/pageserver/src/tenant/mgr.rs
+++ b/pageserver/src/tenant/mgr.rs
@@ -58,9 +58,10 @@ static TENANTS: Lazy<RwLock<TenantsMap>> = Lazy::new(|| RwLock::new(TenantsMap::
 /// Initialize repositories with locally available timelines.
 /// Timelines that are only partially available locally (remote storage has more data than this pageserver)
 /// are scheduled for download and added to the tenant once download is completed.
-#[instrument(skip(conf, remote_storage))]
+#[instrument(skip_all)]
 pub async fn init_tenant_mgr(
    conf: &'static PageServerConf,
+    broker_client: storage_broker::BrokerClientChannel,
    remote_storage: Option<GenericRemoteStorage>,
 ) -> anyhow::Result<()> {
    // Scan local filesystem for attached tenants
@@ -116,6 +117,7 @@ pub async fn init_tenant_mgr(
                    match schedule_local_tenant_processing(
                        conf,
                        &tenant_dir_path,
+                        broker_client.clone(),
                        remote_storage.clone(),
                        &ctx,
                    ) {
@@ -150,6 +152,7 @@ pub async fn init_tenant_mgr(
 pub fn schedule_local_tenant_processing(
    conf: &'static PageServerConf,
    tenant_path: &Path,
+    broker_client: storage_broker::BrokerClientChannel,
    remote_storage: Option<GenericRemoteStorage>,
    ctx: &RequestContext,
 ) -> anyhow::Result<Arc<Tenant>> {
@@ -186,7 +189,7 @@ pub fn schedule_local_tenant_processing(
    let tenant = if conf.tenant_attaching_mark_file_path(&tenant_id).exists() {
        info!("tenant {tenant_id} has attaching mark file, resuming its attach operation");
        if let Some(remote_storage) = remote_storage {
-            match Tenant::spawn_attach(conf, tenant_id, remote_storage, ctx) {
+            match Tenant::spawn_attach(conf, tenant_id, broker_client, remote_storage, ctx) {
                Ok(tenant) => tenant,
                Err(e) => {
                    error!("Failed to spawn_attach tenant {tenant_id}, reason: {e:#}");
@@ -204,7 +207,7 @@ pub fn schedule_local_tenant_processing(
    } else {
        info!("tenant {tenant_id} is assumed to be loadable, starting load operation");
        // Start loading the tenant into memory. It will initially be in Loading state.
-        Tenant::spawn_load(conf, tenant_id, remote_storage, ctx)
+        Tenant::spawn_load(conf, tenant_id, broker_client, remote_storage, ctx)
    };
    Ok(tenant)
 }
@@ -245,7 +248,7 @@ pub async fn shutdown_all_tenants() {
    for (_, tenant) in tenants_to_shut_down {
        if tenant.is_active() {
            // updates tenant state, forbidding new GC and compaction iterations from starting
-            tenant.set_stopping();
+            tenant.set_stopping().await;
            tenants_to_freeze_and_flush.push(tenant);
        }
    }
@@ -275,6 +278,7 @@ pub async fn create_tenant(
    conf: &'static PageServerConf,
    tenant_conf: TenantConfOpt,
    tenant_id: TenantId,
+    broker_client: storage_broker::BrokerClientChannel,
    remote_storage: Option<GenericRemoteStorage>,
    ctx: &RequestContext,
 ) -> Result<Arc<Tenant>, TenantMapInsertError> {
@@ -287,7 +291,7 @@ pub async fn create_tenant(
        //       See https://github.com/neondatabase/neon/issues/4233

        let created_tenant =
-            schedule_local_tenant_processing(conf, &tenant_directory, remote_storage, ctx)?;
+            schedule_local_tenant_processing(conf, &tenant_directory, broker_client, remote_storage, ctx)?;
        // TODO: tenant object & its background loops remain, untracked in tenant map, if we fail here.
        //      See https://github.com/neondatabase/neon/issues/4233

@@ -315,7 +319,7 @@ pub async fn set_new_tenant_config(
        new_tenant_conf,
        false,
    )?;
-    tenant.set_new_tenant_config(new_tenant_conf);
+    tenant.set_new_tenant_config(new_tenant_conf).await;
    Ok(())
 }

@@ -404,6 +408,7 @@ pub async fn detach_tenant(
 pub async fn load_tenant(
    conf: &'static PageServerConf,
    tenant_id: TenantId,
+    broker_client: storage_broker::BrokerClientChannel,
    remote_storage: Option<GenericRemoteStorage>,
    ctx: &RequestContext,
 ) -> Result<(), TenantMapInsertError> {
@@ -415,7 +420,7 @@ pub async fn load_tenant(
                .with_context(|| format!("Failed to remove tenant ignore mark {tenant_ignore_mark:?} during tenant loading"))?;
        }

-        let new_tenant = schedule_local_tenant_processing(conf, &tenant_path, remote_storage, ctx)
+        let new_tenant = schedule_local_tenant_processing(conf, &tenant_path, broker_client, remote_storage, ctx)
            .with_context(|| {
                format!("Failed to schedule tenant processing in path {tenant_path:?}")
            })?;
@@ -472,6 +477,7 @@ pub async fn attach_tenant(
    conf: &'static PageServerConf,
    tenant_id: TenantId,
    tenant_conf: TenantConfOpt,
+    broker_client: storage_broker::BrokerClientChannel,
    remote_storage: GenericRemoteStorage,
    ctx: &RequestContext,
 ) -> Result<(), TenantMapInsertError> {
@@ -487,7 +493,7 @@ pub async fn attach_tenant(
            .context("check for attach marker file existence")?;
        anyhow::ensure!(marker_file_exists, "create_tenant_files should have created the attach marker file");

-        let attached_tenant = schedule_local_tenant_processing(conf, &tenant_dir, Some(remote_storage), ctx)?;
+        let attached_tenant = schedule_local_tenant_processing(conf, &tenant_dir, broker_client, Some(remote_storage), ctx)?;
        // TODO: tenant object & its background loops remain, untracked in tenant map, if we fail here.
        //      See https://github.com/neondatabase/neon/issues/4233

@@ -569,8 +575,9 @@ where
            Some(tenant) => match tenant.current_state() {
                TenantState::Attaching
                | TenantState::Loading
+                | TenantState::Activating
                | TenantState::Broken { .. }
-                | TenantState::Active => tenant.set_stopping(),
+                | TenantState::Active => tenant.set_stopping().await,
                TenantState::Stopping => return Err(TenantStateError::IsStopping(tenant_id)),
            },
            None => return Err(TenantStateError::NotFound(tenant_id)),
@@ -597,7 +604,7 @@ where
            let tenants_accessor = TENANTS.read().await;
            match tenants_accessor.get(&tenant_id) {
                Some(tenant) => {
-                    tenant.set_broken(e.to_string());
+                    tenant.set_broken(e.to_string()).await;
                }
                None => {
                    warn!("Tenant {tenant_id} got removed from memory");
@@ -679,6 +686,7 @@ pub async fn immediate_compact(

    let timeline = tenant
        .get_timeline(timeline_id, true)
+        .await
        .map_err(ApiError::NotFound)?;

    // Run in task_mgr to avoid race with tenant_detach operation
--- a/pageserver/src/tenant/remote_timeline_client.rs
+++ b/pageserver/src/tenant/remote_timeline_client.rs
@@ -1264,7 +1264,12 @@ mod tests {
            let harness = TenantHarness::create(test_name)?;
            let (tenant, ctx) = runtime.block_on(harness.load());
            // create an empty timeline directory
-            let _ = tenant.create_test_timeline(TIMELINE_ID, Lsn(0), DEFAULT_PG_VERSION, &ctx)?;
+            let _ = runtime.block_on(tenant.create_test_timeline(
+                TIMELINE_ID,
+                Lsn(0),
+                DEFAULT_PG_VERSION,
+                &ctx,
+            ))?;

            let remote_fs_dir = harness.conf.workdir.join("remote_fs");
            std::fs::create_dir_all(remote_fs_dir)?;
--- a/pageserver/src/tenant/size.rs
+++ b/pageserver/src/tenant/size.rs
@@ -136,7 +136,7 @@ pub(super) async fn gather_inputs(
        .context("Failed to refresh gc_info before gathering inputs")?;

    // Collect information about all the timelines
-    let mut timelines = tenant.list_timelines();
+    let mut timelines = tenant.list_timelines().await;

    if timelines.is_empty() {
        // perhaps the tenant has just been created, and as such doesn't have any data yet
--- a/pageserver/src/tenant/timeline.rs
+++ b/pageserver/src/tenant/timeline.rs
@@ -31,7 +31,6 @@ use std::sync::atomic::{AtomicI64, Ordering as AtomicOrdering};
 use std::sync::{Arc, Mutex, MutexGuard, RwLock, Weak};
 use std::time::{Duration, Instant, SystemTime};

-use crate::broker_client::{get_broker_client, is_broker_client_initialized};
 use crate::context::{DownloadBehavior, RequestContext};
 use crate::tenant::remote_timeline_client::{self, index::LayerFileMetadata};
 use crate::tenant::storage_layer::{
@@ -227,7 +226,7 @@ pub struct Timeline {
    /// or None if WAL receiver has not received anything for this timeline
    /// yet.
    pub last_received_wal: Mutex<Option<WalReceiverInfo>>,
-    pub walreceiver: WalReceiver,
+    pub walreceiver: Mutex<Option<WalReceiver>>,

    /// Relation size cache
    pub rel_size_cache: RwLock<HashMap<RelTag, (Lsn, BlockNumber)>>,
@@ -622,17 +621,27 @@ impl Timeline {
            .await
        {
            Ok(()) => Ok(()),
-            seqwait_error => {
+            Err(e) => {
+                // walreceiver.status() locks internally, don't count that towards the wait_lsn_time_histo
                drop(_timer);
-                let walreceiver_status = self.walreceiver.status().await;
-                seqwait_error.with_context(|| format!(
-                    "Timed out while waiting for WAL record at LSN {} to arrive, last_record_lsn {} disk consistent LSN={}, {}",
-                    lsn,
-                    self.get_last_record_lsn(),
-                    self.get_disk_consistent_lsn(),
-                    walreceiver_status.map(|status| status.to_human_readable_string())
-                            .unwrap_or_else(|| "WalReceiver status: Not active".to_string()),
-                ))
+                let walreceiver_status = {
+                    match &*self.walreceiver.lock().unwrap() {
+                        None => "stopping or stopped".to_string(),
+                        Some(walreceiver) => match walreceiver.status() {
+                            Some(status) => status.to_human_readable_string(),
+                            None => "Not active".to_string(),
+                        },
+                    }
+                };
+                Err(anyhow::Error::new(e).context({
+                    format!(
+                        "Timed out while waiting for WAL record at LSN {} to arrive, last_record_lsn {} disk consistent LSN={}, WalReceiver status: {}",
+                        lsn,
+                        self.get_last_record_lsn(),
+                        self.get_disk_consistent_lsn(),
+                        walreceiver_status,
+                    )
+                }))
            }
        }
    }
@@ -907,18 +916,10 @@ impl Timeline {
        Ok(())
    }

-    pub fn activate(self: &Arc<Self>, ctx: &RequestContext) -> anyhow::Result<()> {
-        if is_broker_client_initialized() {
-            self.launch_wal_receiver(ctx, get_broker_client().clone())?;
-        } else if cfg!(test) {
-            info!("not launching WAL receiver because broker client hasn't been initialized");
-        } else {
-            anyhow::bail!("broker client not initialized");
-        }
-
+    pub fn activate(self: &Arc<Self>, broker_client: BrokerClientChannel, ctx: &RequestContext) {
+        self.launch_wal_receiver(ctx, broker_client);
        self.set_state(TimelineState::Active);
        self.launch_eviction_task();
-        Ok(())
    }

    pub fn set_state(&self, new_state: TimelineState) {
@@ -1327,15 +1328,7 @@ impl Timeline {
        let (layer_flush_done_tx, _) = tokio::sync::watch::channel((0, Ok(())));

        let tenant_conf_guard = tenant_conf.read().unwrap();
-        let wal_connect_timeout = tenant_conf_guard
-            .walreceiver_connect_timeout
-            .unwrap_or(conf.default_tenant_conf.walreceiver_connect_timeout);
-        let lagging_wal_timeout = tenant_conf_guard
-            .lagging_wal_timeout
-            .unwrap_or(conf.default_tenant_conf.lagging_wal_timeout);
-        let max_lsn_wal_lag = tenant_conf_guard
-            .max_lsn_wal_lag
-            .unwrap_or(conf.default_tenant_conf.max_lsn_wal_lag);
+
        let evictions_low_residence_duration_metric_threshold =
            Self::get_evictions_low_residence_duration_metric_threshold(
                &tenant_conf_guard,
@@ -1344,18 +1337,6 @@ impl Timeline {
        drop(tenant_conf_guard);

        Arc::new_cyclic(|myself| {
-            let walreceiver = WalReceiver::new(
-                TenantTimelineId::new(tenant_id, timeline_id),
-                Weak::clone(myself),
-                WalReceiverConf {
-                    wal_connect_timeout,
-                    lagging_wal_timeout,
-                    max_lsn_wal_lag,
-                    auth_token: crate::config::SAFEKEEPER_AUTH_TOKEN.get().cloned(),
-                    availability_zone: conf.availability_zone.clone(),
-                },
-            );
-
            let mut result = Timeline {
                conf,
                tenant_conf,
@@ -1367,7 +1348,7 @@ impl Timeline {
                wanted_image_layers: Mutex::new(None),

                walredo_mgr,
-                walreceiver,
+                walreceiver: Mutex::new(None),

                remote_client: remote_client.map(Arc::new),

@@ -1487,17 +1468,49 @@ impl Timeline {
        *flush_loop_state = FlushLoopState::Running;
    }

-    pub(super) fn launch_wal_receiver(
-        &self,
+    /// Creates and starts the wal receiver.
+    ///
+    /// This function is expected to be called at most once per Timeline's lifecycle
+    /// when the timeline is activated.
+    fn launch_wal_receiver(
+        self: &Arc<Self>,
        ctx: &RequestContext,
        broker_client: BrokerClientChannel,
-    ) -> anyhow::Result<()> {
+    ) {
        info!(
            "launching WAL receiver for timeline {} of tenant {}",
            self.timeline_id, self.tenant_id
        );
-        self.walreceiver.start(ctx, broker_client)?;
-        Ok(())
+
+        let tenant_conf_guard = self.tenant_conf.read().unwrap();
+        let wal_connect_timeout = tenant_conf_guard
+            .walreceiver_connect_timeout
+            .unwrap_or(self.conf.default_tenant_conf.walreceiver_connect_timeout);
+        let lagging_wal_timeout = tenant_conf_guard
+            .lagging_wal_timeout
+            .unwrap_or(self.conf.default_tenant_conf.lagging_wal_timeout);
+        let max_lsn_wal_lag = tenant_conf_guard
+            .max_lsn_wal_lag
+            .unwrap_or(self.conf.default_tenant_conf.max_lsn_wal_lag);
+        drop(tenant_conf_guard);
+
+        let mut guard = self.walreceiver.lock().unwrap();
+        assert!(
+            guard.is_none(),
+            "multiple launches / re-launches of WAL receiver are not supported"
+        );
+        *guard = Some(WalReceiver::start(
+            Arc::clone(self),
+            WalReceiverConf {
+                wal_connect_timeout,
+                lagging_wal_timeout,
+                max_lsn_wal_lag,
+                auth_token: crate::config::SAFEKEEPER_AUTH_TOKEN.get().cloned(),
+                availability_zone: self.conf.availability_zone.clone(),
+            },
+            broker_client,
+            ctx,
+        ));
    }

    ///
--- a/pageserver/src/tenant/timeline/walreceiver.rs
+++ b/pageserver/src/tenant/timeline/walreceiver.rs
@@ -29,16 +29,14 @@ use crate::tenant::timeline::walreceiver::connection_manager::{
    connection_manager_loop_step, ConnectionManagerState,
 };

-use anyhow::Context;
 use std::future::Future;
 use std::num::NonZeroU64;
 use std::ops::ControlFlow;
-use std::sync::atomic::{self, AtomicBool};
-use std::sync::{Arc, Weak};
+use std::sync::Arc;
 use std::time::Duration;
 use storage_broker::BrokerClientChannel;
 use tokio::select;
-use tokio::sync::{watch, RwLock};
+use tokio::sync::watch;
 use tokio_util::sync::CancellationToken;
 use tracing::*;

@@ -62,46 +60,23 @@ pub struct WalReceiverConf {

 pub struct WalReceiver {
    timeline: TenantTimelineId,
-    timeline_ref: Weak<Timeline>,
-    conf: WalReceiverConf,
-    started: AtomicBool,
-    manager_status: Arc<RwLock<Option<ConnectionManagerStatus>>>,
+    manager_status: Arc<std::sync::RwLock<Option<ConnectionManagerStatus>>>,
 }

 impl WalReceiver {
-    pub fn new(
-        timeline: TenantTimelineId,
-        timeline_ref: Weak<Timeline>,
-        conf: WalReceiverConf,
-    ) -> Self {
-        Self {
-            timeline,
-            timeline_ref,
-            conf,
-            started: AtomicBool::new(false),
-            manager_status: Arc::new(RwLock::new(None)),
-        }
-    }
-
    pub fn start(
-        &self,
-        ctx: &RequestContext,
+        timeline: Arc<Timeline>,
+        conf: WalReceiverConf,
        mut broker_client: BrokerClientChannel,
-    ) -> anyhow::Result<()> {
-        if self.started.load(atomic::Ordering::Acquire) {
-            anyhow::bail!("Wal receiver is already started");
-        }
-
-        let timeline = self.timeline_ref.upgrade().with_context(|| {
-            format!("walreceiver start on a dropped timeline {}", self.timeline)
-        })?;
-
+        ctx: &RequestContext,
+    ) -> Self {
        let tenant_id = timeline.tenant_id;
        let timeline_id = timeline.timeline_id;
        let walreceiver_ctx =
            ctx.detached_child(TaskKind::WalReceiverManager, DownloadBehavior::Error);
-        let wal_receiver_conf = self.conf.clone();
-        let loop_status = Arc::clone(&self.manager_status);
+
+        let loop_status = Arc::new(std::sync::RwLock::new(None));
+        let manager_status = Arc::clone(&loop_status);
        task_mgr::spawn(
            WALRECEIVER_RUNTIME.handle(),
            TaskKind::WalReceiverManager,
@@ -113,7 +88,7 @@ impl WalReceiver {
                info!("WAL receiver manager started, connecting to broker");
                let mut connection_manager_state = ConnectionManagerState::new(
                    timeline,
-                    wal_receiver_conf,
+                    conf,
                );
                loop {
                    select! {
@@ -137,29 +112,29 @@ impl WalReceiver {
                }

                connection_manager_state.shutdown().await;
-                *loop_status.write().await = None;
+                *loop_status.write().unwrap() = None;
                Ok(())
            }
            .instrument(info_span!(parent: None, "wal_connection_manager", tenant = %tenant_id, timeline = %timeline_id))
        );

-        self.started.store(true, atomic::Ordering::Release);
-
-        Ok(())
+        Self {
+            timeline: TenantTimelineId::new(tenant_id, timeline_id),
+            manager_status,
+        }
    }

-    pub async fn stop(&self) {
+    pub async fn stop(self) {
        task_mgr::shutdown_tasks(
            Some(TaskKind::WalReceiverManager),
            Some(self.timeline.tenant_id),
            Some(self.timeline.timeline_id),
        )
        .await;
-        self.started.store(false, atomic::Ordering::Release);
    }

-    pub(super) async fn status(&self) -> Option<ConnectionManagerStatus> {
-        self.manager_status.read().await.clone()
+    pub(super) fn status(&self) -> Option<ConnectionManagerStatus> {
+        self.manager_status.read().unwrap().clone()
    }
 }

--- a/pageserver/src/tenant/timeline/walreceiver/connection_manager.rs
+++ b/pageserver/src/tenant/timeline/walreceiver/connection_manager.rs
@@ -29,7 +29,6 @@ use storage_broker::proto::TenantTimelineId as ProtoTenantTimelineId;
 use storage_broker::BrokerClientChannel;
 use storage_broker::Streaming;
 use tokio::select;
-use tokio::sync::RwLock;
 use tracing::*;

 use crate::{exponential_backoff, DEFAULT_BASE_BACKOFF_SECONDS, DEFAULT_MAX_BACKOFF_SECONDS};
@@ -48,7 +47,7 @@ pub(super) async fn connection_manager_loop_step(
    broker_client: &mut BrokerClientChannel,
    connection_manager_state: &mut ConnectionManagerState,
    ctx: &RequestContext,
-    manager_status: &RwLock<Option<ConnectionManagerStatus>>,
+    manager_status: &std::sync::RwLock<Option<ConnectionManagerStatus>>,
 ) -> ControlFlow<(), ()> {
    match connection_manager_state
        .timeline
@@ -195,7 +194,7 @@ pub(super) async fn connection_manager_loop_step(
                .change_connection(new_candidate, ctx)
                .await
        }
-        *manager_status.write().await = Some(connection_manager_state.manager_status());
+        *manager_status.write().unwrap() = Some(connection_manager_state.manager_status());
    }
 }

@@ -1310,6 +1309,7 @@ mod tests {
        let (tenant, ctx) = harness.load().await;
        let timeline = tenant
            .create_test_timeline(TIMELINE_ID, Lsn(0), crate::DEFAULT_PG_VERSION, &ctx)
+            .await
            .expect("Failed to create an empty timeline for dummy wal connection manager");

        ConnectionManagerState {
--- a/pageserver/src/walingest.rs
+++ b/pageserver/src/walingest.rs
@@ -1209,7 +1209,7 @@ mod tests {
    #[tokio::test]
    async fn test_relsize() -> Result<()> {
        let (tenant, ctx) = TenantHarness::create("test_relsize")?.load().await;
-        let tline = create_test_timeline(&tenant, TIMELINE_ID, DEFAULT_PG_VERSION, &ctx)?;
+        let tline = create_test_timeline(&tenant, TIMELINE_ID, DEFAULT_PG_VERSION, &ctx).await?;
        let mut walingest = init_walingest_test(&tline, &ctx).await?;

        let mut m = tline.begin_modification(Lsn(0x20));
@@ -1428,7 +1428,7 @@ mod tests {
    #[tokio::test]
    async fn test_drop_extend() -> Result<()> {
        let (tenant, ctx) = TenantHarness::create("test_drop_extend")?.load().await;
-        let tline = create_test_timeline(&tenant, TIMELINE_ID, DEFAULT_PG_VERSION, &ctx)?;
+        let tline = create_test_timeline(&tenant, TIMELINE_ID, DEFAULT_PG_VERSION, &ctx).await?;
        let mut walingest = init_walingest_test(&tline, &ctx).await?;

        let mut m = tline.begin_modification(Lsn(0x20));
@@ -1497,7 +1497,7 @@ mod tests {
    #[tokio::test]
    async fn test_truncate_extend() -> Result<()> {
        let (tenant, ctx) = TenantHarness::create("test_truncate_extend")?.load().await;
-        let tline = create_test_timeline(&tenant, TIMELINE_ID, DEFAULT_PG_VERSION, &ctx)?;
+        let tline = create_test_timeline(&tenant, TIMELINE_ID, DEFAULT_PG_VERSION, &ctx).await?;
        let mut walingest = init_walingest_test(&tline, &ctx).await?;

        // Create a 20 MB relation (the size is arbitrary)
@@ -1637,7 +1637,7 @@ mod tests {
    #[tokio::test]
    async fn test_large_rel() -> Result<()> {
        let (tenant, ctx) = TenantHarness::create("test_large_rel")?.load().await;
-        let tline = create_test_timeline(&tenant, TIMELINE_ID, DEFAULT_PG_VERSION, &ctx)?;
+        let tline = create_test_timeline(&tenant, TIMELINE_ID, DEFAULT_PG_VERSION, &ctx).await?;
        let mut walingest = init_walingest_test(&tline, &ctx).await?;

        let mut lsn = 0x10;
Author	SHA1	Message	Date
Christian Schwarz	1ebe92bcf9	make Tenant::timelines a tokio::sync::RwLock This is preliminary work for/from #4220 (async `Layer::get_value_reconstruct_data`). The patch converts `Tenant::timelines` from `std::sync::Mutex` to `tokio::sync::Mutex`. We need this change because we want to switch `Timeline::layers` to an async RwLock. We need that because we hold `Timeline::layers` while calling `Layer::get_value_reconstruct_data`. So, if we want to make get_value_reconstruct_data async, we need to make `Timeline::layers` async first.	2023-05-24 17:53:12 +02:00
Christian Schwarz	413598b19b	fix merge fallout (?)	2023-05-24 17:42:51 +02:00
Christian Schwarz	b345f32e3f	Merge branch 'problame/infallible-timeline-activate/4-make-infallible' into problame/async-timeline-get/dont-hold-timelines-lock-inside-tenant-state-send-modify	2023-05-24 17:25:35 +02:00
Christian Schwarz	69cfa9fe61	launch_wal_receiver: apply joonas's review suggestion (visibility + doc comment)	2023-05-24 17:20:03 +02:00
Christian Schwarz	2c424c8f4e	Revert "activate_timelines counter is now == not_broken_timelines.len()" not_broken_timelines is an iterator, doesn't have `len()`. This reverts commit `4001f441c0`.	2023-05-24 17:19:22 +02:00
Christian Schwarz	4001f441c0	activate_timelines counter is now == not_broken_timelines.len()	2023-05-24 17:14:49 +02:00
Christian Schwarz	ef956c47fc	make it clear that `walreceiver_status` is always used in the branch where it's produced	2023-05-24 17:12:35 +02:00
Christian Schwarz	8606b6abe5	Merge remote-tracking branch 'origin/problame/infallible-timeline-activate/3-funnel-storage-broker-client' into problame/infallible-timeline-activate/4-make-infallible	2023-05-24 17:02:18 +02:00
Christian Schwarz	732f60317b	Merge remote-tracking branch 'origin/main' into problame/infallible-timeline-activate/3-funnel-storage-broker-client	2023-05-24 16:58:25 +02:00
Christian Schwarz	b54431bbd3	pass the BrokerClientChannel by value & clone it as necessary It's a wrapper around an inner Arc anyways Also, this gets rid of the OnceCell	2023-05-24 12:29:05 +02:00
Christian Schwarz	def5eb8542	Merge branch 'problame/infallible-timeline-activate/2-pushup-tenant-and-timeline-activation' into problame/infallible-timeline-activate/3-funnel-storage-broker-client	2023-05-24 11:57:37 +02:00
Christian Schwarz	07da786ed3	apply joonas's suggestion to use parent: None + follows_from	2023-05-24 11:56:26 +02:00
Christian Schwarz	75c3c43b2e	don't unwrap() the `activate()` result in spawn_load / spawn_attach	2023-05-24 11:36:07 +02:00
Christian Schwarz	bdf03eab58	Merge branch 'problame/infallible-timeline-activate/2-pushup-tenant-and-timeline-activation' into problame/infallible-timeline-activate/3-funnel-storage-broker-client	2023-05-24 11:32:38 +02:00
Christian Schwarz	32c85fa87a	Merge remote-tracking branch 'origin/main' into problame/infallible-timeline-activate/2-pushup-tenant-and-timeline-activation	2023-05-24 11:31:00 +02:00
Christian Schwarz	b2e0c58a8c	Merge branch 'problame/infallible-timeline-activate/4-make-infallible' into problame/async-timeline-get/dont-hold-timelines-lock-inside-tenant-state-send-modify	2023-05-23 20:44:34 +02:00
Christian Schwarz	94f30f0660	Merge branch 'problame/infallible-timeline-activate/3-funnel-storage-broker-client' into problame/infallible-timeline-activate/4-make-infallible	2023-05-23 20:44:12 +02:00
Christian Schwarz	a55d224923	tests would fail because broker client needs to be launched on a tokio runtime thread	2023-05-23 20:43:10 +02:00
Christian Schwarz	4f586ac101	Merge branch 'problame/infallible-timeline-activate/2-pushup-tenant-and-timeline-activation' into problame/infallible-timeline-activate/3-funnel-storage-broker-client	2023-05-23 20:42:54 +02:00
Christian Schwarz	feb2e80b83	tests were failing because activate() was outside of a span with tenant_id	2023-05-23 20:36:32 +02:00
Christian Schwarz	ee22e81583	don't hold timelines lock inside set_stopping()	2023-05-23 20:11:15 +02:00
Christian Schwarz	3e604eaa39	refactor: introduce TenantState::Activating to avoid holding timelines lock inside Tenant::activate	2023-05-23 20:03:12 +02:00
Christian Schwarz	8bcb542a3b	refactor: make timeline activation infallible Timeline::activate() was only fallible because `launch_wal_receiver` was. `launch_wal_receiver` was fallible only because of some preliminary checks in `WalReceiver::start`. Turns out these checks can be shifted to the type system by delaying creatinon of the `WalReceiver` struct to the point where we activate the timeline. The changes in this PR were enabled by my previous refactoring that funneled the broker_client from pageserver startup to the activate() call sites.	2023-05-23 19:27:06 +02:00
Christian Schwarz	17b081d294	refactor: eliminate global storage_broker client state (This is prep work to make `Timeline::activate` infallible.) This patch removes the global storage_broker client instance from the pageserver codebase. Instead, pageserver startup instantiates it and passes it down to the `Timeline::activate` function, which in turn passes it to the WalReceiver, which is the entity that actually uses it.	2023-05-23 19:27:06 +02:00
Christian Schwarz	d5337e6a65	refactor responsibility for tenant/timeline activation (This is prep work to make `Timeline::activate()` infallible.) The current possibility for failure in `Timeline::activate()` is the broker client's presence / absence. It should be an assert, but we're careful with these. So, I'm planning to pass in the broker client to activate(), thereby eliminating the possiblity of its absence. In the unit tests, we don't have a broker client. So, I thought I'd be in trouble because the unit tests also called `activate()` before this PR. However, closer inspection reveals a long-standing FIXME about this, which is addressed by this patch. It turns out that the unit tests don't actually need the background loops to be running. They just need the state value to be `Active`. So, for the tests, we just set it to that value but don't spawn the background loops. We'll need to revisit this if we ever do more Rust unit tests in the future. But right now, this refactoring improves the code, so, let's revisit when we get there.	2023-05-23 19:26:36 +02:00
Christian Schwarz	cc96a5186d	tenant_map_insert: don't expose the vacant entry to the closure This tightens up the API a little. Byproduct of some refactoring work that I'm doing right now.	2023-05-23 19:25:47 +02:00