bacnet_network/

layer.rs

//! NetworkLayer for local BACnet packet assembly and dispatch.
//!
//! The network layer wraps a transport and provides APDU-level send/receive
//! by handling NPDU encoding/decoding. This is a non-router implementation:
//! it does not forward messages between networks, but it can address remote
//! devices through local routers via NPDU destination fields (DNET/DADR).
//!
//! # Receive-queue admission
//!
//! [`NetworkLayer`] and [`BACnetRouter`](crate::router::BACnetRouter) use the
//! same drop-arriving policy for application delivery. Each receive queue holds
//! **256 items**. Admission never waits for a consumer and never evicts an older
//! queued item. A tracked NetworkLayer APDU queue and an opted-in control queue
//! have independent capacities, so saturation of either does not stall
//! admission to the other. All router ports share one local APDU queue; its
//! saturation does not stop forwarding or inline network-message handling.
//!
//! The raw NetworkLayer receiver is different: [`ApduReceiver`] has no queue
//! of its own (#1713). It decodes the transport's NPDUs in the task that polls
//! it, so **network controls flow only while it is polled**: while the
//! consumer is busy in an inline await they wait in the transport's queue,
//! which adds latency, not a deadlock. That queue is the only buffer. It
//! holds 256 NPDUs on B/IP, B/IPv6, BACnet/SC, Ethernet and loopback, and 64
//! on MS/TP, whose deliveries were buffered 64 + 256 deep before. Under
//! overload it bounds memory and the transport counts what it drops: B/IP
//! and B/IPv6 as `npdu_channel_full` in `receive_counters()`, MS/TP as
//! `delivery_drops` in its diagnostics' MAC counters, BACnet/SC in
//! `npdu_drop_counts()`, and loopback in `receive_drops()`. Ethernet has no
//! drop counter, only throttled warnings.
//!
//! ## Raw and tracked receivers
//!
//! | Receiver kind | Entry points | Capacity | Per-source quota | Accounting |
//! | --- | --- | --- | --- | --- |
//! | Raw APDU, NetworkLayer | [`NetworkLayer::start`] ([`ApduReceiver`]) | The transport's queue | None | Drops counted by the transport; no admission snapshot |
//! | Raw APDU, router | [`BACnetRouter::start`](crate::router::BACnetRouter::start) with [`RouterOptions::new`](crate::router::RouterOptions::new) | 256 | None | Full/Closed counted internally; no admission snapshot or depth/high-water tracking |
//! | Tracked APDU | [`NetworkLayer::start_with_admission`], [`RouterOptions::track_admission`](crate::router::RouterOptions::track_admission) | 256 | 16 queued APDUs per key, defined below | Exact depth/high-water and Full/fairness/Closed totals via [`AdmissionReceiver::counters`] |
//! | Raw control | [`NetworkLayer::enable_network_control_receiver`], [`RouterOptions::network_control_receiver`](crate::router::RouterOptions::network_control_receiver) | 256, separate from APDUs | None | Full/Closed counted internally; no admission snapshot or depth/high-water tracking |
//! | Tracked control | [`NetworkLayer::enable_network_control_receiver_with_admission`], [`RouterOptions::network_control_receiver_with_admission`](crate::router::RouterOptions::network_control_receiver_with_admission) | 256, separate from APDUs | None | Exact depth/high-water and Full/Closed totals; fairness always zero |
//!
//! The router's raw receivers and the raw control receiver are
//! `tokio::sync::mpsc::Receiver`s; NetworkLayer's raw APDU receiver is an
//! [`ApduReceiver`]; tracked receivers are the [`AdmissionReceiver`]
//! alternatives. APDU and control opt-ins are
//! independent. Routers process network messages inline; their control receiver
//! gets only the Reject-Message-To-Network messages addressed to the router
//! itself, and every closed admission attempt counts as Closed without
//! disabling the stream. [`priority_channel`](crate::priority_channel) is not
//! used by these ingress paths.
//!
//! ## Source keys and drop precedence
//!
//! Tracked APDU queues cap each key at **16 queued, unconsumed APDUs**:
//! - NetworkLayer: the complete transport-native [`ReceivedApdu::source_mac`]
//!   byte value. Its single transport needs no separate port key.
//! - Router local delivery: ([ingress port network number](ReceivedApdu::ingress_network),
//!   [`source_mac`](ReceivedApdu::source_mac)). Identical MAC bytes on different
//!   ports have separate quotas. Neither key uses routed NPDU SNET/SADR
//!   ([`source_network`](ReceivedApdu::source_network)) or authenticates a peer.
//!
//! Each successful [`recv`](AdmissionReceiver::recv) or
//! [`try_recv`](AdmissionReceiver::try_recv) releases a queue slot and, for
//! tracked APDUs, that key's quota slot before returning the item. Retaining or
//! processing the returned item does not occupy either slot. Sixteen keys at
//! their quota fill the 256-item queue; a key below quota can still encounter
//! global Full. This is a queued-item cap, not a byte/rate limit or a delivery
//! guarantee. Raw receivers and control queues have no per-source quota.
//!
//! Drop attribution is **Closed > fairness > Full**: closed admission wins;
//! otherwise a tracked APDU whose key already holds 16 items is a fairness drop,
//! even if the queue also holds 256 items; otherwise global saturation is Full.
//! Only the winning reason is counted. Every admission drop releases the
//! arriving item's owned payload, metadata and any reply sender without sending
//! reply bytes or generating a wire rejection. This does not suppress ordinary
//! router forwarding/reject behavior or change transport no-reply handling.
//!
//! ## Close, drop, stop, and observation
//!
//! Closing either receiver kind prevents further admission but retains queued
//! items for draining. Dropping a receiver discards them and releases their
//! payloads, metadata and reply senders without sending reply bytes. For tracked
//! receivers, close preserves depth until dequeued; drop sets depth to zero.
//! Neither operation resets high-water/drop totals, and discarding queued items
//! is not an admission drop.
//!
//! A tracked NetworkLayer ends its dispatcher on the first APDU admission
//! attempt after that receiver closes/drops, also ending control delivery.
//! Closing an [`ApduReceiver`] stops the transport handing up NPDUs while
//! those it queued stay drainable; dropping it discards them. A closed/dropped
//! control receiver instead disables only that stream on the first subsequent
//! control admission attempt. Later discarded controls (like controls received
//! without either opt-in) are not admission attempts. Router forwarding and
//! inline control handling continue after local delivery closes/drops; each
//! subsequent local admission attempt is counted as Closed.
//!
//! [`NetworkLayer::stop`] after start and [`BACnetRouter::stop`](crate::router::BACnetRouter::stop)
//! end dispatch without waiting for application queues to drain; queued items
//! remain drainable, for an [`ApduReceiver`] the NPDUs the transport queued. Stopping is not an admission drop. An independently owned
//! [`QueueAdmissionCounters`] handle remains readable after the receiver, layer,
//! or router is dropped; it keeps accounting alive, not the channel or tasks.
//! [`snapshot`](QueueAdmissionCounters::snapshot) is consistent for one queue,
//! not atomic across queues, and exposes counts only, not payloads, per-source
//! identities or per-source totals. Its three drop totals saturate at `u64::MAX`.

use crate::network_number::LocalNetworkNumber;
use bacnet_encoding::npdu::{Npdu, NpduAddress, encode_npdu};
use bacnet_npdu::NetworkPriority;
use bacnet_transport::port::{
    DataAttribute, GroupDestinations, TransportPort, TransportProvenance,
};
use bacnet_types::MacAddr;
use bacnet_types::enums::PduType;
use bacnet_types::error::Error;
use bytes::{Bytes, BytesMut};
use std::sync::Arc;
use std::sync::atomic::{AtomicU64, Ordering};
use tokio::sync::oneshot;
use tokio::task::JoinHandle;

#[path = "layer_issuance.rs"]
mod issuance;
pub use issuance::IssuedApdu;

#[path = "layer_network_message.rs"]
mod network_message;
pub use network_message::{NetworkReport, ReceivedNetworkReport, SENDABLE_NETWORK_MESSAGES};

#[path = "layer_receive.rs"]
pub(crate) mod receive;
pub use receive::Counters;

#[path = "layer_ingress.rs"]
mod ingress;
pub use ingress::ApduReceiver;

#[path = "layer_admission.rs"]
mod admission;
pub(crate) use admission::AdmissionSender;
pub use admission::{AdmissionReceiver, QueueAdmissionCounters, QueueAdmissionSnapshot};

/// A received APDU with source addressing information.
pub struct ReceivedApdu {
    /// Raw APDU bytes.
    pub apdu: Bytes,
    /// Source MAC address in transport-native format (claimed identity; see
    /// [`TransportProvenance`]. On SC this is the source VMAC, not a
    /// certificate principal).
    pub source_mac: MacAddr,
    /// BACnet network number of the router ingress port at admission time.
    ///
    /// Router deliveries set this to `Some`, independently of the routed source
    /// address. Non-router [`NetworkLayer`] deliveries use `None` because the
    /// layer owns one transport without an assigned ingress network number.
    /// Both raw and tracked deliveries carry this metadata. For
    /// [`RouterOptions::track_admission`](crate::router::RouterOptions::track_admission),
    /// it scopes [`source_mac`](Self::source_mac) to the ingress port for the
    /// 16-queued-APDU quota; [`source_network`](Self::source_network) is not part
    /// of that key. See the [receive-queue contract](self#receive-queue-admission).
    pub ingress_network: Option<u16>,
    /// Source network address if the APDU was routed (NPDU had source field).
    /// Claimed SNET/SADR, never a credential (RB-07 compat mode: decisions
    /// unchanged; RB-09 consumes provenance later).
    pub source_network: Option<NpduAddress>,
    /// Whether the NPDU arrived through a data-link multicast or broadcast.
    ///
    /// This preserves raw data-link provenance independently of [`Self::is_group`].
    pub link_layer_group: bool,
    /// Whether the APDU's effective BACnet destination was multicast or broadcast.
    ///
    /// A specific DNET/DADR remains a unicast even when a router used a
    /// group data-link destination to reach that remote device.
    pub is_group: bool,
    /// Whether the NPDU named the global broadcast network (DNET 65535).
    ///
    /// A global broadcast is also a group delivery, but [`Self::is_group`]
    /// alone cannot tell it from a local or one-network broadcast once a
    /// router has passed it on. A Notification Forwarder ignores event
    /// notifications that arrive this way (Clause 12.51).
    pub global_broadcast: bool,
    /// Data-link attributes associated with the NPDU, if the transport supplied any.
    pub data_attributes: Vec<DataAttribute>,
    /// Honest transport + origin provenance, immutable by value (RB-07).
    /// Threaded from [`bacnet_transport::port::ReceivedNpdu`]; cloning
    /// preserves the meaning and never duplicates reply authority. Compat
    /// mode: forwarding/learning/admission decisions ignore it.
    pub provenance: TransportProvenance,
    /// Sealed route back to this envelope's original verified direct socket.
    pub direct_response: Option<bacnet_transport::port::DirectResponse>,
    /// Captured router endpoint and return-port authority for local delivery
    /// of a Confirmed-Request, the only APDU answered on its arrival route;
    /// `None` for every other APDU (#1713).
    pub router_response: Option<crate::router::RouterResponse>,
    /// Optional reply channel for MS/TP DataExpectingReply flows.
    /// The application layer can send NPDU-wrapped reply bytes through this channel.
    /// An APDU dropped at queue admission releases this sender without sending
    /// bytes, just as when the application receiver is closed. The transport
    /// observes channel closure and retains its existing no-reply handling.
    pub reply_tx: Option<oneshot::Sender<Bytes>>,
}

/// A decoded network-layer control message received from a transport peer.
///
/// Non-router users opt in to this stream before [`NetworkLayer::start`].
/// Without that opt-in, network messages retain their historical discard/log
/// behavior. A router opts in with
/// [`RouterOptions::network_control_receiver`](crate::router::RouterOptions::network_control_receiver),
/// and its stream carries only the rejects addressed to the router itself.
#[derive(Clone)]
pub struct ReceivedNetworkControl {
    /// Decoded NPDU, including network-message type and typed-address fields.
    pub npdu: Npdu,
    /// Immediate transport peer that sent the message (claimed identity; see
    /// [`TransportProvenance`]).
    pub source_mac: MacAddr,
    /// Whether the data-link delivery was multicast or broadcast.
    pub link_layer_group: bool,
    /// Data-link attributes supplied by the transport.
    pub data_attributes: Vec<DataAttribute>,
    /// Honest transport + origin provenance, immutable by value (RB-07).
    /// Compat mode: decisions unchanged.
    pub provenance: TransportProvenance,
    /// Monotonic decoded-ingress sequence assigned before channel delivery.
    ///
    /// A consumer can compare this with [`NetworkLayer::network_control_ingress_sequence`]
    /// when activating state so controls already queued at that point cannot
    /// be mistaken for feedback about the new state.
    pub ingress_sequence: u64,
}

impl std::fmt::Debug for ReceivedNetworkControl {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        // Redacted like ReceivedApdu: lengths only, no key material.
        f.debug_struct("ReceivedNetworkControl")
            .field("npdu", &self.npdu)
            .field("source_mac_len", &self.source_mac.len())
            .field("link_layer_group", &self.link_layer_group)
            .field("data_attributes", &self.data_attributes)
            .field("provenance", &self.provenance)
            .field("ingress_sequence", &self.ingress_sequence)
            .finish_non_exhaustive()
    }
}

impl Clone for ReceivedApdu {
    fn clone(&self) -> Self {
        Self {
            apdu: self.apdu.clone(),
            source_mac: self.source_mac.clone(),
            ingress_network: self.ingress_network,
            source_network: self.source_network.clone(),
            link_layer_group: self.link_layer_group,
            is_group: self.is_group,
            global_broadcast: self.global_broadcast,
            data_attributes: self.data_attributes.clone(),
            provenance: self.provenance.clone(),
            direct_response: self.direct_response.clone(),
            router_response: self.router_response.clone(),
            reply_tx: None,
        }
    }
}

impl std::fmt::Debug for ReceivedApdu {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        // Redacted: MAC bytes, routed source bytes, and payload lengths only;
        // no key material. Follows the TimeSyncSourceRestriction
        // finish_non_exhaustive pattern.
        let source_network_len = self
            .source_network
            .as_ref()
            .map(|source| (source.network, source.mac_address.len()));
        f.debug_struct("ReceivedApdu")
            .field("apdu_len", &self.apdu.len())
            .field("source_mac_len", &self.source_mac.len())
            .field("ingress_network", &self.ingress_network)
            .field("source_network", &source_network_len)
            .field("link_layer_group", &self.link_layer_group)
            .field("is_group", &self.is_group)
            .field("global_broadcast", &self.global_broadcast)
            .field("data_attributes", &self.data_attributes)
            .field("provenance", &self.provenance)
            .field("reply_tx", &self.reply_tx.as_ref().map(|_| "Some(...)"))
            .finish_non_exhaustive()
    }
}

/// Refuse a caller-supplied DNET that a send naming one network cannot take.
/// Network numbers start at 1 (Clause 6.2.2.1); the local network is reached
/// with no DNET at all, and `local_form` tells the caller how to do that.
/// DNET 0xFFFF selects every device on every network (Clauses 6.1 and
/// 6.3.2), and its sender puts it on its own network with the broadcast MAC
/// so that each router there can pass it on. Only
/// [`NetworkLayer::broadcast_global_apdu`] sends it that way, so every send
/// that checks here refuses DNET 0xFFFF, with or without a DADR (#1340,
/// #1380), and points to that method.
pub(crate) fn check_destination(network: u16, local_form: &str) -> Result<(), Error> {
    match network {
        0 => Err(Error::Encoding(format!(
            "dest_network 0 is not a network number; {local_form}"
        ))),
        0xFFFF => Err(Error::Encoding(
            "dest_network 0xFFFF is the global broadcast; \
             use broadcast_global_apdu for a global broadcast"
                .into(),
        )),
        _ => Ok(()),
    }
}

/// Refuse an APDU that a broadcast may not carry, before anything is encoded
/// or sent. Clause 6.3 lets only an Unconfirmed-Request PDU go to a broadcast
/// network address, whether local, remote or global: any other PDU names one
/// peer's transaction, and every device reached would have to answer it or
/// act on it (#1479). `what` names the broadcast for the error, which also
/// names the PDU type the APDU's first octet gives (Clause 20.1).
pub(crate) fn check_broadcast_apdu(apdu: &[u8], what: &str) -> Result<(), Error> {
    let pdu_type = match apdu.first() {
        Some(&first) if PduType::from_raw(first >> 4) == PduType::UNCONFIRMED_REQUEST => {
            return Ok(());
        }
        Some(&first) => format!("PDU type {}", PduType::from_raw(first >> 4)),
        None => "an empty APDU".to_string(),
    };
    Err(Error::Encoding(format!(
        "{what} carries only an UNCONFIRMED_REQUEST APDU (Clause 6.3), not {pdu_type}; \
         send anything else to one device"
    )))
}

/// A remote device reached through a local router.
#[derive(Debug, Clone, Copy)]
pub struct RoutedTarget<'a> {
    /// Destination network number written to the NPDU DNET.
    pub network: u16,
    /// Destination MAC on that network (NPDU DADR).
    pub mac: &'a [u8],
    /// Next-hop router MAC on the local network (the link-layer destination).
    pub router_mac: &'a [u8],
}

/// Non-router BACnet network layer.
///
/// Wraps a [`TransportPort`] and provides APDU-level send/receive by handling
/// NPDU framing. This layer does not act as a router (it does not forward
/// messages between networks), but it can send to remote devices through
/// local routers using NPDU destination addressing.
/// See the [receive-queue contract](self#receive-queue-admission) for raw/tracked
/// receiver choices, admission limits, drop attribution and lifecycle.
/// An inbound frame from a link-layer source MAC longer than
/// [`MAX_MAC_LEN`](bacnet_npdu::MAX_MAC_LEN), or whose DLEN or SLEN is past it, is dropped
/// before admission and counted by [`Self::address_length_drops`]; one whose
/// DNET 0xFFFF carries a DADR is dropped and counted by
/// [`Self::global_broadcast_dadr_drops`], and a broadcast whose APDU isn't an
/// Unconfirmed-Request by [`Self::broadcast_pdu_type_drops`]. These rules
/// are `bacnet-router-core`'s, the router engine's own (#1795): the layer
/// also drops a malformed NPDU, and one whose DNET names one network, APDU
/// or network message, which no node but a router acts on (Clauses
/// 6.5.2.1 and 6.5.2.2). [`Self::counters`] totals every drop by reason.
pub struct NetworkLayer<T: TransportPort> {
    transport: T,
    response_scope: bacnet_transport::port::DirectResponseScope,
    dispatch_task: Option<JoinHandle<()>>,
    /// Whether [`Self::start`] or [`Self::start_with_admission`] has run.
    started: bool,
    /// The [`ApduReceiver`]'s transport queue, which [`Self::stop`] and
    /// drop close so its stream ends once drained.
    receiver_queue: std::sync::Weak<
        std::sync::Mutex<tokio::sync::mpsc::Receiver<bacnet_transport::port::ReceivedNpdu>>,
    >,
    network_control_tx: Option<AdmissionSender<ReceivedNetworkControl>>,
    network_control_ingress_sequence: Arc<AtomicU64>,
    /// What the dispatch task's node dropped, since the layer was made.
    node_totals: Arc<receive::NodeTotals>,
    local_network_number: LocalNetworkNumber,
    /// The group destinations the transport's local broadcast stands for,
    /// taken when the layer is made and again when the transport starts and
    /// knows its own addresses, so a send never calls back into the
    /// transport to judge its destination.
    local_broadcasts: GroupDestinations,
}

impl<T: TransportPort + 'static> NetworkLayer<T> {
    /// Create a new network layer wrapping the given transport.
    pub fn new(transport: T) -> Self {
        Self {
            local_broadcasts: transport.local_broadcast_destinations(),
            transport,
            response_scope: Default::default(),
            dispatch_task: None,
            started: false,
            receiver_queue: std::sync::Weak::new(),
            network_control_tx: None,
            network_control_ingress_sequence: Arc::new(AtomicU64::new(0)),
            node_totals: Arc::default(),
            local_network_number: LocalNetworkNumber::default(),
        }
    }

    /// Send an APDU to a specific local destination by MAC address.
    ///
    /// With no DNET, the link's broadcast MAC or any other group address
    /// ([`TransportPort::is_group_destination`]) makes this a local
    /// broadcast, which may carry only an Unconfirmed-Request APDU (Clause
    /// 6.3).
    ///
    /// An Unconfirmed-Request to one of
    /// [`TransportPort::local_broadcast_destinations`] goes out as the
    /// link's local broadcast, through `send_broadcast`, since B/IP refuses a
    /// unicast to a group address (#1677). On B/IP those are the groups at
    /// the link's own UDP port. The layer takes that rule when it is made
    /// and when it starts, so no send calls back into the transport; a
    /// transport that gives no rule has every send go as addressed.
    ///
    /// Anything else goes to the transport as given, and B/IP refuses a
    /// group address there. This layer doesn't refuse it itself, so that is
    /// the caller's to keep: `BACnetClient`, the endpoint's egress and its
    /// requester, which take caller-chosen MACs, refuse anything but an
    /// Unconfirmed-Request to one (#1479). The server refuses one for the
    /// confirmed requests it starts itself, to event recipients, Channel
    /// and Command targets, audit recipients and bound devices (#1493). Its
    /// replies and COV notifications go to the source a request came from,
    /// and the server, the client and the endpoint ignore a confirmed
    /// request from a group source MAC (#1504), so none of those goes to a
    /// group on this link. A routed request's SADR is another network's to
    /// judge: a reply to a group SADR is dropped only where the final router
    /// is a [`BACnetRouter`](crate::router::BACnetRouter).
    pub async fn send_apdu(
        &self,
        apdu: &[u8],
        destination_mac: &[u8],
        expecting_reply: bool,
        priority: NetworkPriority,
    ) -> Result<(), Error> {
        self.send_apdu_with_data_attributes(apdu, destination_mac, expecting_reply, priority, &[])
            .await
    }

    /// Send an APDU with data attributes to a specific local destination.
    pub async fn send_apdu_with_data_attributes(
        &self,
        apdu: &[u8],
        destination_mac: &[u8],
        expecting_reply: bool,
        priority: NetworkPriority,
        data_attributes: &[DataAttribute],
    ) -> Result<(), Error> {
        // Straight from the borrowed APDU, no copy of it first (#1713).
        let buf = crate::response_route::encode_checked_response_npdu(
            apdu,
            None,
            expecting_reply,
            priority,
        )?;

        // An Unconfirmed-Request for a group address of this link is this
        // link's local broadcast, so it goes out as one: B/IP refuses a
        // unicast to a group address (#1677). The rule is the one the
        // transport gave at start, so no send calls back into it.
        let unconfirmed = apdu
            .first()
            .is_some_and(|&first| PduType::from_raw(first >> 4) == PduType::UNCONFIRMED_REQUEST);
        if unconfirmed && self.local_broadcasts.contains(destination_mac) {
            return self
                .transport
                .send_broadcast_with_data_attributes(&buf, data_attributes)
                .await;
        }
        self.transport
            .send_unicast_with_data_attributes(&buf, destination_mac, data_attributes)
            .await
    }

    /// Broadcast an APDU on the local network. Anything but an
    /// Unconfirmed-Request APDU is refused before it is sent (Clause 6.3,
    /// #1479).
    pub async fn broadcast_apdu(
        &self,
        apdu: &[u8],
        expecting_reply: bool,
        priority: NetworkPriority,
    ) -> Result<(), Error> {
        self.broadcast_apdu_with_data_attributes(apdu, expecting_reply, priority, &[])
            .await
    }

    /// Broadcast an APDU with data attributes on the local network.
    pub async fn broadcast_apdu_with_data_attributes(
        &self,
        apdu: &[u8],
        expecting_reply: bool,
        priority: NetworkPriority,
        data_attributes: &[DataAttribute],
    ) -> Result<(), Error> {
        check_broadcast_apdu(apdu, "a local broadcast")?;
        let npdu = Npdu {
            expecting_reply,
            priority,
            destination: None,
            source: None,
            payload: Bytes::copy_from_slice(apdu),
            ..Npdu::default()
        };

        let mut buf = BytesMut::with_capacity(2 + apdu.len());
        encode_npdu(&mut buf, &npdu)?;

        self.transport
            .send_broadcast_with_data_attributes(&buf, data_attributes)
            .await
    }

    /// Broadcast an APDU globally (DNET=0xFFFF, hop_count=255).
    ///
    /// Unlike `broadcast_apdu()` which only reaches the local subnet, this
    /// sets DNET=0xFFFF so routers will forward to all reachable networks.
    /// Anything but an Unconfirmed-Request APDU is refused before it is sent
    /// (Clause 6.3, #1479).
    pub async fn broadcast_global_apdu(
        &self,
        apdu: &[u8],
        expecting_reply: bool,
        priority: NetworkPriority,
    ) -> Result<(), Error> {
        self.broadcast_global_apdu_with_data_attributes(apdu, expecting_reply, priority, &[])
            .await
    }

    /// Broadcast an APDU globally with data attributes (DNET=0xFFFF, hop_count=255).
    pub async fn broadcast_global_apdu_with_data_attributes(
        &self,
        apdu: &[u8],
        expecting_reply: bool,
        priority: NetworkPriority,
        data_attributes: &[DataAttribute],
    ) -> Result<(), Error> {
        check_broadcast_apdu(apdu, "a global broadcast")?;
        let npdu = Npdu {
            expecting_reply,
            priority,
            destination: Some(NpduAddress {
                network: 0xFFFF,
                mac_address: MacAddr::new(),
            }),
            source: None,
            hop_count: 255,
            payload: Bytes::copy_from_slice(apdu),
            ..Npdu::default()
        };

        let mut buf = BytesMut::with_capacity(8 + apdu.len());
        encode_npdu(&mut buf, &npdu)?;
        self.transport
            .send_broadcast_with_data_attributes(&buf, data_attributes)
            .await
    }

    /// Broadcast an APDU to a specific remote network via routers.
    ///
    /// Like `broadcast_global_apdu()` but targets a single network number
    /// instead of all networks (DNET=0xFFFF). Refuses `dest_network` 0 and
    /// 0xFFFF, and anything but an Unconfirmed-Request APDU (Clause 6.3,
    /// #1479), without sending anything. This is the one send that puts a
    /// remote network's broadcast on the link's broadcast MAC;
    /// [`Self::send_apdu_routed`] with an empty `dest_mac` sends it to a
    /// known router instead.
    pub async fn broadcast_to_network(
        &self,
        apdu: &[u8],
        dest_network: u16,
        expecting_reply: bool,
        priority: NetworkPriority,
    ) -> Result<(), Error> {
        self.broadcast_to_network_with_data_attributes(
            apdu,
            dest_network,
            expecting_reply,
            priority,
            &[],
        )
        .await
    }

    /// Broadcast an APDU with data attributes to a specific remote network via routers.
    pub async fn broadcast_to_network_with_data_attributes(
        &self,
        apdu: &[u8],
        dest_network: u16,
        expecting_reply: bool,
        priority: NetworkPriority,
        data_attributes: &[DataAttribute],
    ) -> Result<(), Error> {
        check_destination(dest_network, "use broadcast_apdu for the local network")?;
        check_broadcast_apdu(apdu, "a broadcast to one network")?;
        let npdu = Npdu {
            expecting_reply,
            priority,
            destination: Some(NpduAddress {
                network: dest_network,
                mac_address: MacAddr::new(),
            }),
            source: None,
            hop_count: 255,
            payload: Bytes::copy_from_slice(apdu),
            ..Npdu::default()
        };

        let mut buf = BytesMut::with_capacity(8 + apdu.len());
        encode_npdu(&mut buf, &npdu)?;
        self.transport
            .send_broadcast_with_data_attributes(&buf, data_attributes)
            .await
    }

    /// Send an APDU to a remote device through a local router.
    ///
    /// The NPDU is sent via unicast to `router_mac` (the next-hop router on
    /// the local network), but the NPDU header addresses the final destination
    /// with `dest_network` / `dest_mac`. Refuses `dest_network` 0 and 0xFFFF
    /// without sending anything; this form and
    /// [`Self::send_apdu_routed_via_local_broadcast`] share that check. A
    /// global broadcast unicast to one router would reach only that router's
    /// networks, so it goes through [`Self::broadcast_global_apdu`] instead.
    ///
    /// An empty `dest_mac` (DLEN 0) asks the router to broadcast on
    /// `dest_network`, a remote broadcast, so it carries only an
    /// Unconfirmed-Request APDU (Clause 6.3); anything else is refused
    /// without sending anything (#1479).
    pub async fn send_apdu_routed(
        &self,
        apdu: &[u8],
        dest_network: u16,
        dest_mac: &[u8],
        router_mac: &[u8],
        expecting_reply: bool,
        priority: NetworkPriority,
    ) -> Result<(), Error> {
        self.send_apdu_routed_with_data_attributes(
            apdu,
            RoutedTarget {
                network: dest_network,
                mac: dest_mac,
                router_mac,
            },
            expecting_reply,
            priority,
            &[],
        )
        .await
    }

    /// Send an APDU with data attributes to a remote device through a local router.
    pub async fn send_apdu_routed_with_data_attributes(
        &self,
        apdu: &[u8],
        target: RoutedTarget<'_>,
        expecting_reply: bool,
        priority: NetworkPriority,
        data_attributes: &[DataAttribute],
    ) -> Result<(), Error> {
        check_destination(target.network, "use send_apdu for a local device")?;
        if target.mac.is_empty() {
            check_broadcast_apdu(
                apdu,
                "a routed send with an empty dest_mac (a remote broadcast)",
            )?;
        }
        let buf = Self::encode_routed_npdu_buf(
            apdu,
            target.network,
            target.mac,
            expecting_reply,
            priority,
        )?;
        self.transport
            .send_unicast_with_data_attributes(&buf, target.router_mac, data_attributes)
            .await
    }

    /// Send a routed APDU with a broadcast link DA, for when the next-hop
    /// router's MAC is unknown.
    ///
    /// Clause 6.5.3 normally addresses the frame to the MAC of the router
    /// that serves DNET, and falls back to a link broadcast while that
    /// router has not been located yet. DNET/DADR still select a single
    /// device in the NPDU. Clause 6.3 reserves broadcast network addresses
    /// for Unconfirmed-Request PDUs, but that limits the network-layer
    /// address, not the link DA: a link broadcast carrying a one-device
    /// DNET/DADR is fine for any PDU type, so this send form stays
    /// conformant. DNET 0xFFFF would make it a global broadcast, which this
    /// form doesn't mean, so it refuses that DNET as [`Self::send_apdu_routed`]
    /// does. An empty `dest_mac` would make it a broadcast on `dest_network`,
    /// which [`Self::broadcast_to_network`] already sends, so this form
    /// refuses it too and points there, leaving one way to send a remote
    /// broadcast with the link's broadcast MAC (#1479).
    pub async fn send_apdu_routed_via_local_broadcast(
        &self,
        apdu: &[u8],
        dest_network: u16,
        dest_mac: &[u8],
        expecting_reply: bool,
        priority: NetworkPriority,
    ) -> Result<(), Error> {
        self.send_apdu_routed_via_local_broadcast_with_data_attributes(
            apdu,
            dest_network,
            dest_mac,
            expecting_reply,
            priority,
            &[],
        )
        .await
    }

    /// Send a routed APDU with data attributes and a broadcast link DA.
    pub async fn send_apdu_routed_via_local_broadcast_with_data_attributes(
        &self,
        apdu: &[u8],
        dest_network: u16,
        dest_mac: &[u8],
        expecting_reply: bool,
        priority: NetworkPriority,
        data_attributes: &[DataAttribute],
    ) -> Result<(), Error> {
        check_destination(dest_network, "use send_apdu for a local device")?;
        if dest_mac.is_empty() {
            return Err(Error::Encoding(format!(
                "an empty dest_mac addresses the broadcast on network {dest_network}, \
                 not one device; use broadcast_to_network for a remote broadcast"
            )));
        }
        let buf =
            Self::encode_routed_npdu_buf(apdu, dest_network, dest_mac, expecting_reply, priority)?;
        self.transport
            .send_broadcast_with_data_attributes(&buf, data_attributes)
            .await
    }

    /// Attach selected-object protection before starting the owned transport.
    #[doc(hidden)]
    pub fn retain_network_port_lease_internal(&mut self, lease: Arc<()>) -> Result<(), Error> {
        self.transport.retain_network_port_lease_internal(lease)
    }

    /// Access the underlying transport.
    ///
    /// Useful for transport-specific operations like BBMD registration
    /// after the network layer has been started.
    pub fn transport(&self) -> &T {
        &self.transport
    }

    /// Get the transport's local MAC address.
    pub fn local_mac(&self) -> &[u8] {
        self.transport.local_mac()
    }

    /// The number of the network this layer's one port is attached to, as
    /// the owner of its Number controls last published it, and unknown until
    /// then. The full server and the standalone client publish into their own
    /// layers internally; a stack built directly on this layer publishes its
    /// own, and can clone the handle to publish from another task.
    pub fn local_network_number(&self) -> &LocalNetworkNumber {
        &self.local_network_number
    }

    /// Sequence assigned to the most recent opted-in control admission attempt.
    ///
    /// The value advances before admission, including Full and Closed drops,
    /// but not for controls discarded without an enabled stream or after its
    /// first Closed drop. See the [receive-queue contract](self#receive-queue-admission).
    /// At counter exhaustion it remains at `u64::MAX`, which makes sequence-based
    /// consumers fail closed rather than accepting an alias.
    pub fn network_control_ingress_sequence(&self) -> u64 {
        self.network_control_ingress_sequence.load(Ordering::SeqCst)
    }

    /// Inbound NPDUs dropped since this layer was created because an address
    /// in them was past [`MAX_MAC_LEN`](bacnet_npdu::MAX_MAC_LEN): their DLEN or SLEN
    /// (#1141), or the link-layer source MAC the transport reported (#1198).
    /// Saturates at `u64::MAX`.
    ///
    /// Only routers answer an NPDU with Reject-Message-To-Network (Clause
    /// 6.6.3.5), so this non-router layer discards the NPDU and counts it.
    /// The NPDU reaches neither the APDU nor the control receiver, so it
    /// never shows up in their admission counters. No built-in transport
    /// reports a source MAC that long, so a nonzero count with well-formed
    /// traffic points at a custom transport.
    pub fn address_length_drops(&self) -> u64 {
        self.counters().address_length_drops
    }

    /// Inbound NPDUs dropped since this layer was created because their DNET
    /// was 0xFFFF and they also carried a DADR (#1379). Saturates at
    /// `u64::MAX`.
    ///
    /// DNET 0xFFFF already names every device on every network (Clauses
    /// 6.2.2 and 6.3.2), so a DADR beside it contradicts it. Such an NPDU
    /// reaches neither the APDU nor the control receiver, so it never shows
    /// up in their admission counters. It is counted here rather than in
    /// [`Self::address_length_drops`], whose addresses are too long to hold:
    /// this one is well formed, only contradictory, and points at the peer
    /// that sent it.
    pub fn global_broadcast_dadr_drops(&self) -> u64 {
        self.counters().global_broadcast_dadr_drops
    }

    /// Inbound NPDUs dropped since this layer was created because they were
    /// addressed to a broadcast, global (DNET 0xFFFF) or remote (a DNET with
    /// DLEN 0), and carried an APDU other than an Unconfirmed-Request
    /// (#1491). Saturates at `u64::MAX`.
    ///
    /// Only an Unconfirmed-Request may go to a broadcast network address
    /// (Clause 6.3), so this layer hands no other APDU sent that way to the
    /// application, and the NPDU never shows up in the receivers' admission
    /// counters. A remote broadcast, which a non-router discards anyway, is
    /// counted here too when its APDU is of another type. The PDU type comes
    /// from the first APDU octet, with no decode; network messages, and an
    /// APDU sent with no DNET, are not affected.
    pub fn broadcast_pdu_type_drops(&self) -> u64 {
        self.counters().broadcast_pdu_type_drops
    }

    /// What this layer has dropped on receipt since it was created, by
    /// reason: the totals of the receive rules it shares with the router's
    /// engine (`bacnet-router-core`'s `node`, #1795). Reading them takes no
    /// lock.
    pub fn counters(&self) -> Counters {
        self.node_totals.load()
    }

    /// Encode an APDU into an NPDU whose destination is `dest_network` /
    /// `dest_mac`, ready for whichever link send the caller chooses. Each
    /// caller checks the DNET, and what an empty `dest_mac` may carry, first.
    fn encode_routed_npdu_buf(
        apdu: &[u8],
        dest_network: u16,
        dest_mac: &[u8],
        expecting_reply: bool,
        priority: NetworkPriority,
    ) -> Result<BytesMut, Error> {
        let npdu = Npdu {
            expecting_reply,
            priority,
            destination: Some(NpduAddress {
                network: dest_network,
                mac_address: MacAddr::from_slice(dest_mac),
            }),
            source: None,
            hop_count: 255,
            payload: Bytes::copy_from_slice(apdu),
            ..Npdu::default()
        };
        let mut buf = BytesMut::with_capacity(8 + dest_mac.len() + apdu.len());
        encode_npdu(&mut buf, &npdu)?;
        Ok(buf)
    }

    /// Stop the network layer and underlying transport.
    ///
    /// After start, aborts and awaits dispatch before stopping the transport;
    /// it does not wait for consumers, even with full APDU/control queues.
    /// An [`ApduReceiver`] yields what the transport had queued, then `None`.
    /// Queued items remain drainable, retaining their reply senders until
    /// consumed or the receiver is dropped. Stop does not count as an admission
    /// drop or clear depth/high-water/drop totals. See the
    /// [receive-queue contract](self#receive-queue-admission).
    pub async fn stop(&mut self) -> Result<(), Error> {
        self.seal_responses();
        // The raw receiver drains what the transport queued, then ends.
        ingress::close_queue(&self.receiver_queue);
        if let Some(task) = self.abort_dispatch_task() {
            let _ = task.await;
        }
        self.transport.stop().await
    }
}

/// Advance a control ingress sequence, saturating at `u64::MAX`. Shared by
/// [`NetworkLayer`] and the router's control receiver.
pub(crate) fn next_ingress_sequence(sequence: &AtomicU64) -> u64 {
    let previous = sequence
        .try_update(Ordering::SeqCst, Ordering::SeqCst, |value| {
            value.checked_add(1)
        })
        .unwrap_or(u64::MAX);
    previous.saturating_add(1)
}

impl<T: TransportPort> NetworkLayer<T> {
    fn abort_dispatch_task(&mut self) -> Option<JoinHandle<()>> {
        let task = self.dispatch_task.take()?;
        task.abort();
        Some(task)
    }
}

impl<T: TransportPort> Drop for NetworkLayer<T> {
    fn drop(&mut self) {
        self.response_scope.seal();
        ingress::close_queue(&self.receiver_queue);
        let _ = self.abort_dispatch_task();
        self.transport.abort();
    }
}

#[cfg(test)]
#[path = "issuance_tests.rs"]
mod issuance_tests;

#[cfg(test)]
mod tests {
    use super::*;
    use bacnet_encoding::npdu::decode_npdu;
    use bacnet_transport::bip::BipTransport;
    use bacnet_transport::sc::{LoopbackWebSocket, ScTransport, WebSocketPort};
    use bacnet_transport::sc_frame::{
        ScFunction, ScMessage, Vmac, decode_sc_message, encode_sc_message,
    };
    use std::net::Ipv4Addr;
    use tokio::time::{Duration, timeout};

    async fn sc_hub_accept(ws_hub: &LoopbackWebSocket, hub_vmac: Vmac) {
        let data = ws_hub.recv().await.unwrap();
        let req = decode_sc_message(&data).unwrap();
        assert_eq!(req.function, ScFunction::ConnectRequest);

        let mut accept_payload = Vec::with_capacity(26);
        accept_payload.extend_from_slice(&hub_vmac);
        accept_payload.extend_from_slice(&[0x33; 16]);
        accept_payload.extend_from_slice(&1476u16.to_be_bytes());
        accept_payload.extend_from_slice(&1476u16.to_be_bytes());

        let accept = ScMessage {
            function: ScFunction::ConnectAccept,
            message_id: req.message_id,
            originating_vmac: None,
            destination_vmac: None,
            dest_options: Vec::new(),
            data_options: Vec::new(),
            payload: Bytes::from(accept_payload),
        };
        let mut buf = BytesMut::new();
        encode_sc_message(&mut buf, &accept);
        ws_hub.send(&buf).await.unwrap();
    }

    async fn assert_sc_socket_closed_after_drop(ws_hub: &LoopbackWebSocket, context: &str) {
        timeout(Duration::from_secs(1), async {
            while let Ok(data) = ws_hub.recv().await {
                let msg = decode_sc_message(&data).unwrap();
                assert_ne!(
                    msg.function,
                    ScFunction::HeartbeatAck,
                    "{context} must not leave SC answering heartbeats"
                );
            }
        })
        .await
        .unwrap_or_else(|_| panic!("{context} did not close the SC WebSocket"));

        let heartbeat = ScMessage {
            function: ScFunction::HeartbeatRequest,
            message_id: 0x66,
            originating_vmac: None,
            destination_vmac: None,
            dest_options: Vec::new(),
            data_options: Vec::new(),
            payload: Bytes::new(),
        };
        let mut buf = BytesMut::new();
        encode_sc_message(&mut buf, &heartbeat);
        assert!(
            ws_hub.send(&buf).await.is_err(),
            "{context} must reject post-drop Heartbeat-Request on the closed socket"
        );
    }

    #[tokio::test]
    async fn network_layer_drop_releases_sc_transport_socket() {
        let (ws_client, ws_hub) = LoopbackWebSocket::pair();
        let hub_vmac = [0x10; 6];
        let mut net =
            NetworkLayer::new(ScTransport::new(ws_client, [0x01; 6]).with_device_uuid([1; 16]));

        let hub_accept_task = tokio::spawn(async move {
            sc_hub_accept(&ws_hub, hub_vmac).await;
            ws_hub
        });

        let _rx = net.start().await.unwrap();
        let ws_hub = hub_accept_task.await.unwrap();

        drop(net);

        assert_sc_socket_closed_after_drop(&ws_hub, "dropped NetworkLayer").await;
    }

    #[tokio::test]
    async fn send_apdu_data_attributes_reach_sc_data_options() {
        let (ws_client, ws_hub) = LoopbackWebSocket::pair();
        let hub_vmac = [0x10; 6];
        let dest_vmac: Vmac = [0x02, 0x03, 0x04, 0x05, 0x06, 0x07];
        let mut net =
            NetworkLayer::new(ScTransport::new(ws_client, [0x01; 6]).with_device_uuid([1; 16]));
        let data_attributes = vec![
            DataAttribute {
                option_type: 1,
                must_understand: true,
                data: Vec::new(),
            },
            DataAttribute {
                option_type: 31,
                must_understand: false,
                data: vec![0x12, 0x34, 0x56],
            },
        ];

        let hub_accept_task = tokio::spawn(async move {
            sc_hub_accept(&ws_hub, hub_vmac).await;
            ws_hub
        });

        let _rx = net.start().await.unwrap();
        let ws_hub = hub_accept_task.await.unwrap();

        let apdu = Bytes::from_static(&[0x10, 0x08]);
        net.send_apdu_with_data_attributes(
            &apdu,
            &dest_vmac,
            false,
            NetworkPriority::Normal,
            &data_attributes,
        )
        .await
        .unwrap();

        let data = ws_hub.recv().await.unwrap();
        let msg = decode_sc_message(&data).unwrap();
        assert_eq!(msg.function, ScFunction::EncapsulatedNpdu);
        assert_eq!(msg.destination_vmac, Some(dest_vmac));
        assert_eq!(msg.data_options.len(), 2);
        assert_eq!(msg.data_options[0].option_type, 1);
        assert!(msg.data_options[0].must_understand);
        assert_eq!(msg.data_options[1].option_type, 31);
        assert!(!msg.data_options[1].must_understand);
        assert_eq!(msg.data_options[1].data, vec![0x12, 0x34, 0x56]);

        let npdu = decode_npdu(msg.payload).unwrap();
        assert_eq!(npdu.payload, apdu);
        assert!(!npdu.expecting_reply);
        assert_eq!(npdu.priority, NetworkPriority::Normal);

        net.stop().await.unwrap();
    }

    #[tokio::test]
    async fn end_to_end_who_is() {
        use bacnet_encoding::apdu::{Apdu, UnconfirmedRequest, decode_apdu, encode_apdu};
        use bacnet_types::enums::UnconfirmedServiceChoice;

        let transport_a = BipTransport::new(Ipv4Addr::LOCALHOST, 0, Ipv4Addr::BROADCAST);
        let transport_b = BipTransport::new(Ipv4Addr::LOCALHOST, 0, Ipv4Addr::BROADCAST);

        let mut net_a = NetworkLayer::new(transport_a);
        let mut net_b = NetworkLayer::new(transport_b);

        let _rx_a = net_a.start().await.unwrap();
        let mut rx_b = net_b.start().await.unwrap();

        let who_is_apdu = Apdu::UnconfirmedRequest(UnconfirmedRequest {
            service_choice: UnconfirmedServiceChoice::WHO_IS,
            service_request: Bytes::new(),
        });
        let mut apdu_buf = BytesMut::new();
        encode_apdu(&mut apdu_buf, &who_is_apdu).expect("valid APDU encoding");

        net_a
            .send_apdu(&apdu_buf, net_b.local_mac(), false, NetworkPriority::Normal)
            .await
            .unwrap();

        let received = timeout(Duration::from_secs(2), rx_b.recv())
            .await
            .expect("Timed out waiting for APDU")
            .expect("Channel closed");

        let decoded_apdu = decode_apdu(received.apdu.clone()).unwrap();
        match decoded_apdu {
            Apdu::UnconfirmedRequest(req) => {
                assert_eq!(req.service_choice, UnconfirmedServiceChoice::WHO_IS);
                assert!(req.service_request.is_empty());
            }
            other => panic!("Expected UnconfirmedRequest, got {:?}", other),
        }

        net_a.stop().await.unwrap();
        net_b.stop().await.unwrap();
    }

    #[test]
    fn transport_accessor() {
        let transport = BipTransport::new(Ipv4Addr::LOCALHOST, 0, Ipv4Addr::BROADCAST);
        let net = NetworkLayer::new(transport);
        let mac = net.transport().local_mac();
        assert_eq!(mac.len(), 6);
    }
}

#[cfg(test)]
#[path = "layer_delivery_tests.rs"]
mod delivery_tests;

#[cfg(test)]
#[path = "layer_admission_tests.rs"]
mod admission_tests;

#[cfg(test)]
#[path = "layer_sc_data_options_tests.rs"]
mod sc_data_options_tests;

#[cfg(test)]
#[path = "broadcast_pdu_tests.rs"]
mod broadcast_pdu_tests;

#[cfg(test)]
#[path = "layer_group_send_tests.rs"]
mod group_send_tests;