pageserver: warn on unexpected refs to Tenant after shutdown

This should help to validate that our shutdown logic is working
the way we expect.

pageserver/src/tenant.rs

@@ -2076,74 +2076,57 @@ impl Tenant {
     ) -> Tenant {
         let (state, mut rx) = watch::channel(state);
 
-        task_mgr::spawn(
-            task_mgr::BACKGROUND_RUNTIME.handle(),
-            TaskKind::MetricsCollection,
-            Some(tenant_id),
-            None,
-            &format!("state metrics collector for tenant {tenant_id}"),
-            false,
-            async move {
-                let cancel = task_mgr::shutdown_token();
+        tokio::spawn(async move {
+            let tid = tenant_id.to_string();
 
-                let tid = tenant_id.to_string();
+            fn inspect_state(state: &TenantState) -> ([&'static str; 1], bool) {
+                ([state.into()], matches!(state, TenantState::Broken { .. }))
+            }
 
-                fn inspect_state(state: &TenantState) -> ([&'static str; 1], bool) {
-                    ([state.into()], matches!(state, TenantState::Broken { .. }))
+            let mut tuple = inspect_state(&rx.borrow_and_update());
+
+            let is_broken = tuple.1;
+            let mut counted_broken = if !is_broken {
+                // the tenant might be ignored and reloaded, so first remove any previous set
+                // element. it most likely has already been scraped, as these are manual operations
+                // right now. most likely we will add it back very soon.
+                drop(crate::metrics::BROKEN_TENANTS_SET.remove_label_values(&[&tid]));
+                false
+            } else {
+                // add the id to the set right away, there should not be any updates on the channel
+                // after
+                crate::metrics::BROKEN_TENANTS_SET
+                    .with_label_values(&[&tid])
+                    .set(1);
+                true
+            };
+
+            loop {
+                let labels = &tuple.0;
+                let current = TENANT_STATE_METRIC.with_label_values(labels);
+                current.inc();
+
+                if rx.changed().await.is_err() {
+                    // tenant has been dropped; decrement the counter because a tenant with that
+                    // state is no longer in tenant map, but allow any broken set item to exist
+                    // still.
+                    current.dec();
+                    break;
                 }
 
-                let mut tuple = inspect_state(&rx.borrow_and_update());
+                current.dec();
+                tuple = inspect_state(&rx.borrow_and_update());
 
                 let is_broken = tuple.1;
-                let mut counted_broken = if !is_broken {
-                    // the tenant might be ignored and reloaded, so first remove any previous set
-                    // element. it most likely has already been scraped, as these are manual operations
-                    // right now. most likely we will add it back very soon.
-                    drop(crate::metrics::BROKEN_TENANTS_SET.remove_label_values(&[&tid]));
-                    false
-                } else {
-                    // add the id to the set right away, there should not be any updates on the channel
-                    // after
+                if is_broken && !counted_broken {
+                    counted_broken = true;
+                    // insert the tenant_id (back) into the set
                     crate::metrics::BROKEN_TENANTS_SET
                         .with_label_values(&[&tid])
-                        .set(1);
-                    true
-                };
-
-                loop {
-                    let labels = &tuple.0;
-                    let current = TENANT_STATE_METRIC.with_label_values(labels);
-                    current.inc();
-
-                    let changed = tokio::select! {
-                        changed = rx.changed() => {changed},
-                        _ = cancel.cancelled() => {return Ok(())}
-                    };
-
-                    if changed.is_err() {
-                        // tenant has been dropped; decrement the counter because a tenant with that
-                        // state is no longer in tenant map, but allow any broken set item to exist
-                        // still.
-                        current.dec();
-                        break;
-                    }
-
-                    current.dec();
-                    tuple = inspect_state(&rx.borrow_and_update());
-
-                    let is_broken = tuple.1;
-                    if is_broken && !counted_broken {
-                        counted_broken = true;
-                        // insert the tenant_id (back) into the set
-                        crate::metrics::BROKEN_TENANTS_SET
-                            .with_label_values(&[&tid])
-                            .inc();
-                    }
+                        .inc();
                 }
-                Ok(())
             }
-            .instrument(info_span!("state_metrics", tenant_id = %tenant_id)),
-        );
+        });
 
         Tenant {
             tenant_id,

pageserver/src/tenant/mgr.rs

@@ -274,12 +274,15 @@ async fn shutdown_all_tenants0(tenants: &tokio::sync::RwLock<TenantsMap>) {
                 let ordering = std::sync::atomic::Ordering::Relaxed;
                 let joined_other = std::sync::atomic::AtomicBool::new(false);
 
+                let tenant_weak = Arc::downgrade(&tenant);
                 let mut shutdown = std::pin::pin!(async {
                     let freeze_and_flush = true;
 
                     let res = {
                         let (_guard, shutdown_progress) = completion::channel();
-                        tenant.shutdown(shutdown_progress, freeze_and_flush).await
+                        // Always safe to upgrade this weak ptr, because the outer code block
+                        // holds an Arc to it past the point it awaits this async block.
+                        tenant_weak.upgrade().unwrap().shutdown(shutdown_progress, freeze_and_flush).await
                     };
 
                     if let Err(other_progress) = res {
@@ -304,6 +307,13 @@ async fn shutdown_all_tenants0(tenants: &tokio::sync::RwLock<TenantsMap>) {
                     }
                 };
 
+                // Shutdown should have stopped all the background tasks that referenced
+                // the tenant: when we drop our ref it should drop the Tenant.  If that isn't
+                // the case it's a bug.
+                if Arc::into_inner(tenant).is_none() {
+                    warn!("Tenant still has references after shutdown");
+                }
+
                 debug!("tenant successfully stopped");
             }
             .instrument(info_span!("shutdown", %tenant_id)),