Large spec review around objects, services, client/server.

Author: jscott3201

crates/bacnet-client/src/client.rs

@@ -15,8 +15,8 @@ use tokio::time::{timeout, Duration};
 use tracing::{debug, warn};
 
 use bacnet_encoding::apdu::{
-    self, encode_apdu, Apdu, ConfirmedRequest as ConfirmedRequestPdu, SegmentAck as SegmentAckPdu,
-    SimpleAck,
+    self, encode_apdu, AbortPdu, Apdu, ConfirmedRequest as ConfirmedRequestPdu,
+    SegmentAck as SegmentAckPdu, SimpleAck,
 };
 use bacnet_encoding::npdu::NpduAddress;
 use bacnet_network::layer::NetworkLayer;
@@ -233,6 +233,10 @@ struct SegmentedReceiveState {
     expected_next_seq: u8,
     /// Timestamp of last received segment (for reaping stale sessions).
     last_activity: Instant,
+    /// Window position counter for per-window SegmentAck (Clause 5.2.2).
+    window_position: u8,
+    /// Proposed window size from the server.
+    proposed_window_size: u8,
 }
 
 /// Timeout for idle segmented reassembly sessions.
@@ -536,15 +540,47 @@ impl<T: TransportPort + 'static> BACnetClient<T> {
         let seg_ack_senders: Arc<Mutex<HashMap<SegKey, mpsc::Sender<SegmentAckPdu>>>> =
             Arc::new(Mutex::new(HashMap::new()));
         let seg_ack_senders_dispatch = Arc::clone(&seg_ack_senders);
+        let segmented_response_accepted = config.segmented_response_accepted;
 
         let dispatch_task = tokio::spawn(async move {
             let mut seg_state: HashMap<SegKey, SegmentedReceiveState> = HashMap::new();
+            let mut last_device_purge = Instant::now();
+            const DEVICE_PURGE_INTERVAL: Duration = Duration::from_secs(300);
+            const DEVICE_MAX_AGE: Duration = Duration::from_secs(600);
 
             while let Some(received) = apdu_rx.recv().await {
                 let now = Instant::now();
-                seg_state.retain(|_key, state| {
-                    now.duration_since(state.last_activity) < SEG_RECEIVER_TIMEOUT
-                });
+
+                // Periodically purge stale device table entries
+                if now.duration_since(last_device_purge) >= DEVICE_PURGE_INTERVAL {
+                    device_table_dispatch
+                        .lock()
+                        .await
+                        .purge_stale(DEVICE_MAX_AGE);
+                    last_device_purge = now;
+                }
+                // Reap stale segmented sessions and send Abort to the server
+                let stale_keys: Vec<SegKey> = seg_state
+                    .iter()
+                    .filter(|(_, state)| {
+                        now.duration_since(state.last_activity) >= SEG_RECEIVER_TIMEOUT
+                    })
+                    .map(|(key, _)| key.clone())
+                    .collect();
+                for key in &stale_keys {
+                    if let Some(_state) = seg_state.remove(key) {
+                        let abort = Apdu::Abort(AbortPdu {
+                            sent_by_server: false,
+                            invoke_id: key.1,
+                            abort_reason: bacnet_types::enums::AbortReason::TSM_TIMEOUT,
+                        });
+                        let mut buf = BytesMut::with_capacity(4);
+                        encode_apdu(&mut buf, &abort);
+                        let _ = network_dispatch
+                            .send_apdu(&buf, &key.0, false, NetworkPriority::NORMAL)
+                            .await;
+                    }
+                }
 
                 match apdu::decode_apdu(received.apdu.clone()) {
                     Ok(decoded) => {
@@ -558,6 +594,7 @@ impl<T: TransportPort + 'static> BACnetClient<T> {
                             &received.source_mac,
                             &received.source_network,
                             decoded,
+                            segmented_response_accepted,
                         )
                         .await;
                     }
@@ -592,6 +629,7 @@ impl<T: TransportPort + 'static> BACnetClient<T> {
         source_mac: &[u8],
         source_network: &Option<NpduAddress>,
         apdu: Apdu,
+        segmented_response_accepted: bool,
     ) {
         match apdu {
             Apdu::SimpleAck(ack) => {
@@ -601,8 +639,15 @@ impl<T: TransportPort + 'static> BACnetClient<T> {
             }
             Apdu::ComplexAck(ack) => {
                 if ack.segmented {
-                    Self::handle_segmented_complex_ack(tsm, network, seg_state, source_mac, ack)
-                        .await;
+                    Self::handle_segmented_complex_ack(
+                        tsm,
+                        network,
+                        seg_state,
+                        source_mac,
+                        ack,
+                        segmented_response_accepted,
+                    )
+                    .await;
                 } else {
                     debug!(invoke_id = ack.invoke_id, "Received ComplexAck");
                     let mut tsm = tsm.lock().await;
@@ -760,6 +805,7 @@ impl<T: TransportPort + 'static> BACnetClient<T> {
         seg_state: &mut HashMap<SegKey, SegmentedReceiveState>,
         source_mac: &[u8],
         ack: bacnet_encoding::apdu::ComplexAck,
+        segmented_response_accepted: bool,
     ) {
         let seq = ack.sequence_number.unwrap_or(0);
         let key = (MacAddr::from_slice(source_mac), ack.invoke_id);
@@ -771,6 +817,21 @@ impl<T: TransportPort + 'static> BACnetClient<T> {
             "Received segmented ComplexAck"
         );
 
+        // If client doesn't support segmented reception, send Abort per Clause 5.4.4.2
+        if !segmented_response_accepted {
+            let abort = Apdu::Abort(AbortPdu {
+                sent_by_server: false,
+                invoke_id: ack.invoke_id,
+                abort_reason: bacnet_types::enums::AbortReason::SEGMENTATION_NOT_SUPPORTED,
+            });
+            let mut buf = BytesMut::with_capacity(4);
+            encode_apdu(&mut buf, &abort);
+            let _ = network
+                .send_apdu(&buf, source_mac, false, NetworkPriority::NORMAL)
+                .await;
+            return;
+        }
+
         const MAX_CONCURRENT_SEG_SESSIONS: usize = 64;
         if !seg_state.contains_key(&key) && seg_state.len() >= MAX_CONCURRENT_SEG_SESSIONS {
             warn!(
@@ -781,37 +842,55 @@ impl<T: TransportPort + 'static> BACnetClient<T> {
             return;
         }
 
+        let proposed_ws = ack.proposed_window_size.unwrap_or(1);
         let state = seg_state
             .entry(key.clone())
             .or_insert_with(|| SegmentedReceiveState {
                 receiver: SegmentReceiver::new(),
                 expected_next_seq: 0,
                 last_activity: Instant::now(),
+                window_position: 0,
+                proposed_window_size: proposed_ws,
             });
 
         state.last_activity = Instant::now();
 
         if seq != state.expected_next_seq {
-            warn!(
-                invoke_id = ack.invoke_id,
-                expected = state.expected_next_seq,
-                received = seq,
-                "Segment gap detected, sending negative SegmentAck"
-            );
+            // Check for duplicate (already received) vs true gap
+            if seq < state.expected_next_seq {
+                // Duplicate segment — discard silently and ack
+                debug!(
+                    invoke_id = ack.invoke_id,
+                    seq, "Discarding duplicate segment"
+                );
+            } else {
+                // True gap — send negative SegmentAck with last correctly received seq
+                warn!(
+                    invoke_id = ack.invoke_id,
+                    expected = state.expected_next_seq,
+                    received = seq,
+                    "Segment gap detected, sending negative SegmentAck"
+                );
+            }
             let neg_ack = Apdu::SegmentAck(SegmentAckPdu {
-                negative_ack: true,
+                negative_ack: seq >= state.expected_next_seq,
                 sent_by_server: false,
                 invoke_id: ack.invoke_id,
-                sequence_number: state.expected_next_seq,
-                actual_window_size: ack.proposed_window_size.unwrap_or(1),
+                // Spec: sequence_number = last correctly received sequence number
+                sequence_number: if state.expected_next_seq > 0 {
+                    state.expected_next_seq.wrapping_sub(1)
+                } else {
+                    0
+                },
+                actual_window_size: proposed_ws,
             });
             let mut buf = BytesMut::with_capacity(4);
             encode_apdu(&mut buf, &neg_ack);
             if let Err(e) = network
                 .send_apdu(&buf, source_mac, false, NetworkPriority::NORMAL)
                 .await
             {
-                warn!(error = %e, "Failed to send negative SegmentAck");
+                warn!(error = %e, "Failed to send SegmentAck");
             }
             return;
         }
@@ -821,21 +900,28 @@ impl<T: TransportPort + 'static> BACnetClient<T> {
             return;
         }
         state.expected_next_seq = seq.wrapping_add(1);
+        state.window_position += 1;
 
-        let seg_ack = Apdu::SegmentAck(SegmentAckPdu {
-            negative_ack: false,
-            sent_by_server: false,
-            invoke_id: ack.invoke_id,
-            sequence_number: seq,
-            actual_window_size: ack.proposed_window_size.unwrap_or(1),
-        });
-        let mut buf = BytesMut::with_capacity(4);
-        encode_apdu(&mut buf, &seg_ack);
-        if let Err(e) = network
-            .send_apdu(&buf, source_mac, false, NetworkPriority::NORMAL)
-            .await
-        {
-            warn!(error = %e, "Failed to send SegmentAck");
+        // Per-window SegmentAck: only ack at window boundary or final segment (Clause 5.2.2)
+        let should_ack = !ack.more_follows || state.window_position >= state.proposed_window_size;
+
+        if should_ack {
+            state.window_position = 0;
+            let seg_ack = Apdu::SegmentAck(SegmentAckPdu {
+                negative_ack: false,
+                sent_by_server: false,
+                invoke_id: ack.invoke_id,
+                sequence_number: seq,
+                actual_window_size: proposed_ws,
+            });
+            let mut buf = BytesMut::with_capacity(4);
+            encode_apdu(&mut buf, &seg_ack);
+            if let Err(e) = network
+                .send_apdu(&buf, source_mac, false, NetworkPriority::NORMAL)
+                .await
+            {
+                warn!(error = %e, "Failed to send SegmentAck");
+            }
         }
 
         if !ack.more_follows {
@@ -1100,6 +1186,7 @@ impl<T: TransportPort + 'static> BACnetClient<T> {
             self.seg_ack_senders.lock().await.insert(key, seg_ack_tx);
         }
 
+        // Tseg: use APDU timeout for now (configurable via apdu_timeout_ms)
         let timeout_duration = Duration::from_millis(self.config.apdu_timeout_ms);
         let max_ack_retries = self.config.apdu_retries;
         let mut window_size = self.config.proposed_window_size.max(1) as usize;

crates/bacnet-encoding/src/apdu.rs

@@ -81,6 +81,10 @@ fn decode_max_apdu(value: u8) -> u16 {
     if idx < MAX_APDU_DECODE.len() {
         MAX_APDU_DECODE[idx]
     } else {
+        tracing::warn!(
+            value = idx,
+            "Reserved max-APDU-length field value, defaulting to 1476"
+        );
         1476
     }
 }
@@ -468,12 +472,19 @@ fn decode_segment_ack(data: Bytes) -> Result<SegmentAck, Error> {
     }
 
     let byte0 = data[0];
+    let raw_window = data[3];
+    if raw_window == 0 || raw_window > 127 {
+        tracing::warn!(
+            actual_window_size = raw_window,
+            "SegmentAck window size out of range (1-127), clamping"
+        );
+    }
     Ok(SegmentAck {
         negative_ack: byte0 & 0x02 != 0,
         sent_by_server: byte0 & 0x01 != 0,
         invoke_id: data[1],
         sequence_number: data[2],
-        actual_window_size: data[3],
+        actual_window_size: raw_window.clamp(1, 127),
     })
 }
 

crates/bacnet-integration-tests/tests/server.rs

@@ -929,23 +929,27 @@ async fn server_handles_segmented_request() {
 
 /// DCC DISABLE (deprecated in 2020 spec) is rejected with SERVICE_REQUEST_DENIED.
 #[tokio::test]
-async fn dcc_disable_rejected_per_2020_spec() {
+async fn dcc_disable_sets_comm_state() {
     use bacnet_types::enums::EnableDisable;
 
     let mut server = make_server().await;
     let mut client = make_client().await;
     let server_mac = server.local_mac().to_vec();
 
-    // Clause 16.1.1.3.1: deprecated DISABLE shall be rejected
+    // Clause 16.1: DISABLE sets comm_state to 1 (per spec, all three values are supported)
     let result = client
         .device_communication_control(&server_mac, EnableDisable::DISABLE, None, None)
         .await;
-    assert!(
-        result.is_err(),
-        "DCC DISABLE should be rejected per 2020 spec"
-    );
+    assert!(result.is_ok(), "DCC DISABLE should succeed per Clause 16.1");
+
+    // Server should be in DISABLE state (1)
+    assert_eq!(server.comm_state(), 1);
 
-    // Server should remain in ENABLE state
+    // Re-enable should work (DCC is allowed even when disabled)
+    let result = client
+        .device_communication_control(&server_mac, EnableDisable::ENABLE, None, None)
+        .await;
+    assert!(result.is_ok(), "DCC ENABLE should succeed");
     assert_eq!(server.comm_state(), 0);
 
     client.stop().await.unwrap();