[Bug]: Cascading priority deadlock in Safekeeper via `acquire_term()` returning long-lived RwLock ReadGuard

// safekeeper/src/timeline.rs:1058-1069
impl WalResidentTimeline {
    /// Ensure that current term is t, erroring otherwise, and lock the state.
    pub async fn acquire_term(&self, t: Term) -> Result<ReadGuardSharedState> {
        let ss = self.read_shared_state().await;  // Acquires RwLock::read()
        if ss.sk.state().acceptor_state.term != t {
            bail!(
                "failed to acquire term {}, current term {}",
                t,
                ss.sk.state().acceptor_state.term
            );
        }
        Ok(ss)  //  Returns the ReadGuard to the caller — lifetime uncontrolled!
    }
}

// safekeeper/src/timeline.rs:746-748
pub async fn read_shared_state(&self) -> ReadGuardSharedState {
    self.mutex.read().await  // tokio::sync::RwLock<SharedState>
}

// safekeeper/src/timeline.rs:1110-1136
impl WalResidentTimeline {
    pub async fn process_msg(
        &self,
        msg: &ProposerAcceptorMessage,
    ) -> Result<Option<AcceptorProposerMessage>> {
        let mut shared_state = self.write_shared_state().await;  //  Needs WRITE lock
        // ... WAL acceptance processing
        shared_state.sk.safekeeper().process_msg(msg).await?;
    }
}

// safekeeper/src/timeline.rs:742-743
pub async fn write_shared_state(self: &Arc<Self>) -> WriteGuardSharedState<'_> {
    WriteGuardSharedState::new(self.clone(), self.mutex.write().await)
}

// safekeeper/src/timeline.rs:467
pub struct Timeline {
    // ...
    /// Safekeeper and other state, that should remain consistent and
    /// synchronized with the disk. This is tokio mutex as we write WAL to disk
    /// while holding it, ensuring that consensus checks are in order.
    mutex: RwLock<SharedState>,
    // ...
}

use std::sync::Arc;
use tokio::sync::RwLock;
use tokio::time::{sleep, Duration, Instant};

/// Simulates the Safekeeper's shared state locking pattern:
/// - Task A: WalSender::run() holding acquire_term() ReadGuard across WAL send I/O
/// - Task B: process_msg() requesting write lock for WAL acceptance
/// - Task C: get_safekeeper_info() / metrics collection requesting read lock
#[tokio::main]
async fn main() {
    let shared_state = Arc::new(RwLock::new("term=5, flush_lsn=0/1234"));

    // Task A: WalSender — holds ReadGuard across WAL sending (network I/O)
    let state_a = shared_state.clone();
    tokio::spawn(async move {
        println!("[WalSender] acquire_term() — requesting read lock...");
        let guard = state_a.read().await;
        println!("[WalSender] Read lock acquired (term verified: {})", *guard);
        println!("[WalSender] Sending WAL to pageserver (slow network, 5s)...");
        // In production: WAL streaming over TCP to pageserver
        // If pageserver is slow to acknowledge, this can take seconds to minutes
        sleep(Duration::from_secs(5)).await;
        println!("[WalSender] WAL sent. Releasing read lock.");
        drop(guard);
    });

    // Ensure WalSender gets the lock first
    sleep(Duration::from_millis(100)).await;

    // Task B: WAL acceptance (process_msg) — needs write lock
    let state_b = shared_state.clone();
    tokio::spawn(async move {
        println!("[process_msg] Compute proposing WAL! Requesting write lock...");
        let _guard = state_b.write().await;
        println!("[process_msg] Write lock acquired. WAL accepted.");
    });

    // Ensure write request is queued
    sleep(Duration::from_millis(100)).await;

    // Task C: Metrics / broker info — needs read lock
    let state_c = shared_state.clone();
    let start_c = Instant::now();
    let handle_c = tokio::spawn(async move {
        println!("[metrics] Collecting safekeeper info — requesting read lock...");
        let _guard = state_c.read().await; //  Blocked by pending writer!
        println!(
            "[metrics] Read lock acquired after {:?} — metrics were stalled!",
            start_c.elapsed()
        );
    });

    // Task D spawned CONCURRENTLY with Task C (not sequentially after it)
    sleep(Duration::from_millis(50)).await;

    // Task D: Another WalSender starting — also blocked
    let state_d = shared_state.clone();
    let start_d = Instant::now();
    let handle_d = tokio::spawn(async move {
        println!("[WalSender2] New connection! acquire_term() — requesting read lock...");
        let _guard = state_d.read().await; //  Also blocked by pending writer!
        println!(
            "[WalSender2] Read lock acquired after {:?} — new connection stalled!",
            start_d.elapsed()
        );
    });

    // Wait for both to complete
    handle_c.await.unwrap();
    handle_d.await.unwrap();
}

[WalSender] acquire_term() — requesting read lock...
[WalSender] Read lock acquired (term verified: term=5, flush_lsn=0/1234)
[WalSender] Sending WAL to pageserver (slow network, 5s)...
[process_msg] Compute proposing WAL! Requesting write lock...
[metrics] Collecting safekeeper info — requesting read lock...
[WalSender2] New connection! acquire_term() — requesting read lock...
[WalSender] WAL sent. Releasing read lock.
[process_msg] Write lock acquired. WAL accepted.
[WalSender2] Read lock acquired after 4.748686056s — new connection stalled!
[metrics] Read lock acquired after 4.799372098s — metrics were stalled!