hypercall/runtime/

standby.rs

use std::sync::Arc;
use tokio::sync::{mpsc, oneshot, Mutex};
use tracing::info;

use crate::rsm::unified_engine::UnifiedEngineRequest;

/// Controls the engine standby → draining → active lifecycle.
///
/// In standby mode, orders are queued in memory. On `promote()`, the current
/// queue is handed off for draining and the controller enters a draining phase;
/// orders arriving during that phase continue to queue until `finish_drain()`
/// transitions the controller to active mode, after which `try_enqueue()`
/// returns `NotStandby` and callers forward directly to the engine.
#[derive(Clone)]
pub struct StandbyController {
    state: Arc<Mutex<StandbyState>>,
}

enum StandbyState {
    Standby {
        promote_tx: Option<oneshot::Sender<()>>,
        queue: Vec<UnifiedEngineRequest>,
        capacity: usize,
    },
    Draining {
        queue: Vec<UnifiedEngineRequest>,
        capacity: usize,
    },
    Active,
}

/// Result of trying to enqueue an order during standby.
pub enum EnqueueResult {
    /// Order was queued (standby mode). Will be drained on promote.
    Queued,
    /// Not in standby mode — caller should send directly to engine.
    NotStandby(UnifiedEngineRequest),
    /// Queue is full. Caller should return 503.
    QueueFull(UnifiedEngineRequest),
}

/// Result of a promote attempt.
#[derive(Debug, PartialEq)]
pub enum PromoteResult {
    /// Successfully promoted. Returns number of queued orders to drain.
    Promoted { queued_orders: usize },
    /// Already active — promote was already called.
    AlreadyActive,
}

impl StandbyController {
    /// Create a new controller in standby mode.
    /// Returns the controller and a future that resolves when promote is called.
    pub fn new_standby(capacity: usize) -> (Self, oneshot::Receiver<()>) {
        let (promote_tx, promote_rx) = oneshot::channel();
        let controller = Self {
            state: Arc::new(Mutex::new(StandbyState::Standby {
                promote_tx: Some(promote_tx),
                queue: Vec::with_capacity(capacity.min(1024)),
                capacity,
            })),
        };
        (controller, promote_rx)
    }

    /// Create a controller that starts in active mode (no standby).
    pub fn new_active() -> Self {
        Self {
            state: Arc::new(Mutex::new(StandbyState::Active)),
        }
    }

    /// Returns true if the controller is currently in standby or draining mode
    /// (i.e., not yet fully active — orders are still being queued).
    pub async fn is_standby(&self) -> bool {
        !matches!(*self.state.lock().await, StandbyState::Active)
    }

    /// Returns true if promote() has been called (in Draining or Active state).
    pub async fn is_promoted(&self) -> bool {
        !matches!(*self.state.lock().await, StandbyState::Standby { .. })
    }

    /// Try to enqueue an order. Returns the appropriate result for each state:
    /// - `Queued`: order was accepted into the standby queue
    /// - `NotStandby(request)`: not in standby mode, caller should forward to engine directly
    /// - `QueueFull(request)`: queue at capacity, caller should return 503
    pub async fn try_enqueue(&self, request: UnifiedEngineRequest) -> EnqueueResult {
        let mut state = self.state.lock().await;
        match &mut *state {
            StandbyState::Active => EnqueueResult::NotStandby(request),
            StandbyState::Standby {
                queue, capacity, ..
            }
            | StandbyState::Draining { queue, capacity } => {
                if queue.len() >= *capacity {
                    EnqueueResult::QueueFull(request)
                } else {
                    queue.push(request);
                    EnqueueResult::Queued
                }
            }
        }
    }

    /// Promote from standby to draining.
    /// Sends the promote signal so any waiting tasks (engine, batcher, etc.) unblock.
    ///
    /// Orders remain in the controller — call `take_drain_batch()` to retrieve
    /// them for draining. New orders arriving during drain are still queued
    /// (not forwarded directly) to preserve ordering.
    pub async fn promote(&self) -> PromoteResult {
        let mut state = self.state.lock().await;
        match &*state {
            StandbyState::Standby { .. } => {
                let capacity = match &*state {
                    StandbyState::Standby { capacity, .. } => *capacity,
                    _ => unreachable!(),
                };
                match std::mem::replace(
                    &mut *state,
                    StandbyState::Draining {
                        queue: Vec::new(),
                        capacity,
                    },
                ) {
                    StandbyState::Standby {
                        promote_tx,
                        queue,
                        capacity,
                        ..
                    } => {
                        let count = queue.len();
                        // Move existing orders into the new Draining queue
                        *state = StandbyState::Draining { queue, capacity };
                        if let Some(tx) = promote_tx {
                            let _ = tx.send(());
                        }
                        info!(queued_orders = count, "Promoted from standby to draining");
                        PromoteResult::Promoted {
                            queued_orders: count,
                        }
                    }
                    _ => unreachable!(),
                }
            }
            StandbyState::Draining { .. } | StandbyState::Active => PromoteResult::AlreadyActive,
        }
    }

    /// Take a batch of orders to drain into the engine.
    /// Returns queued orders and keeps the controller in Draining state.
    /// When the returned vec is empty, transitions to Active.
    /// Call this in a loop until it returns an empty vec.
    pub async fn take_drain_batch(&self) -> Vec<UnifiedEngineRequest> {
        let mut state = self.state.lock().await;
        match &mut *state {
            StandbyState::Draining { queue, .. } => {
                if queue.is_empty() {
                    *state = StandbyState::Active;
                    info!("Drain complete, transitioned to active");
                    Vec::new()
                } else {
                    let orders = std::mem::take(queue);
                    info!(batch_size = orders.len(), "Taking drain batch");
                    orders
                }
            }
            _ => Vec::new(),
        }
    }

    /// Drain queued orders into the engine channel.
    /// Returns (sent_count, unsent_orders) — unsent orders are returned on channel failure
    /// so callers can retry or persist them.
    pub async fn drain_to(
        orders: Vec<UnifiedEngineRequest>,
        engine_tx: &mpsc::Sender<UnifiedEngineRequest>,
    ) -> (usize, Vec<UnifiedEngineRequest>) {
        let total = orders.len();
        let mut sent = 0;
        let mut iter = orders.into_iter();
        for order in iter.by_ref() {
            match engine_tx.send(order).await {
                Ok(()) => sent += 1,
                Err(mpsc::error::SendError(failed_order)) => {
                    let remaining = total - sent - 1;
                    tracing::error!(
                        sent = sent,
                        remaining = remaining + 1,
                        "Engine channel closed during queue drain"
                    );
                    let mut unsent = vec![failed_order];
                    unsent.extend(iter);
                    return (sent, unsent);
                }
            }
        }
        (sent, Vec::new())
    }
}

#[cfg(test)]
mod tests {
    use super::*;
    use rust_decimal::Decimal;
    use tokio::sync::mpsc as tokio_mpsc;

    fn make_test_request() -> (
        UnifiedEngineRequest,
        mpsc::Receiver<hypercall_types::OrderUpdateMessage>,
    ) {
        use hypercall_types::{OrderAction, OrderActionMessage};
        use hypercall_types::{OrderInfo, Side, TimeInForce, WalletAddress};

        let (response_tx, response_rx) = tokio_mpsc::channel(1);
        let request = UnifiedEngineRequest {
            message: OrderActionMessage {
                timestamp: 0,
                info: OrderInfo {
                    symbol: "BTC-20260501-100000-C".to_string(),
                    price: Decimal::ZERO,
                    size: Decimal::ONE,
                    side: Side::Buy,
                    tif: TimeInForce::GTC,
                    client_id: None,
                    order_id: None,
                    is_perp: false,
                    underlying: None,
                    reduce_only: None,
                    nonce: None,
                    signature: None,
                    mmp_enabled: false,
                    builder_code_address: None,
                },
                action: OrderAction::CancelOrder,
                wallet: WalletAddress::default(),
                api_wallet_address: None,
                mmp_triggered: false,
                request_id: None,
            },
            response_tx,
            enqueued_at: std::time::Instant::now(),
            #[cfg(feature = "otel-tracing")]
            trace_context: None,
        };
        (request, response_rx)
    }

    #[tokio::test]
    async fn active_mode_does_not_enqueue() {
        let controller = StandbyController::new_active();
        let (req, _rx) = make_test_request();

        let result = controller.try_enqueue(req).await;
        assert!(
            matches!(result, EnqueueResult::NotStandby(_)),
            "Active controller should return NotStandby"
        );
        assert!(!controller.is_standby().await);
    }

    #[tokio::test]
    async fn standby_mode_queues_orders() {
        let (controller, _promote_rx) = StandbyController::new_standby(100);
        assert!(controller.is_standby().await);

        let (req, _rx) = make_test_request();
        let result = controller.try_enqueue(req).await;
        assert!(matches!(result, EnqueueResult::Queued));
    }

    #[tokio::test]
    async fn standby_rejects_when_full() {
        let (controller, _promote_rx) = StandbyController::new_standby(2);

        let (req1, _rx1) = make_test_request();
        let (req2, _rx2) = make_test_request();
        let (req3, _rx3) = make_test_request();

        assert!(matches!(
            controller.try_enqueue(req1).await,
            EnqueueResult::Queued
        ));
        assert!(matches!(
            controller.try_enqueue(req2).await,
            EnqueueResult::Queued
        ));
        assert!(
            matches!(
                controller.try_enqueue(req3).await,
                EnqueueResult::QueueFull(_)
            ),
            "Should reject when at capacity"
        );
    }

    #[tokio::test]
    async fn promote_drains_queued_orders() {
        let (controller, _promote_rx) = StandbyController::new_standby(100);

        let (req1, _rx1) = make_test_request();
        let (req2, _rx2) = make_test_request();
        controller.try_enqueue(req1).await;
        controller.try_enqueue(req2).await;

        let result = controller.promote().await;
        let orders = controller.take_drain_batch().await;
        assert_eq!(result, PromoteResult::Promoted { queued_orders: 2 });
        assert_eq!(orders.len(), 2);
        // Still in draining state until finish_drain
        assert!(controller.is_standby().await);

        let remaining = controller.take_drain_batch().await;
        assert!(remaining.is_empty());
        assert!(!controller.is_standby().await);
    }

    #[tokio::test]
    async fn promote_sends_signal() {
        let (controller, promote_rx) = StandbyController::new_standby(100);

        let result = controller.promote().await;
        assert_eq!(result, PromoteResult::Promoted { queued_orders: 0 });

        // The receiver should have gotten the signal
        assert!(promote_rx.await.is_ok());
    }

    #[tokio::test]
    async fn double_promote_returns_already_active() {
        let (controller, _promote_rx) = StandbyController::new_standby(100);

        let result1 = controller.promote().await;
        assert_eq!(result1, PromoteResult::Promoted { queued_orders: 0 });

        // Second promote while draining returns AlreadyActive
        let result2 = controller.promote().await;
        assert_eq!(result2, PromoteResult::AlreadyActive);

        // Also after finish_drain
        controller.take_drain_batch().await;
        let result3 = controller.promote().await;
        assert_eq!(result3, PromoteResult::AlreadyActive);
    }

    #[tokio::test]
    async fn during_drain_orders_still_queue() {
        let (controller, _promote_rx) = StandbyController::new_standby(100);
        controller.promote().await;

        // During drain phase, orders still queue (preserving ordering)
        let (req, _rx) = make_test_request();
        let result = controller.try_enqueue(req).await;
        assert!(
            matches!(result, EnqueueResult::Queued),
            "Should still queue during drain"
        );

        // After finish_drain, new orders get NotStandby
        let remaining = controller.take_drain_batch().await;
        assert_eq!(remaining.len(), 1);
        // Drain the additional order
        let remaining2 = controller.take_drain_batch().await;
        assert!(remaining2.is_empty());
        assert!(!controller.is_standby().await);

        let (req2, _rx2) = make_test_request();
        let result2 = controller.try_enqueue(req2).await;
        assert!(
            matches!(result2, EnqueueResult::NotStandby(_)),
            "Should return NotStandby after drain complete"
        );
    }

    #[tokio::test]
    async fn drain_to_sends_all_orders() {
        let (controller, _promote_rx) = StandbyController::new_standby(100);

        let (req1, _rx1) = make_test_request();
        let (req2, _rx2) = make_test_request();
        let (req3, _rx3) = make_test_request();
        controller.try_enqueue(req1).await;
        controller.try_enqueue(req2).await;
        controller.try_enqueue(req3).await;

        controller.promote().await;
        let orders = controller.take_drain_batch().await;

        let (engine_tx, mut engine_rx) = tokio_mpsc::channel(100);
        let (sent, unsent) = StandbyController::drain_to(orders, &engine_tx).await;
        assert_eq!(sent, 3);
        assert!(unsent.is_empty());

        // Verify orders arrived
        let mut received = 0;
        while engine_rx.try_recv().is_ok() {
            received += 1;
        }
        assert_eq!(received, 3);
    }

    #[tokio::test]
    async fn drain_to_handles_closed_channel() {
        let (controller, _promote_rx) = StandbyController::new_standby(100);

        let (req1, _rx1) = make_test_request();
        let (req2, _rx2) = make_test_request();
        controller.try_enqueue(req1).await;
        controller.try_enqueue(req2).await;

        controller.promote().await;
        let orders = controller.take_drain_batch().await;

        let (engine_tx, engine_rx) = tokio_mpsc::channel(100);
        drop(engine_rx);
        let (sent, unsent) = StandbyController::drain_to(orders, &engine_tx).await;
        // Should handle gracefully — sends fail but doesn't panic
        assert_eq!(sent, 0);
        // All orders preserved (failed order + remaining)
        assert_eq!(unsent.len(), 2);
    }

    #[tokio::test]
    async fn concurrent_enqueue_is_safe() {
        let (controller, _promote_rx) = StandbyController::new_standby(1000);
        let mut handles = Vec::new();

        for _ in 0..100 {
            let c = controller.clone();
            handles.push(tokio::spawn(async move {
                let (req, _rx) = make_test_request();
                matches!(c.try_enqueue(req).await, EnqueueResult::Queued)
            }));
        }

        let mut queued = 0;
        for h in handles {
            if h.await.unwrap() {
                queued += 1;
            }
        }
        assert_eq!(queued, 100);

        let result = controller.promote().await;
        let orders = controller.take_drain_batch().await;
        assert_eq!(result, PromoteResult::Promoted { queued_orders: 100 });
        assert_eq!(orders.len(), 100);
    }

    #[tokio::test]
    async fn concurrent_enqueue_respects_capacity() {
        let (controller, _promote_rx) = StandbyController::new_standby(50);
        let mut handles = Vec::new();

        for _ in 0..100 {
            let c = controller.clone();
            handles.push(tokio::spawn(async move {
                let (req, _rx) = make_test_request();
                matches!(c.try_enqueue(req).await, EnqueueResult::Queued)
            }));
        }

        let mut queued = 0;
        for h in handles {
            if h.await.unwrap() {
                queued += 1;
            }
        }
        assert_eq!(queued, 50);
    }

    #[tokio::test]
    async fn promote_during_concurrent_enqueue() {
        let (controller, _promote_rx) = StandbyController::new_standby(1000);

        // Start enqueueing
        let c1 = controller.clone();
        let enqueue_handle = tokio::spawn(async move {
            let mut count = 0;
            for _ in 0..50 {
                let (req, _rx) = make_test_request();
                if matches!(c1.try_enqueue(req).await, EnqueueResult::Queued) {
                    count += 1;
                }
                tokio::task::yield_now().await;
            }
            count
        });

        // Promote mid-stream
        tokio::task::yield_now().await;
        let result = controller.promote().await;
        let orders = controller.take_drain_batch().await;

        let enqueued_after = enqueue_handle.await.unwrap();
        match result {
            PromoteResult::Promoted { queued_orders } => {
                // All 50 orders should have been accepted: some before promote
                // (in standby queue returned as `orders`), rest during drain
                // (in drain queue, retrievable via finish_drain).
                // enqueued_after counts ALL successful enqueues from the spawned task,
                // which includes both pre-promote and during-drain enqueues.
                assert!(queued_orders <= 50);
                assert!(enqueued_after <= 50);
                assert_eq!(orders.len(), queued_orders);
            }
            _ => panic!("Should have promoted"),
        }
    }

    #[tokio::test]
    async fn drain_preserves_order() {
        let (controller, _promote_rx) = StandbyController::new_standby(100);

        // Enqueue orders with distinguishable request_ids
        for i in 0..5 {
            let (mut req, _rx) = make_test_request();
            req.message.request_id = Some(format!("order-{}", i));
            controller.try_enqueue(req).await;
        }

        controller.promote().await;
        let orders = controller.take_drain_batch().await;
        let ids: Vec<String> = orders
            .iter()
            .map(|o| o.message.request_id.clone().unwrap())
            .collect();
        assert_eq!(
            ids,
            vec!["order-0", "order-1", "order-2", "order-3", "order-4"]
        );
    }

    #[tokio::test]
    async fn drain_to_preserves_fifo_into_engine() {
        let (controller, _promote_rx) = StandbyController::new_standby(100);

        for i in 0..3 {
            let (mut req, _rx) = make_test_request();
            req.message.request_id = Some(format!("req-{}", i));
            controller.try_enqueue(req).await;
        }

        controller.promote().await;
        let orders = controller.take_drain_batch().await;
        let (engine_tx, mut engine_rx) = tokio_mpsc::channel(100);
        let (sent, unsent) = StandbyController::drain_to(orders, &engine_tx).await;
        assert_eq!(sent, 3);
        assert!(unsent.is_empty());

        let mut received_ids = Vec::new();
        while let Ok(req) = engine_rx.try_recv() {
            received_ids.push(req.message.request_id.clone().unwrap());
        }
        assert_eq!(received_ids, vec!["req-0", "req-1", "req-2"]);
    }

    #[tokio::test]
    async fn response_channels_survive_queue() {
        let (controller, _promote_rx) = StandbyController::new_standby(100);

        let (req, mut response_rx) = make_test_request();
        let _response_tx = req.response_tx.clone();
        controller.try_enqueue(req).await;

        controller.promote().await;
        let orders = controller.take_drain_batch().await;
        let (engine_tx, mut engine_rx) = tokio_mpsc::channel(100);
        StandbyController::drain_to(orders, &engine_tx).await;

        // Simulate engine processing: receive order and send response
        let received = engine_rx.try_recv().unwrap();
        let response = hypercall_types::OrderUpdateMessage {
            order_id: Some(42),
            info: received.message.info.clone(),
            status: hypercall_types::OrderUpdateStatus::Canceled,
            timestamp: 1,
            reason: None,
            filled_size: Decimal::ZERO,
            wallet_address: hypercall_types::WalletAddress::default(),
            mmp_triggered: false,
            request_id: None,
        };
        received.response_tx.send(response).await.unwrap();

        // The original handler's response_rx should receive it
        let resp = response_rx.recv().await.unwrap();
        assert_eq!(resp.order_id, Some(42));
    }

    #[tokio::test]
    async fn zero_capacity_rejects_everything() {
        let (controller, _promote_rx) = StandbyController::new_standby(0);

        let (req, _rx) = make_test_request();
        assert!(matches!(
            controller.try_enqueue(req).await,
            EnqueueResult::QueueFull(_)
        ));

        // But promote still works
        let result = controller.promote().await;
        let orders = controller.take_drain_batch().await;
        assert_eq!(result, PromoteResult::Promoted { queued_orders: 0 });
        assert!(orders.is_empty());
    }

    #[tokio::test]
    async fn controller_dropped_closes_response_channels() {
        // Simulates crash during standby: controller is dropped,
        // HTTP handlers waiting on response_rx should get a clean error
        let (req, mut response_rx) = make_test_request();

        {
            let (controller, _promote_rx) = StandbyController::new_standby(100);
            controller.try_enqueue(req).await;
            // controller + queued orders dropped here
        }

        // The response_tx inside the queued request is now dropped
        // HTTP handler should see None (channel closed), not hang forever
        let result =
            tokio::time::timeout(std::time::Duration::from_millis(100), response_rx.recv()).await;
        assert!(
            result.is_ok(),
            "recv should return immediately, not timeout"
        );
        assert!(
            result.unwrap().is_none(),
            "Should get None (channel closed)"
        );
    }

    #[tokio::test]
    async fn promote_signal_dropped_without_sending() {
        // Simulates: controller is dropped without calling promote().
        // The engine task waiting on promote_rx should get an error, not hang.
        let promote_rx;
        {
            let (controller, rx) = StandbyController::new_standby(100);
            promote_rx = rx;
            // controller dropped without promote()
            drop(controller);
        }

        let result = tokio::time::timeout(std::time::Duration::from_millis(100), promote_rx).await;
        assert!(result.is_ok(), "promote_rx should resolve, not hang");
        assert!(
            result.unwrap().is_err(),
            "Should get RecvError (sender dropped)"
        );
    }

    #[tokio::test]
    async fn partial_drain_does_not_corrupt_remaining() {
        // Engine channel has capacity 1. We drain 3 orders.
        // First goes through, channel blocks, we verify partial drain is clean.
        let (controller, _promote_rx) = StandbyController::new_standby(100);

        for i in 0..3 {
            let (mut req, _rx) = make_test_request();
            req.message.request_id = Some(format!("order-{}", i));
            controller.try_enqueue(req).await;
        }

        controller.promote().await;
        let orders = controller.take_drain_batch().await;

        // Channel capacity 1: first send succeeds, second blocks
        let (engine_tx, mut engine_rx) = tokio_mpsc::channel(1);

        // Drain with a timeout to simulate partial drain before crash
        let drain_handle =
            tokio::spawn(async move { StandbyController::drain_to(orders, &engine_tx).await });

        // Read one order to make room, then drop the receiver (simulate crash)
        let first = engine_rx.recv().await.unwrap();
        assert_eq!(first.message.request_id.as_deref(), Some("order-0"));
        drop(engine_rx);

        let (sent, unsent) =
            tokio::time::timeout(std::time::Duration::from_millis(500), drain_handle)
                .await
                .expect("drain should finish, not hang")
                .expect("drain task should not panic");

        // Some orders sent before channel closed — at least 1 (order-0),
        // possibly 2 (order-1 was buffered in the channel capacity)
        assert!((1..=3).contains(&sent));
        // All orders accounted for: sent + unsent = total
        assert_eq!(sent + unsent.len(), 3);
    }

    #[tokio::test]
    async fn queued_order_response_tx_remains_valid_through_promote() {
        // End-to-end: enqueue → promote → drain → engine processes → HTTP handler gets response
        // This is the golden path for the standby queue during deploy switchover
        let (controller, promote_rx) = StandbyController::new_standby(100);

        // HTTP handler enqueues 3 orders and holds response receivers
        let mut response_rxs = Vec::new();
        for i in 0..3 {
            let (mut req, rx) = make_test_request();
            req.message.request_id = Some(format!("deploy-order-{}", i));
            controller.try_enqueue(req).await;
            response_rxs.push(rx);
        }

        // Simulate engine waiting for promote, then starting
        let (engine_tx, mut engine_rx) = tokio_mpsc::channel::<UnifiedEngineRequest>(100);
        let engine_handle = tokio::spawn(async move {
            // Wait for promote signal
            promote_rx.await.ok();

            // Process all orders
            let mut processed = 0;
            while let Some(req) = engine_rx.recv().await {
                let resp: hypercall_types::OrderUpdateMessage =
                    hypercall_types::OrderUpdateMessage {
                        order_id: Some(100 + processed as u64),
                        info: req.message.info.clone(),
                        status: hypercall_types::OrderUpdateStatus::Open,
                        timestamp: 1,
                        reason: None,
                        filled_size: Decimal::ZERO,
                        wallet_address: hypercall_types::WalletAddress::default(),
                        mmp_triggered: false,
                        request_id: req.message.request_id.clone(),
                    };
                req.response_tx.send(resp).await.ok();
                processed += 1;
            }
            processed
        });

        // Deploy script: promote and drain
        controller.promote().await;
        let orders = controller.take_drain_batch().await;
        StandbyController::drain_to(orders, &engine_tx).await;
        // Drain any orders that arrived during the drain phase
        loop {
            let additional = controller.take_drain_batch().await;
            if additional.is_empty() {
                break;
            }
            StandbyController::drain_to(additional, &engine_tx).await;
        }
        drop(engine_tx); // Signal no more orders

        // Engine should process all 3
        let processed = tokio::time::timeout(std::time::Duration::from_millis(500), engine_handle)
            .await
            .unwrap()
            .unwrap();
        assert_eq!(processed, 3);

        // All HTTP handlers should have received their responses
        for (i, mut rx) in response_rxs.into_iter().enumerate() {
            let resp = rx.recv().await.unwrap();
            assert_eq!(resp.order_id, Some(100 + i as u64));
            assert_eq!(
                resp.request_id.as_deref(),
                Some(&format!("deploy-order-{}", i) as &str)
            );
        }
    }

    #[tokio::test]
    async fn promote_signal_unblocks_waiting_task() {
        let (controller, promote_rx) = StandbyController::new_standby(100);

        let controller_clone = controller.clone();
        let waiter = tokio::spawn(async move {
            // Simulate engine waiting for promote
            promote_rx.await.ok();
            true
        });

        // Small delay to ensure waiter is parked
        tokio::time::sleep(std::time::Duration::from_millis(10)).await;

        controller_clone.promote().await;

        let unblocked = tokio::time::timeout(std::time::Duration::from_millis(100), waiter)
            .await
            .expect("Waiter should complete within timeout")
            .expect("Waiter task should not panic");

        assert!(unblocked);
    }
}