InviteDialog::cancel() returns Ok(()) and does nothing: UAC before the first provisional, and any UAS dialog #197

//! Issue C: `InviteDialog::cancel()` returns `Ok(())` and sends nothing.
//!
//! Part A (UAC, no provisional yet): `cancel()` after the INVITE went out and
//! before any response. The peer then sends 180 and, 1 s later, 200.
//! Expected: CANCEL after the 180 (RFC 3261 §9.1), or ACK + BYE on a 2xx.
//! Part B (UAS): `cancel()` on an unanswered server dialog.
//!
//! Usage: cargo run --release
#![allow(dead_code)] // the harness at the bottom is shared by several issues
use rsipstack::dialog::dialog::DialogState;
use rsipstack::dialog::invitation::InviteOption;
use std::net::SocketAddr;
use std::time::{Duration, Instant};
use tokio::sync::mpsc::{unbounded_channel, UnboundedReceiver};

/// First lines received within `wait`; a BYE is answered 200.
async fn watch(peer: &Peer, wait: Duration) -> Result<Vec<String>, BoxError> {
    let deadline = Instant::now() + wait;
    let mut out = Vec::new();
    while let Some(rx) = peer
        .recv(deadline.saturating_duration_since(Instant::now()))
        .await
    {
        if rx.is_request("BYE") {
            peer.send(rx.from, &peer.response_to(&rx, 200, "OK", ""))
                .await?;
        }
        out.push(rx.first_line().to_string());
    }
    Ok(out)
}

#[tokio::main(flavor = "multi_thread")]
async fn main() -> Result<(), BoxError> {
    println!(
        "== issue C, cancel() no-op | {} | 127.0.0.1 only",
        rsipstack::VERSION
    );

    // ---- Part A: UAC ----
    let stack = Stack::start().await?;
    let peer = Peer::bind().await?;
    let me = stack.addr;
    let (state_tx, mut state_rx) = unbounded_channel();
    let opt = InviteOption {
        caller: format!("sip:alice@{me}").as_str().try_into()?,
        callee: format!("sip:bob@{}", peer.addr).as_str().try_into()?,
        contact: format!("sip:alice@{me}").as_str().try_into()?,
        ..Default::default()
    };
    let (dialog, join) = stack.dialogs.do_invite_async(opt, state_tx)?;
    let invite = peer
        .recv_until(Duration::from_secs(2), |r| r.is_request("INVITE"))
        .await
        .ok_or("no INVITE")?;
    let cancel = dialog.cancel().await;
    let after_cancel = peer.collect(Duration::from_secs(1)).await;
    peer.send(invite.from, &peer.response_to(&invite, 180, "Ringing", ""))
        .await?;
    let after_180 = watch(&peer, Duration::from_secs(1)).await?;
    let cancel_seen = after_180.iter().any(|l| l.starts_with("CANCEL "));
    let mut after_200 = Vec::new();
    if !cancel_seen {
        peer.send(invite.from, &peer.response_to(&invite, 200, "OK", ""))
            .await?;
        after_200 = watch(&peer, Duration::from_millis(1500)).await?;
    }
    let states_a = drain_states(&mut state_rx, Duration::from_millis(200)).await;
    join.abort();
    println!(
        "A: cancel() before any response -> {:?}",
        cancel.map_err(|e| e.to_string())
    );
    println!("A: peer received in the 1 s after cancel(): {after_cancel:?}");
    println!("A: after the peer's 180: {after_180:?}");
    if !cancel_seen {
        println!("A: after the peer's 200: {after_200:?}");
    }
    println!(
        "A: dialog states: {states_a:?}; final state(): {}",
        state_name(&dialog.state())
    );
    let ack_and_bye = ["ACK ", "BYE "]
        .iter()
        .all(|m| after_200.iter().any(|l| l.starts_with(m)));
    let a_repro = !cancel_seen && !ack_and_bye;
    drop(stack);

    // ---- Part B: UAS ----
    let stack = Stack::start().await?;
    let mut calls = stack.uas_loop()?;
    let uas = stack.addr;
    let peer = Peer::bind().await?;
    peer.send(uas, &peer.invite(uas, "c-uas")).await?;
    let mut call = calls.recv().await.ok_or("no call")?;
    call.spawn_handle();
    tokio::time::sleep(Duration::from_millis(50)).await;
    let cancel_b = call.dialog.cancel().await;
    let got = peer.collect(Duration::from_secs(2)).await;
    let states_b = drain_states(&mut call.states, Duration::from_millis(100)).await;
    println!(
        "B: UAS cancel() on an unanswered INVITE -> {:?}",
        cancel_b.as_ref().map_err(|e| e.to_string())
    );
    println!("B: peer received in 2 s: {got:?}; dialog states: {states_b:?}");
    let final_sent = got
        .iter()
        .any(|l| l.starts_with("SIP/2.0 4") || l.starts_with("SIP/2.0 6"));
    let b_repro = cancel_b.is_ok() && !final_sent;

    verdict(
        a_repro || b_repro,
        &format!("UAC cancel() lost: {a_repro}; UAS cancel() silent Ok: {b_repro}"),
    )
}

// ---- harness: the stack under test and a raw UDP peer, 127.0.0.1 only ----

type BoxError = Box<dyn std::error::Error + Send + Sync>;

fn verdict(reproduced: bool, what: &str) -> ! {
    let tag = if reproduced {
        "REPRODUCED"
    } else {
        "not reproduced"
    };
    println!("RESULT: {tag} - {what}");
    std::process::exit(i32::from(reproduced))
}

/// The stack under test: one UDP transport on 127.0.0.1:0, default options.
struct Stack {
    endpoint: rsipstack::transaction::Endpoint,
    dialogs: std::sync::Arc<rsipstack::dialog::dialog_layer::DialogLayer>,
    addr: SocketAddr,
    cancel: tokio_util::sync::CancellationToken,
}

impl Stack {
    async fn start() -> Result<Stack, BoxError> {
        let cancel = tokio_util::sync::CancellationToken::new();
        let tl = rsipstack::transport::TransportLayer::new(cancel.child_token());
        let udp = rsipstack::transport::udp::UdpConnection::create_connection(
            "127.0.0.1:0".parse()?,
            None,
            Some(cancel.child_token()),
        )
        .await?;
        let addr = udp.get_addr().get_socketaddr()?;
        tl.add_transport(udp.into());
        let endpoint = rsipstack::EndpointBuilder::new()
            .with_user_agent("repro")
            .with_transport_layer(tl)
            .with_cancel_token(cancel.child_token())
            .build();
        let inner = endpoint.inner.clone();
        tokio::spawn(async move { inner.serve().await });
        let dialogs = std::sync::Arc::new(rsipstack::dialog::dialog_layer::DialogLayer::new(
            endpoint.inner.clone(),
        ));
        Ok(Stack {
            endpoint,
            dialogs,
            addr,
            cancel,
        })
    }
}

impl Drop for Stack {
    fn drop(&mut self) {
        self.cancel.cancel();
    }
}

/// A new server INVITE dialog, its state receiver and its (not yet handled)
/// transaction.
struct UasCall {
    dialog: rsipstack::dialog::invite_dialog::InviteDialog,
    states: UnboundedReceiver<DialogState>,
    tx: Option<rsipstack::transaction::Transaction>,
}

impl UasCall {
    fn spawn_handle(&mut self) {
        if let Some(mut tx) = self.tx.take() {
            let mut d = self.dialog.clone();
            tokio::spawn(async move { d.handle(&mut tx).await });
        }
    }
}

impl Stack {
    /// The usual UAS loop: in-dialog requests go to `match_dialog` + `handle()`;
    /// each new INVITE gets `get_or_create_server_invite` and is handed over.
    fn uas_loop(&self) -> Result<UnboundedReceiver<UasCall>, BoxError> {
        use rsipstack::sip::prelude::HeadersExt;
        let mut incoming = self.endpoint.incoming_transactions()?;
        let dl = self.dialogs.clone();
        let (call_tx, call_rx) = unbounded_channel();
        tokio::spawn(async move {
            while let Some(mut tx) = incoming.recv().await {
                let to_tag = tx
                    .original
                    .to_header()
                    .ok()
                    .and_then(|h| h.tag().ok().flatten());
                if to_tag.is_some() {
                    if let Some(mut d) = dl.match_dialog(&tx) {
                        tokio::spawn(async move { d.handle(&mut tx).await });
                    }
                } else if tx.original.method == rsipstack::sip::Method::Invite {
                    let (state_tx, states) = unbounded_channel();
                    if let Ok(dialog) = dl.get_or_create_server_invite(&tx, state_tx, None, None) {
                        let _ = call_tx.send(UasCall {
                            dialog,
                            states,
                            tx: Some(tx),
                        });
                    }
                }
            }
        });
        Ok(call_rx)
    }
}

fn state_name(s: &DialogState) -> String {
    match s {
        DialogState::Calling(_) => "Calling".into(),
        DialogState::Trying(_) => "Trying".into(),
        DialogState::Early(..) => "Early".into(),
        DialogState::WaitAck(..) => "WaitAck".into(),
        DialogState::Confirmed(..) => "Confirmed".into(),
        DialogState::Terminated(_, r) => format!("Terminated({r:?})"),
        _ => "Other".into(),
    }
}

/// Every state queued on `rx` within `wait`.
async fn drain_states(rx: &mut UnboundedReceiver<DialogState>, wait: Duration) -> Vec<String> {
    let mut out = Vec::new();
    let deadline = Instant::now() + wait;
    while let Ok(Some(s)) = tokio::time::timeout_at(deadline.into(), rx.recv()).await {
        out.push(state_name(&s));
    }
    out
}

/// The far end: a raw UDP socket that writes and reads SIP text by hand.
struct Peer {
    sock: tokio::net::UdpSocket,
    addr: SocketAddr,
}

/// One datagram received by the peer.
struct Rx {
    text: String,
    from: SocketAddr,
}

impl Rx {
    fn first_line(&self) -> &str {
        self.text.lines().next().unwrap_or("")
    }
    fn is_request(&self, method: &str) -> bool {
        self.first_line().starts_with(&format!("{method} "))
    }
    fn is_response(&self, code: u16, cseq_method: &str) -> bool {
        self.first_line().starts_with(&format!("SIP/2.0 {code} "))
            && header(&self.text, "CSeq").is_some_and(|c| c.ends_with(cseq_method))
    }
}

/// The value of the first header `name` (full name, case-insensitive).
fn header(text: &str, name: &str) -> Option<String> {
    let prefix = format!("{}:", name.to_ascii_lowercase());
    text.split("\r\n")
        .find(|l| l.to_ascii_lowercase().starts_with(&prefix))
        .map(|l| l[prefix.len()..].trim().to_string())
}

impl Peer {
    async fn bind() -> Result<Peer, BoxError> {
        let sock = tokio::net::UdpSocket::bind("127.0.0.1:0").await?;
        let addr = sock.local_addr()?;
        Ok(Peer { sock, addr })
    }

    async fn send(&self, to: SocketAddr, text: &str) -> Result<(), BoxError> {
        self.sock.send_to(text.as_bytes(), to).await?;
        Ok(())
    }

    async fn recv(&self, wait: Duration) -> Option<Rx> {
        let mut buf = vec![0u8; 65536];
        let (n, from) = tokio::time::timeout(wait, self.sock.recv_from(&mut buf))
            .await
            .ok()?
            .ok()?;
        Some(Rx {
            text: String::from_utf8_lossy(&buf[..n]).into_owned(),
            from,
        })
    }

    /// The first datagram matching `pred` within `wait`; others are dropped.
    async fn recv_until(&self, wait: Duration, pred: impl Fn(&Rx) -> bool) -> Option<Rx> {
        let deadline = Instant::now() + wait;
        loop {
            let rx = self
                .recv(deadline.saturating_duration_since(Instant::now()))
                .await?;
            if pred(&rx) {
                return Some(rx);
            }
        }
    }
}

impl Peer {
    /// The first line of everything received within `wait`.
    async fn collect(&self, wait: Duration) -> Vec<String> {
        let deadline = Instant::now() + wait;
        let mut out = Vec::new();
        while let Some(rx) = self
            .recv(deadline.saturating_duration_since(Instant::now()))
            .await
        {
            out.push(rx.first_line().to_string());
        }
        out
    }
}

impl Peer {
    /// An out-of-dialog INVITE from this peer to `uas`.
    fn invite(&self, uas: SocketAddr, call_id: &str) -> String {
        let me = self.addr;
        format!(
            "INVITE sip:bob@{uas} SIP/2.0\r\n\
             Via: SIP/2.0/UDP {me};branch=z9hG4bK-{call_id}-inv;rport\r\n\
             Max-Forwards: 70\r\n\
             From: <sip:alice@{me}>;tag=a1\r\n\
             To: <sip:bob@{uas}>\r\n\
             Call-ID: {call_id}\r\n\
             CSeq: 1 INVITE\r\n\
             Contact: <sip:alice@{me}>\r\n\
             Content-Length: 0\r\n\r\n"
        )
    }

    /// The ACK for a 2xx (new branch, To copied from the 2xx).
    fn ack_for(&self, uas: SocketAddr, ok: &Rx) -> String {
        let me = self.addr;
        let to = header(&ok.text, "To").unwrap_or_default();
        let call_id = header(&ok.text, "Call-ID").unwrap_or_default();
        format!(
            "ACK sip:bob@{uas} SIP/2.0\r\n\
             Via: SIP/2.0/UDP {me};branch=z9hG4bK-{call_id}-ack;rport\r\n\
             Max-Forwards: 70\r\n\
             From: <sip:alice@{me}>;tag=a1\r\n\
             To: {to}\r\n\
             Call-ID: {call_id}\r\n\
             CSeq: 1 ACK\r\n\
             Content-Length: 0\r\n\r\n"
        )
    }
}

impl Peer {
    /// A response to `req`; a To tag is added when the request has none.
    fn response_to(&self, req: &Rx, code: u16, reason: &str, extra: &str) -> String {
        let mut out = format!("SIP/2.0 {code} {reason}\r\n");
        for line in req.text.split("\r\n") {
            let lower = line.to_ascii_lowercase();
            if ["via:", "from:", "call-id:", "cseq:", "to:"]
                .iter()
                .any(|h| lower.starts_with(h))
            {
                out.push_str(line);
                if lower.starts_with("to:") && !lower.contains(";tag=") {
                    out.push_str(";tag=b1");
                }
                out.push_str("\r\n");
            }
        }
        let me = self.addr;
        out + &format!("Contact: <sip:bob@{me}>\r\n{extra}Content-Length: 0\r\n\r\n")
    }
}