Function await_relay_stream

crates/openshell-server/src/grpc/sandbox.rs:884–913

Wait for the supervisor's reverse CONNECT to deliver a relay stream. Returns `Some(stream)` on success. On any failure the error is sent on `tx` and `None` is returned; the caller should then `return` immediately.

(
    relay_rx: oneshot::Receiver<Result<tokio::io::DuplexStream, Status>>,
    tx: &mpsc::Sender<Result<T, Status>>,
    sandbox_id: &str,
    channel_id: &str,
    context: &str,
)

/// Returns `Some(stream)` on success. On any failure the error is sent on `tx`
/// and `None` is returned; the caller should then `return` immediately.
async fn await_relay_stream<T: Send + 'static>(
    relay_rx: oneshot::Receiver<Result<tokio::io::DuplexStream, Status>>,
    tx: &mpsc::Sender<Result<T, Status>>,
    sandbox_id: &str,
    channel_id: &str,
    context: &str,
) -> Option<tokio::io::DuplexStream> {
    match tokio::time::timeout(std::time::Duration::from_secs(10), relay_rx).await {
        Ok(Ok(Ok(stream))) => Some(stream),
        Ok(Ok(Err(status))) => {
            warn!(sandbox_id = %sandbox_id, channel_id = %channel_id, error = %status.message(), "{context}: relay target open failed");
            let _ = tx.send(Err(status)).await;
            None
        }
        Ok(Err(_)) => {
            warn!(sandbox_id = %sandbox_id, channel_id = %channel_id, "{context}: relay channel dropped");
            let _ = tx
                .send(Err(Status::unavailable("relay channel dropped")))
                .await;
            None
        }
        Err(_) => {
            warn!(sandbox_id = %sandbox_id, channel_id = %channel_id, "{context}: relay open timed out");
            let _ = tx
                .send(Err(Status::deadline_exceeded("relay open timed out")))
                .await;
            None
        }
    }
}

pub(super) async fn handle_forward_tcp(
    state: &Arc<ServerState>,