Deprecated ServerInviteDialog::handle moves the dialog from WaitAck back to Calling when accept() came first #196

//! Issue B: `ServerInviteDialog::handle()` moves the dialog from `WaitAck`
//! back to `Calling` when `accept()` ran first.
//!
//! Part A (deterministic, deprecated `ServerInviteDialog`): `accept()`, then
//! `handle()`. The peer does not ACK yet.
//! Part B (race, `InviteDialog`): `accept()` and `handle()` start on two
//! tasks; calls whose states go `WaitAck` -> `Calling` are counted.
//!
//! Usage: cargo run --release -- [calls for part B]   (default 1000)
#![allow(deprecated, dead_code)] // dead_code: the harness is shared by several issues
use rsipstack::dialog::dialog::{Dialog, DialogState};
use rsipstack::dialog::server_dialog::ServerInviteDialog;
use std::net::SocketAddr;
use std::time::{Duration, Instant};
use tokio::sync::mpsc::{unbounded_channel, UnboundedReceiver};

fn went_back(states: &[String]) -> bool {
    states
        .windows(2)
        .any(|w| w[0] == "WaitAck" && w[1] == "Calling")
}

#[tokio::main(flavor = "multi_thread")]
async fn main() -> Result<(), BoxError> {
    println!(
        "== issue B, WaitAck -> Calling | {} | 127.0.0.1 only",
        rsipstack::VERSION
    );
    let n: usize = std::env::args()
        .nth(1)
        .and_then(|a| a.parse().ok())
        .unwrap_or(1000);
    let stack = Stack::start().await?;
    let mut calls = stack.uas_loop()?;
    let uas = stack.addr;
    let peer = Peer::bind().await?;

    // Part A.
    peer.send(uas, &peer.invite(uas, "b-a")).await?;
    let mut call = calls.recv().await.ok_or("no call")?;
    let server = ServerInviteDialog::try_from(&Dialog::Invite(call.dialog.clone()))?;
    server.accept(None, None)?;
    let mut tx = call.tx.take().ok_or("no tx")?;
    let mut s2 = server.clone();
    tokio::spawn(async move { s2.handle(&mut tx).await });
    let states_a = drain_states(&mut call.states, Duration::from_millis(300)).await;
    let now_a = state_name(&server.state());
    let got_200 = peer
        .recv_until(Duration::from_millis(200), |r| r.is_response(200, "INVITE"))
        .await
        .is_some();
    let bye = tokio::time::timeout(Duration::from_secs(1), server.bye()).await;
    let bye_sent = peer
        .recv_until(Duration::from_millis(300), |r| r.is_request("BYE"))
        .await
        .is_some();
    println!(
        "A (ServerInviteDialog, accept() then handle()): 200 sent: {got_200}; states: {states_a:?}"
    );
    println!(
        "A: state() 300 ms later, no ACK yet: {now_a}; bye() -> {:?}; BYE on the wire: {bye_sent}",
        bye.map(|r| r.map_err(|e| e.to_string()))
    );

    // Part B.
    let mut backwards = 0;
    for i in 0..n {
        let call_id = format!("b-b-{i}");
        peer.send(uas, &peer.invite(uas, &call_id)).await?;
        let mut call = calls.recv().await.ok_or("no call")?;
        let d = call.dialog.clone();
        // Shift accept() by 0..63 scheduler yields to sweep the window.
        let accept = tokio::spawn(async move {
            for _ in 0..(i % 64) {
                tokio::task::yield_now().await;
            }
            d.accept(None, None)
        });
        call.spawn_handle();
        let _ = accept.await?;
        if went_back(&drain_states(&mut call.states, Duration::from_millis(30)).await) {
            backwards += 1;
        }
        let ok = peer
            .recv_until(Duration::from_millis(500), |r| {
                r.is_response(200, "INVITE")
                    && header(&r.text, "Call-ID").as_deref() == Some(call_id.as_str())
            })
            .await;
        if let Some(ok) = ok {
            peer.send(uas, &peer.ack_for(uas, &ok)).await?;
        }
    }
    println!(
        "B (InviteDialog, accept() racing handle()): {backwards}/{n} calls went WaitAck -> Calling"
    );
    let a = went_back(&states_a);
    verdict(
        a || backwards > 0,
        &format!("part A backwards: {a}; part B backwards: {backwards}/{n}"),
    )
}

// ---- 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 {
    /// 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"
        )
    }
}