bacnet-network 0.15.0

use super::*;
use bacnet_transport::port::ReceivedNpdu;
use bacnet_transport::port::TransportProvenance;
use tokio::time::{Duration, timeout};

// Same owned-NPDU fake-transport pattern as layer_admission_tests: the real
// dispatch task receives payload, provenance, attributes and the reply sender.
struct IngressTransport(Option<mpsc::Receiver<ReceivedNpdu>>);

impl TransportPort for IngressTransport {
    async fn start(&mut self) -> Result<mpsc::Receiver<ReceivedNpdu>, Error> {
        Ok(self.0.take().expect("transport already started"))
    }

    async fn stop(&mut self) -> Result<(), Error> {
        Ok(())
    }

    async fn send_unicast(&self, _npdu: &[u8], _mac: &[u8]) -> Result<(), Error> {
        panic!("admission must not send wire replies or rejections")
    }

    async fn send_broadcast(&self, _npdu: &[u8]) -> Result<(), Error> {
        panic!("admission must not send wire replies or rejections")
    }

    fn local_receive_apdu_capacity(&self) -> u16 {
        1476
    }

    fn local_mac(&self) -> &[u8] {
        &[1]
    }
}

fn fixture() -> (NetworkLayer<IngressTransport>, mpsc::Sender<ReceivedNpdu>) {
    let (tx, rx) = mpsc::channel(1024);
    (NetworkLayer::new(IngressTransport(Some(rx))), tx)
}

fn incoming(source: &[u8], id: u16) -> ReceivedNpdu {
    let npdu = Npdu {
        expecting_reply: true,
        // Deliberately identical routed source: fairness must use the immediate
        // transport MAC, not the NPDU SNET/SADR or payload.
        source: Some(NpduAddress {
            network: 200,
            mac_address: MacAddr::from_slice(&[0x55]),
        }),
        payload: Bytes::copy_from_slice(&id.to_be_bytes()),
        ..Npdu::default()
    };
    let mut bytes = BytesMut::new();
    encode_npdu(&mut bytes, &npdu).unwrap();
    ReceivedNpdu {
        direct_response: None,
        npdu: bytes.freeze(),
        source_mac: MacAddr::from_slice(source),
        link_layer_group: true,
        data_attributes: vec![DataAttribute {
            option_type: 31,
            must_understand: false,
            data: vec![0x12, 0x34],
        }],
        provenance: TransportProvenance::unverified(),
        reply_tx: None,
    }
}

fn control() -> ReceivedNpdu {
    let mut arrival = incoming(&[2], 0);
    // Independent Router-Busy network-message vector.
    arrival.npdu = Bytes::from_static(&[1, 0x80, 4]);
    arrival
}

async fn barrier(tx: &mpsc::Sender<ReceivedNpdu>) {
    let (reply_tx, reply_rx) = oneshot::channel();
    let mut arrival = control();
    arrival.reply_tx = Some(reply_tx);
    tx.send(arrival).await.unwrap();
    // The control reply is released after all earlier ingress has dispatched,
    // without consuming any APDU or relying on scheduling sleeps.
    assert!(
        timeout(Duration::from_secs(1), reply_rx)
            .await
            .expect("dispatch stalled")
            .is_err()
    );
}

fn assert_apdu(apdu: &ReceivedApdu, source: &[u8], id: u16) {
    assert_eq!(apdu.apdu.as_ref(), id.to_be_bytes());
    assert_eq!(apdu.source_mac.as_slice(), source);
    assert_eq!(apdu.ingress_network, None);
    assert_eq!(
        apdu.source_network,
        Some(NpduAddress {
            network: 200,
            mac_address: MacAddr::from_slice(&[0x55]),
        })
    );
    assert!(apdu.link_layer_group);
    assert!(apdu.is_group);
    assert_eq!(apdu.data_attributes, incoming(source, id).data_attributes);
}

fn assert_sources(counters: &QueueAdmissionCounters, entries: usize, depth: usize) {
    let state = counters.0.lock().unwrap();
    assert_eq!(state.snapshot.current_depth, depth);
    assert_eq!(state.queued_by_source.len(), entries);
    state.assert_source_depth();
}

#[tokio::test(start_paused = true)]
async fn single_source_flood_caps_at_sixteen_and_another_source_progresses() {
    let (mut network, tx) = fixture();
    let mut apdus = network.start_with_admission().await.unwrap();
    let counters = apdus.counters();
    let (reply_tx, mut accepted_reply) = oneshot::channel();
    let mut first = incoming(&[2], 0);
    first.reply_tx = Some(reply_tx);
    tx.send(first).await.unwrap();
    for id in 1..256 {
        let (reply_tx, reply_rx) = oneshot::channel();
        let mut arrival = incoming(&[2], id);
        arrival.reply_tx = Some(reply_tx);
        tx.send(arrival).await.unwrap();
        if id >= 16 {
            assert!(
                timeout(Duration::from_secs(1), reply_rx)
                    .await
                    .expect("fairness drop retained reply sender")
                    .is_err()
            );
        }
    }
    for id in 0..16 {
        tx.send(incoming(&[3], id)).await.unwrap();
    }
    barrier(&tx).await;
    assert_eq!(
        counters.snapshot(),
        QueueAdmissionSnapshot {
            current_depth: 32,
            high_water: 32,
            fairness_drops: 240,
            ..Default::default()
        }
    );
    assert_sources(&counters, 2, 32);
    assert_eq!(
        accepted_reply.try_recv(),
        Err(oneshot::error::TryRecvError::Empty)
    );
    for source in [2, 3] {
        for id in 0..16 {
            let apdu = apdus.try_recv().unwrap();
            assert_apdu(&apdu, &[source], id);
            if source == 2 && id == 0 {
                apdu.reply_tx
                    .unwrap()
                    .send(Bytes::from_static(b"reply"))
                    .unwrap();
            }
        }
    }
    assert_eq!(accepted_reply.await.unwrap(), Bytes::from_static(b"reply"));
    assert_eq!(
        apdus.try_recv().unwrap_err(),
        mpsc::error::TryRecvError::Empty
    );
    assert_sources(&counters, 0, 0);
    network.stop().await.unwrap();
}

#[tokio::test(start_paused = true)]
async fn every_recv_and_try_recv_releases_one_slot_even_if_consumer_retains_items() {
    let (mut network, tx) = fixture();
    let mut apdus = network.start_with_admission().await.unwrap();
    let counters = apdus.counters();
    for id in 0..16 {
        tx.send(incoming(&[2], id)).await.unwrap();
    }
    barrier(&tx).await;
    let retained = [apdus.recv().await.unwrap(), apdus.try_recv().unwrap()];
    assert_sources(&counters, 1, 14);
    for id in 16..19 {
        tx.send(incoming(&[2], id)).await.unwrap();
    }
    barrier(&tx).await;
    assert_eq!(counters.snapshot().fairness_drops, 1);
    assert_sources(&counters, 1, 16);
    for id in 2..18 {
        let apdu = if id % 2 == 0 {
            apdus.recv().await.unwrap()
        } else {
            apdus.try_recv().unwrap()
        };
        assert_apdu(&apdu, &[2], id);
        let remaining = usize::from(17 - id);
        assert_sources(&counters, usize::from(remaining != 0), remaining);
    }
    // Empty and cancelled receives neither release a second slot nor leak keys.
    assert_eq!(
        apdus.try_recv().unwrap_err(),
        mpsc::error::TryRecvError::Empty
    );
    assert!(timeout(Duration::from_secs(1), apdus.recv()).await.is_err());
    for id in 19..35 {
        tx.send(incoming(&[2], id)).await.unwrap();
    }
    barrier(&tx).await;
    assert_sources(&counters, 1, 16);
    assert_eq!(counters.snapshot().fairness_drops, 1);
    assert_eq!(counters.snapshot().high_water, 16);
    assert_apdu(&retained[0], &[2], 0);
    assert_apdu(&retained[1], &[2], 1);
    network.stop().await.unwrap();
    for id in 19..35 {
        assert_apdu(&apdus.recv().await.unwrap(), &[2], id);
    }
    assert!(apdus.recv().await.is_none());
    assert_sources(&counters, 0, 0);
}

#[tokio::test(start_paused = true)]
async fn sixteen_sources_compose_to_capacity_and_fairness_precedes_full() {
    let (mut network, tx) = fixture();
    let mut apdus = network.start_with_admission().await.unwrap();
    let counters = apdus.counters();
    for source in 0..16 {
        for id in 0..16 {
            tx.send(incoming(&[source], id)).await.unwrap();
        }
    }
    tx.send(incoming(&[0], 16)).await.unwrap(); // Over quota AND globally full.
    tx.send(incoming(&[16], 0)).await.unwrap(); // Below quota, globally full.
    barrier(&tx).await;
    assert_eq!(
        counters.snapshot(),
        QueueAdmissionSnapshot {
            current_depth: 256,
            high_water: 256,
            fairness_drops: 1,
            full_drops: 1,
            closed_drops: 0,
        }
    );
    assert_sources(&counters, 16, 256);
    for source in 0..16 {
        for id in 0..16 {
            assert_apdu(&apdus.try_recv().unwrap(), &[source], id);
        }
    }
    assert_sources(&counters, 0, 0);
    network.stop().await.unwrap();
}

#[tokio::test(start_paused = true)]
async fn unique_source_churn_cannot_grow_map_beyond_queued_items() {
    let (mut network, tx) = fixture();
    let mut apdus = network.start_with_admission().await.unwrap();
    let counters = apdus.counters();
    for round in 0..3u16 {
        for id in 0..512u16 {
            tx.send(incoming(&(round * 512 + id).to_be_bytes(), id))
                .await
                .unwrap();
        }
        barrier(&tx).await;
        assert_sources(&counters, 256, 256);
        assert_eq!(counters.snapshot().full_drops, u64::from(round + 1) * 256);
        assert_eq!(counters.snapshot().fairness_drops, 0);
        for id in 0..256u16 {
            assert_apdu(
                &apdus.try_recv().unwrap(),
                &(round * 512 + id).to_be_bytes(),
                id,
            );
            assert_sources(&counters, usize::from(255 - id), usize::from(255 - id));
        }
        assert_sources(&counters, 0, 0);
    }
    network.stop().await.unwrap();
}

#[tokio::test(start_paused = true)]
async fn close_and_drop_preserve_closed_precedence_and_release_all_source_entries() {
    for close_then_drain in [true, false] {
        let (mut network, tx) = fixture();
        let mut apdus = network.start_with_admission().await.unwrap();
        let counters = apdus.counters();
        let (reply_tx, mut queued_reply) = oneshot::channel();
        let mut first = incoming(&[2], 0);
        first.reply_tx = Some(reply_tx);
        tx.send(first).await.unwrap();
        for id in 1..16 {
            tx.send(incoming(&[2], id)).await.unwrap();
        }
        barrier(&tx).await;
        let mut apdus = if close_then_drain {
            apdus.close();
            assert_sources(&counters, 1, 16);
            assert_eq!(
                queued_reply.try_recv(),
                Err(oneshot::error::TryRecvError::Empty)
            );
            Some(apdus)
        } else {
            drop(apdus);
            assert_sources(&counters, 0, 0);
            None
        };
        let (reply_tx, reply_rx) = oneshot::channel();
        let mut arrival = incoming(&[2], 16);
        arrival.reply_tx = Some(reply_tx);
        tx.send(arrival).await.unwrap();
        assert!(
            timeout(Duration::from_secs(1), reply_rx)
                .await
                .unwrap()
                .is_err()
        );
        timeout(Duration::from_secs(1), tx.closed()).await.unwrap();
        assert_eq!(counters.snapshot().closed_drops, 1);
        assert_eq!(counters.snapshot().fairness_drops, 0);
        assert_eq!(counters.snapshot().full_drops, 0);
        if let Some(apdus) = apdus.as_mut() {
            for id in 0..16 {
                assert_apdu(&apdus.recv().await.unwrap(), &[2], id);
            }
            assert!(apdus.recv().await.is_none());
        }
        assert!(queued_reply.await.is_err());
        assert_sources(&counters, 0, 0);
        network.stop().await.unwrap();
    }
}

#[tokio::test(start_paused = true)]
async fn raw_apdus_and_tracked_controls_have_no_per_source_cap() {
    let (mut network, tx) = fixture();
    let mut controls = network
        .enable_network_control_receiver_with_admission()
        .unwrap();
    let mut apdus = network.start().await.unwrap();
    for id in 0..257 {
        tx.send(incoming(&[2], id)).await.unwrap();
        tx.send(control()).await.unwrap();
    }
    // The raw receiver decodes as it is polled (#1713), so nothing is
    // admitted yet; then all 257 APDUs from one source come through.
    assert_eq!(
        controls.counters().snapshot(),
        QueueAdmissionSnapshot::default()
    );
    for id in 0..257 {
        assert_apdu(&apdus.try_recv().unwrap(), &[2], id);
    }
    assert_eq!(
        apdus.try_recv().unwrap_err(),
        mpsc::error::TryRecvError::Empty
    );
    assert_eq!(
        controls.counters().snapshot(),
        QueueAdmissionSnapshot {
            current_depth: 256,
            high_water: 256,
            full_drops: 1,
            ..Default::default()
        }
    );
    assert!(
        controls
            .counters
            .0
            .lock()
            .unwrap()
            .queued_by_source
            .is_empty()
    );
    for id in 0..256u64 {
        let control = controls.try_recv().unwrap();
        assert_eq!(control.source_mac.as_slice(), &[2]);
        assert_eq!(control.ingress_sequence, id + 1);
    }
    network.stop().await.unwrap();
}

#[tokio::test(start_paused = true)]
async fn mac_keys_use_complete_byte_value_not_allocation_or_prefix() {
    fn requires_hash_eq<T: std::hash::Hash + Eq>() {}
    requires_hash_eq::<MacAddr>();
    let (mut network, tx) = fixture();
    let apdus = network.start_with_admission().await.unwrap();
    let counters = apdus.counters();
    for id in 0..16 {
        tx.send(incoming(&[2], id)).await.unwrap();
    }
    let mut spilled = MacAddr::with_capacity(32);
    spilled.push(2);
    assert!(spilled.spilled());
    let mut arrival = incoming(&[2], 16);
    arrival.source_mac = spilled;
    tx.send(arrival).await.unwrap();
    for source in [&[2, 0][..], &[2; 18][..]] {
        for id in 0..17 {
            tx.send(incoming(source, id)).await.unwrap();
        }
    }
    barrier(&tx).await;
    assert_sources(&counters, 3, 48);
    assert_eq!(counters.snapshot().fairness_drops, 3);
    assert_eq!(counters.snapshot().full_drops, 0);
    network.stop().await.unwrap();
    assert_sources(&counters, 3, 48); // Stop retains queued items and keys.
    drop(apdus);
    assert_sources(&counters, 0, 0);
}

#[tokio::test(start_paused = true)]
async fn fairness_drop_counter_saturates_without_altering_depth_or_other_drops() {
    let (mut network, tx) = fixture();
    let apdus = network.start_with_admission().await.unwrap();
    let counters = apdus.counters();
    for id in 0..16 {
        tx.send(incoming(&[2], id)).await.unwrap();
    }
    barrier(&tx).await;
    counters.0.lock().unwrap().snapshot.fairness_drops = u64::MAX - 1;
    for id in 16..19 {
        tx.send(incoming(&[2], id)).await.unwrap();
    }
    barrier(&tx).await;
    assert_eq!(
        counters.snapshot(),
        QueueAdmissionSnapshot {
            current_depth: 16,
            high_water: 16,
            fairness_drops: u64::MAX,
            ..Default::default()
        }
    );
    assert_sources(&counters, 1, 16);
    network.stop().await.unwrap();
}

#[tokio::test]
async fn concurrent_dispatch_and_dequeues_keep_source_and_depth_accounting_in_step() {
    let (mut network, tx) = fixture();
    let mut apdus = network.start_with_admission().await.unwrap();
    let counters = apdus.counters();
    let receiver_counters = counters.clone();
    // The crate enables Tokio's current-thread runtime, not rt-multi-thread.
    // Moving the receiver to a second OS thread still races its accounting with
    // real NetworkLayer dispatch, without broadening the dependency features.
    let consumer = std::thread::spawn(move || {
        tokio::runtime::Builder::new_current_thread()
            .enable_all()
            .build()
            .unwrap()
            .block_on(async move {
                let mut seen = std::collections::HashSet::new();
                for _ in 0..256 {
                    let apdu = timeout(Duration::from_secs(5), apdus.recv())
                        .await
                        .unwrap()
                        .unwrap();
                    let id = u16::from_be_bytes(apdu.apdu.as_ref().try_into().unwrap());
                    let source = apdu.source_mac[0];
                    assert!(source < 16 && id < 16);
                    assert!(seen.insert((source, id)));
                    assert_apdu(&apdu, &[source], id);
                    receiver_counters.0.lock().unwrap().assert_source_depth();
                }
                assert_sources(&receiver_counters, 0, 0);
            });
    });
    // Sixteen items per source fit even if dispatch outruns the consumer;
    // if they race, source-key insertions and removals must stay in step.
    for id in 0..16 {
        for source in 0..16 {
            tx.send(incoming(&[source], id)).await.unwrap();
        }
    }
    barrier(&tx).await;
    consumer.join().unwrap();
    assert_sources(&counters, 0, 0);
    assert_eq!(counters.snapshot().fairness_drops, 0);
    assert_eq!(counters.snapshot().full_drops, 0);
    assert_eq!(counters.snapshot().closed_drops, 0);
    network.stop().await.unwrap();
}

fn router_apdu(ingress_network: u16, source: &[u8], id: u16) -> ReceivedApdu {
    ReceivedApdu {
        router_response: None,
        direct_response: None,
        apdu: Bytes::copy_from_slice(&id.to_be_bytes()),
        source_mac: MacAddr::from_slice(source),
        ingress_network: Some(ingress_network),
        source_network: None,
        link_layer_group: false,
        is_group: false,
        global_broadcast: false,
        data_attributes: Vec::new(),
        provenance: TransportProvenance::unverified(),
        reply_tx: None,
    }
}

#[test]
fn router_key_churn_keeps_entries_bounded_by_depth_and_drop_clears_them() {
    let (tx, rx, counters) = AdmissionReceiver::channel(true);
    let mut apdus = AdmissionReceiver::from_apdu_parts(rx, counters.clone());
    for round in 0..3 {
        for id in 0..512 {
            // Same MAC on 256 distinct ports, then fresh keys while full.
            let _ = tx.try_send_apdu(router_apdu(id % 256, &[round, (id / 256) as u8], id));
        }
        assert_sources(&counters, 256, 256);
        assert_eq!(counters.snapshot().full_drops, u64::from(round + 1) * 256);
        assert_eq!(counters.snapshot().fairness_drops, 0);
        for port in 0..256 {
            let mut apdu = apdus.try_recv().unwrap();
            assert_eq!(apdu.ingress_network, Some(port));
            assert!(
                !counters
                    .0
                    .lock()
                    .unwrap()
                    .queued_by_source
                    .contains_key(&apdu_source(&apdu))
            );
            apdu.ingress_network = None;
            apdu.source_mac.clear();
            assert_sources(&counters, usize::from(255 - port), usize::from(255 - port));
        }
    }
    for port in 0..256 {
        tx.try_send_apdu(router_apdu(port, &[2], 0)).unwrap();
    }
    apdus.close();
    assert_sources(&counters, 256, 256);
    drop(apdus);
    assert_sources(&counters, 0, 0);
}

#[test]
fn cloned_router_senders_enforce_composite_quotas_across_threads_and_dequeues() {
    let (tx, rx, counters) = AdmissionReceiver::channel(true);
    let mut apdus = AdmissionReceiver::from_apdu_parts(rx, counters.clone());
    let start = std::sync::Barrier::new(4);
    std::thread::scope(|scope| {
        for port in 100..104 {
            let tx = tx.clone();
            let start = &start;
            scope.spawn(move || {
                start.wait();
                for id in 0..256 {
                    let _ = tx.try_send_apdu(router_apdu(port, &[2], id));
                }
            });
        }
    });
    assert_sources(&counters, 4, 64);
    assert_eq!(counters.snapshot().fairness_drops, 960);
    for _ in 0..64 {
        apdus.try_recv().unwrap();
    }
    assert_sources(&counters, 0, 0);
    let start = std::sync::Barrier::new(5);
    std::thread::scope(|scope| {
        for port in 100..104 {
            let tx = tx.clone();
            let start = &start;
            scope.spawn(move || {
                start.wait();
                for id in 0..16 {
                    tx.try_send_apdu(router_apdu(port, &[2], id)).unwrap();
                }
            });
        }
        start.wait();
        let deadline = std::time::Instant::now() + Duration::from_secs(5);
        let mut seen = std::collections::HashSet::new();
        while seen.len() < 64 {
            match apdus.try_recv() {
                Ok(apdu) => assert!(seen.insert((apdu.ingress_network, apdu.apdu))),
                Err(mpsc::error::TryRecvError::Empty) => {
                    assert!(std::time::Instant::now() < deadline, "sender stalled");
                    std::thread::yield_now();
                }
                Err(e) => panic!("unexpected receive error: {e}"),
            }
            counters.0.lock().unwrap().assert_source_depth();
        }
    });
    assert_sources(&counters, 0, 0);
    assert_eq!(counters.snapshot().fairness_drops, 960);
    assert_eq!(counters.snapshot().full_drops, 0);
    assert_eq!(counters.snapshot().closed_drops, 0);
}

/// The raw receiver's `recv` is cancel safe (#1713): the server and client
/// wait on it in `select!`, which drops it whenever another branch wins. Here
/// it is dropped over and over while a sender feeds APDUs and network
/// messages in gaps; every APDU arrives once, in order, and every control
/// reaches its receiver.
#[tokio::test(start_paused = true)]
async fn raw_receive_cancelled_mid_stream_loses_nothing() {
    const COUNT: u16 = 200;
    let (mut network, tx) = fixture();
    let mut controls = network.enable_network_control_receiver().unwrap();
    let mut apdus = network.start().await.unwrap();
    let sender = tokio::spawn(async move {
        for id in 0..COUNT {
            tx.send(incoming(&[2], id)).await.unwrap();
            tx.send(control()).await.unwrap();
            if id % 3 == 0 {
                tokio::time::sleep(Duration::from_millis(5)).await;
            }
        }
        tx
    });
    let mut next = 0;
    let mut cancelled = 0;
    while next < COUNT {
        tokio::select! {
            biased;
            received = apdus.recv() => {
                assert_apdu(&received.unwrap(), &[2], next);
                next += 1;
            }
            () = tokio::time::sleep(Duration::from_millis(1)) => {
                cancelled += 1;
                assert!(cancelled < 10_000, "APDU {next} never arrived");
            }
        }
    }
    assert!(cancelled > 0, "the receive was never cancelled");
    let _tx = sender.await.unwrap();
    crate::loopback_fixture::pump(&mut apdus);
    for _ in 0..COUNT {
        controls.try_recv().unwrap();
    }
    assert!(controls.try_recv().is_err());
    network.stop().await.unwrap();
}