bacnet_client/client/
requests.rs

1use super::*;
2use bacnet_encoding::apdu::MINIMUM_MESSAGE_SIZE;
3
4/// Which term of Clause 5.2.1.2's minimum bound the transmittable length down.
5///
6/// Named separately so the error blames the right one. The checked value is a
7/// minimum over several terms, and the peer is only sometimes the binding one:
8/// BACnet/SC recomputes its own limit from the hub's Connect-Accept, so a
9/// transport can fall below the floor while the peer is perfectly conformant.
10enum LengthBoundedBy {
11    /// A length advertised by a discovered peer, from I-Am.
12    DiscoveredPeer(u16),
13    /// This client's own configured maximum.
14    LocalConfig(u16),
15    /// The data link, after any routed NPDU header.
16    Transport(u16),
17}
18
19impl core::fmt::Display for LengthBoundedBy {
20    fn fmt(&self, f: &mut core::fmt::Formatter<'_>) -> core::fmt::Result {
21        match self {
22            Self::DiscoveredPeer(v) => {
23                write!(f, "the peer's advertised Max APDU Length Accepted of {v}")
24            }
25            Self::LocalConfig(v) => write!(f, "this client's configured maximum of {v}"),
26            Self::Transport(v) => write!(f, "the transport's limit of {v}"),
27        }
28    }
29}
30
31/// Reject a maximum transmittable length no conformant device could accept.
32///
33/// Clause 5.2.1.2 derives this length as the smallest of the local capability,
34/// the internetwork limit, and "(c) the maximum APDU size accepted by the
35/// remote peer device, which must be at least 50 octets". Below that floor no
36/// conformant APDU can be formed at all. Clause 20.1.2.5 gives the same number
37/// a name, spelling the lowest max-APDU-length-accepted code `B'0000'` as "Up
38/// to MinimumMessageSize (50 octets)".
39///
40/// The check is a floor, not membership of the six values Clause 20.1.2.5
41/// encodes. A discovered peer's length comes from I-Am's `Max APDU Length
42/// Accepted`, an Unsigned octet count rather than the four-bit code, and
43/// Clause 20.1.2.5 notes the true value "may be larger than indicated in this
44/// parameter" — so 600 and 1500 are legitimate and must not be rejected.
45///
46/// Failing rather than clamping up to 50 keeps the client from inventing a
47/// capability the peer never claimed: a device advertising less than 50 is
48/// already non-conformant, and a typed error names that where a silent clamp
49/// would send it frames it said it cannot hold.
50fn check_transmittable_length(peer_or_local: LengthBoundedBy, transport: u16) -> Result<(), Error> {
51    let advertised = match peer_or_local {
52        LengthBoundedBy::DiscoveredPeer(v) | LengthBoundedBy::LocalConfig(v) => v,
53        LengthBoundedBy::Transport(v) => v,
54    };
55    let combined = advertised.min(transport);
56    if combined >= MINIMUM_MESSAGE_SIZE {
57        return Ok(());
58    }
59    let binding = if transport < advertised {
60        LengthBoundedBy::Transport(transport)
61    } else {
62        peer_or_local
63    };
64    Err(Error::Encoding(format!(
65        "maximum transmittable length {combined} is below the {MINIMUM_MESSAGE_SIZE}-octet \
66         MinimumMessageSize every BACnet device accepts (Clause 5.2.1.2); the binding limit is \
67         {binding}"
68    )))
69}
70
71impl<T: TransportPort + 'static> BACnetClient<T> {
72    /// Send a confirmed request and wait for the response.
73    ///
74    /// Returns the service response data (empty for SimpleAck). Automatically
75    /// uses segmented transfer when the payload exceeds the remote device's
76    /// max APDU length.
77    pub async fn confirmed_request(
78        &self,
79        destination_mac: &[u8],
80        service_choice: ConfirmedServiceChoice,
81        service_data: &[u8],
82    ) -> Result<Bytes, Error> {
83        self.confirmed_request_inner(
84            ConfirmedTarget::Local {
85                mac: destination_mac,
86            },
87            service_choice,
88            service_data,
89        )
90        .await
91    }
92
93    /// Send a confirmed request routed through a BACnet router.
94    ///
95    /// The NPDU is sent as a unicast to `router_mac` with DNET/DADR set so
96    /// the router forwards it to `dest_network`/`dest_mac`.
97    pub async fn confirmed_request_routed(
98        &self,
99        router_mac: &[u8],
100        dest_network: u16,
101        dest_mac: &[u8],
102        service_choice: ConfirmedServiceChoice,
103        service_data: &[u8],
104    ) -> Result<Bytes, Error> {
105        self.confirmed_request_inner(
106            ConfirmedTarget::Routed {
107                router_mac,
108                dest_network,
109                dest_mac,
110            },
111            service_choice,
112            service_data,
113        )
114        .await
115    }
116
117    pub(super) async fn confirmed_request_inner(
118        &self,
119        target: ConfirmedTarget<'_>,
120        service_choice: ConfirmedServiceChoice,
121        service_data: &[u8],
122    ) -> Result<Bytes, Error> {
123        // The lease serializes one active request per (immediate router, DNET)
124        // and remains live through every return path, including cancellation.
125        let path_lease = match target {
126            ConfirmedTarget::Local { .. } => None,
127            ConfirmedTarget::Routed {
128                router_mac,
129                dest_network,
130                ..
131            } => Some(
132                self.routed_path_limits
133                    .acquire(router_mac, dest_network)
134                    .await?,
135            ),
136        };
137        let (routed_path_max_apdu, routed_forwarded_npci_len) = match (path_lease.as_ref(), target)
138        {
139            (
140                Some(lease),
141                ConfirmedTarget::Routed {
142                    dest_mac,
143                    dest_network: _,
144                    router_mac: _,
145                },
146            ) => (
147                Some(lease.max_apdu(dest_mac.len(), self.local_mac.len())?),
148                Some(lease.forwarded_npci_len(dest_mac.len(), self.local_mac.len())?),
149            ),
150            _ => (None, None),
151        };
152        let transaction_peer = target.transaction_peer();
153        let tsm_mac = transaction_peer.tsm_mac;
154        let unsegmented_apdu_size = 4 + service_data.len();
155        let target_transport_max_apdu = self.target_transport_max_apdu_length(target);
156
157        let (peer_max_apdu, remote_max_segments, advertised, peer_segmentation) = {
158            let dt = self.device_table.lock().await;
159            // A routed peer is recorded under the router's MAC, so only the
160            // SNET/SADR of the NPDU that carried its I-Am identifies it
161            // (Clause 5.2.1.2 term (c) binds the peer's Max APDU Length
162            // Accepted regardless of how the peer is reached).
163            let (device, peer_segmentation) = match target {
164                ConfirmedTarget::Local { mac } => {
165                    (dt.get_by_mac(mac), dt.local_peer_segmentation(mac))
166                }
167                ConfirmedTarget::Routed {
168                    dest_network,
169                    dest_mac,
170                    ..
171                } => (
172                    dt.get_by_network_address(dest_network, dest_mac),
173                    dt.routed_peer_segmentation(dest_network, dest_mac),
174                ),
175            };
176            let max_apdu = device
177                .map(|d| u16::try_from(d.max_apdu_length).unwrap_or(u16::MAX))
178                .unwrap_or(self.config.max_apdu_length);
179            let max_seg = device.and_then(|d| d.max_segments_accepted);
180            let advertised = if device.is_some() {
181                LengthBoundedBy::DiscoveredPeer(max_apdu)
182            } else {
183                LengthBoundedBy::LocalConfig(max_apdu)
184            };
185            (max_apdu, max_seg, advertised, peer_segmentation)
186        };
187        check_transmittable_length(advertised, target_transport_max_apdu)?;
188        if routed_path_max_apdu.is_some_and(|path_max_apdu| {
189            path_max_apdu < MINIMUM_MESSAGE_SIZE
190                && path_max_apdu <= peer_max_apdu.min(target_transport_max_apdu)
191        }) {
192            let ConfirmedTarget::Routed { dest_network, .. } = target else {
193                unreachable!("a routed path limit exists only for a routed target")
194            };
195            return Err(Error::RoutedPathTooLong { dnet: dest_network });
196        }
197        let remote_max_apdu = peer_max_apdu
198            .min(target_transport_max_apdu)
199            .min(routed_path_max_apdu.unwrap_or(u16::MAX));
200        if unsegmented_apdu_size > remote_max_apdu as usize {
201            // Clause 12.11: a NO_SEGMENTATION or SEGMENTED_TRANSMIT peer
202            // accepts exactly one segment — only unsegmented requests — and
203            // Clause 18's SEGMENTATION_NOT_SUPPORTED abort is the certain
204            // outcome of sending anyway. Refuse locally when the capability
205            // is authoritative (I-Am or explicit configuration); a legacy
206            // placeholder row stays unknown and keeps today's behavior.
207            if let Some(crate::discovery::PeerSegmentation::Authoritative(capability)) =
208                peer_segmentation
209            {
210                if matches!(
211                    capability,
212                    bacnet_types::enums::Segmentation::NONE
213                        | bacnet_types::enums::Segmentation::TRANSMIT
214                ) {
215                    return Err(Error::Segmentation(format!(
216                        "request requires segmentation but the peer advertised \
217                         {capability:?} and cannot receive segmented requests"
218                    )));
219                }
220            }
221            return self
222                .segmented_confirmed_request(
223                    target,
224                    service_choice,
225                    service_data,
226                    remote_max_apdu,
227                    remote_max_segments,
228                    routed_forwarded_npci_len,
229                    path_lease.as_ref(),
230                )
231                .await;
232        }
233
234        let advertised_max_apdu = self.advertised_max_apdu_length_for_target(target)?;
235        let (invoke_id, registration) = {
236            let mut tsm = self.tsm.lock().await;
237            tsm.register_coordinated_transaction_with_progress(
238                tsm_mac.clone(),
239                transaction_peer.canonical,
240                service_choice,
241                false,
242            )
243            .map_err(|error| Error::Encoding(error.to_string()))?
244        };
245
246        let owner = registration.owner.clone();
247        let mut guard = TransactionGuard::new(
248            Arc::clone(&self.tsm),
249            self.cleanup_tx.clone(),
250            tsm_mac.clone(),
251            invoke_id,
252            owner.clone(),
253            None,
254        );
255
256        let pdu = Apdu::ConfirmedRequest(ConfirmedRequestPdu {
257            segmented: false,
258            more_follows: false,
259            segmented_response_accepted: self.config.segmented_response_accepted,
260            max_segments: self.config.max_segments,
261            max_apdu_length: advertised_max_apdu,
262            invoke_id,
263            sequence_number: None,
264            proposed_window_size: None,
265            service_choice,
266            service_request: Bytes::copy_from_slice(service_data),
267        });
268
269        let mut buf = BytesMut::with_capacity(6 + service_data.len());
270        encode_apdu(&mut buf, &pdu)?;
271
272        if let Some(forwarded_npci_len) = routed_forwarded_npci_len {
273            self.routed_path_limits.install_active(
274                target,
275                tsm_mac.clone(),
276                invoke_id,
277                owner.clone(),
278                forwarded_npci_len,
279                self.network.network_control_ingress_sequence(),
280            );
281        }
282        if !self.routed_path_limits.authorize_attempt(
283            target,
284            &tsm_mac,
285            invoke_id,
286            &owner,
287            buf.len(),
288        ) {
289            guard.mark_completed();
290            self.tsm
291                .lock()
292                .await
293                .cancel_transaction_for_owner(&tsm_mac, invoke_id, &owner);
294            self.enqueue_transaction_cleanup(&tsm_mac, invoke_id, &owner, false, None);
295            return Err(Error::Encoding(
296                "routed send lost its active path authorization".into(),
297            ));
298        }
299
300        if let Err(e) = self.send_confirmed_target_apdu(target, &buf).await {
301            guard.mark_completed();
302            let mut tsm = self.tsm.lock().await;
303            tsm.cancel_transaction_for_owner(&tsm_mac, invoke_id, &owner);
304            drop(tsm);
305            self.enqueue_transaction_cleanup(&tsm_mac, invoke_id, &owner, false, None);
306            return Err(e);
307        }
308
309        let response = self
310            .wait_for_confirmed_response(
311                target,
312                &tsm_mac,
313                invoke_id,
314                &owner,
315                registration.response,
316                registration.progress,
317                Some(&buf),
318            )
319            .await;
320        if response.is_ok() {
321            if let Some(lease) = path_lease.as_ref() {
322                lease.mark_terminal_observed();
323            }
324        }
325        guard.mark_completed();
326        response.and_then(Self::confirmed_response_result)
327    }
328
329    /// Wait through AWAIT_CONFIRMATION and SEGMENTED_CONF without allowing
330    /// their timers to cancel each other's phase.
331    #[allow(clippy::too_many_arguments)]
332    pub(super) async fn wait_for_confirmed_response(
333        &self,
334        target: ConfirmedTarget<'_>,
335        tsm_mac: &MacAddr,
336        invoke_id: u8,
337        owner: &TransactionOwner,
338        mut response_rx: oneshot::Receiver<TsmResponse>,
339        mut progress_rx: tokio::sync::watch::Receiver<TransactionProgress>,
340        retry_apdu: Option<&[u8]>,
341    ) -> Result<TsmResponse, Error> {
342        let (request_timeout, segment_timeout, max_retries) = {
343            let tsm = self.tsm.lock().await;
344            let config = tsm.config();
345            (
346                Duration::from_millis(config.apdu_timeout_ms),
347                Duration::from_millis(config.apdu_segment_timeout_ms.saturating_mul(4)),
348                config.apdu_retries,
349            )
350        };
351        let mut retries_sent = 0u16;
352        let mut progress = *progress_rx.borrow_and_update();
353
354        loop {
355            let wait = match progress {
356                TransactionProgress::AwaitingResponse => request_timeout,
357                TransactionProgress::SegmentedResponse { .. } => segment_timeout,
358            };
359            let timer = tokio::time::sleep(wait);
360            tokio::pin!(timer);
361
362            tokio::select! {
363                response = &mut response_rx => {
364                    return response.map_err(|_| Error::Encoding("TSM response channel closed".into()));
365                }
366                changed = progress_rx.changed() => {
367                    if changed.is_err() {
368                        return (&mut response_rx)
369                            .await
370                            .map_err(|_| Error::Encoding("TSM response channel closed".into()));
371                    }
372                    progress = *progress_rx.borrow_and_update();
373                }
374                _ = &mut timer => {
375                    match progress {
376                        TransactionProgress::AwaitingResponse => {
377                            let retry = retry_apdu.is_some()
378                                && retries_sent < u16::from(max_retries);
379                            let mut tsm = self.tsm.lock().await;
380                            match tsm.expire_request_timer(tsm_mac, invoke_id, owner, !retry) {
381                                RequestTimerExpiration::Retry => {
382                                    // Authorization and segment admission use the TSM lock,
383                                    // but transport I/O must not. Once authorized, this retry
384                                    // may finish even if segment zero is admitted meanwhile.
385                                    let Some(retry_apdu) = retry_apdu else {
386                                        tsm.cancel_transaction_for_owner(tsm_mac, invoke_id, owner);
387                                        drop(tsm);
388                                        self.enqueue_transaction_cleanup(
389                                            tsm_mac,
390                                            invoke_id,
391                                            owner,
392                                            false,
393                                            None,
394                                        );
395                                        return Err(Error::Timeout(request_timeout));
396                                    };
397                                    retries_sent += 1;
398                                    drop(tsm);
399                                    if !self.routed_path_limits.authorize_attempt(
400                                        target,
401                                        tsm_mac,
402                                        invoke_id,
403                                        owner,
404                                        retry_apdu.len(),
405                                    ) {
406                                        return (&mut response_rx).await.map_err(|_| {
407                                            Error::Encoding("TSM response channel closed".into())
408                                        });
409                                    }
410                                    let send_result = self
411                                        .send_confirmed_target_apdu(target, retry_apdu)
412                                        .await;
413                                    if let Err(error) = send_result {
414                                        let mut tsm = self.tsm.lock().await;
415                                        match tsm.expire_request_timer(tsm_mac, invoke_id, owner, true) {
416                                            RequestTimerExpiration::SegmentedResponse { generation } => {
417                                                progress = TransactionProgress::SegmentedResponse { generation };
418                                                continue;
419                                            }
420                                            RequestTimerExpiration::NoTransaction => {
421                                                drop(tsm);
422                                                return (&mut response_rx)
423                                                    .await
424                                                    .map_err(|_| Error::Encoding("TSM response channel closed".into()));
425                                            }
426                                            RequestTimerExpiration::TimedOut => {
427                                                drop(tsm);
428                                                self.enqueue_transaction_cleanup(
429                                                    tsm_mac,
430                                                    invoke_id,
431                                                    owner,
432                                                    false,
433                                                    None,
434                                                );
435                                                return Err(error);
436                                            }
437                                            RequestTimerExpiration::Retry => unreachable!(
438                                                "final retry-send failure disposition cannot authorize another retry"
439                                            ),
440                                        }
441                                    }
442                                    debug!(
443                                        invoke_id,
444                                        attempt = retries_sent,
445                                        max_retries,
446                                        "APDU timeout, retrying confirmed request"
447                                    );
448                                }
449                                RequestTimerExpiration::SegmentedResponse { generation } => {
450                                    progress = TransactionProgress::SegmentedResponse { generation };
451                                }
452                                RequestTimerExpiration::TimedOut => {
453                                    drop(tsm);
454                                    self.enqueue_transaction_cleanup(
455                                        tsm_mac,
456                                        invoke_id,
457                                        owner,
458                                        false,
459                                        None,
460                                    );
461                                    return Err(Error::Abort {
462                                        reason: bacnet_types::enums::AbortReason::TSM_TIMEOUT.to_raw(),
463                                    });
464                                }
465                                RequestTimerExpiration::NoTransaction => {
466                                    drop(tsm);
467                                    return (&mut response_rx)
468                                        .await
469                                        .map_err(|_| Error::Encoding("TSM response channel closed".into()));
470                                }
471                            }
472                        }
473                        TransactionProgress::SegmentedResponse { generation } => {
474                            let mut tsm = self.tsm.lock().await;
475                            match tsm.expire_segment_timer(tsm_mac, invoke_id, owner, generation) {
476                                SegmentTimerExpiration::Activity { generation } => {
477                                    progress = TransactionProgress::SegmentedResponse { generation };
478                                }
479                                SegmentTimerExpiration::AwaitingResponse => {
480                                    progress = TransactionProgress::AwaitingResponse;
481                                }
482                                SegmentTimerExpiration::TimedOut => {
483                                    drop(tsm);
484                                    #[cfg(test)]
485                                    self.segmented_cleanup.pause_if_enabled().await;
486                                    self.enqueue_transaction_cleanup(
487                                        tsm_mac,
488                                        invoke_id,
489                                        owner,
490                                        false,
491                                        None,
492                                    );
493                                    return Err(Error::Abort {
494                                        reason: bacnet_types::enums::AbortReason::TSM_TIMEOUT.to_raw(),
495                                    });
496                                }
497                                SegmentTimerExpiration::NoTransaction => {
498                                    drop(tsm);
499                                    return (&mut response_rx)
500                                        .await
501                                        .map_err(|_| Error::Encoding("TSM response channel closed".into()));
502                                }
503                            }
504                        }
505                    }
506                }
507            }
508        }
509    }
510
511    pub(super) fn enqueue_transaction_cleanup(
512        &self,
513        mac: &MacAddr,
514        invoke_id: u8,
515        owner: &TransactionOwner,
516        cancel_tsm: bool,
517        seg_ack_sender: Option<mpsc::Sender<SegmentAckPdu>>,
518    ) {
519        let _ = self.cleanup_tx.send(TransactionCleanup {
520            mac: mac.clone(),
521            invoke_id,
522            owner: owner.clone(),
523            cancel_tsm,
524            seg_ack_sender,
525        });
526    }
527
528    pub(super) fn confirmed_response_result(response: TsmResponse) -> Result<Bytes, Error> {
529        super::confirmed_response_result(response)
530    }
531
532    pub(super) async fn send_confirmed_target_apdu(
533        &self,
534        target: ConfirmedTarget<'_>,
535        apdu: &[u8],
536    ) -> Result<(), Error> {
537        match target {
538            ConfirmedTarget::Local { mac } => {
539                self.network
540                    .send_apdu(apdu, mac, true, NetworkPriority::NORMAL)
541                    .await
542            }
543            ConfirmedTarget::Routed {
544                router_mac,
545                dest_network,
546                dest_mac,
547            } => {
548                self.network
549                    .send_apdu_routed(
550                        apdu,
551                        dest_network,
552                        dest_mac,
553                        router_mac,
554                        true,
555                        NetworkPriority::NORMAL,
556                    )
557                    .await
558            }
559        }
560    }
561
562    pub(super) fn target_transport_max_apdu_length(&self, target: ConfirmedTarget<'_>) -> u16 {
563        self.network
564            .transport()
565            .max_apdu_length()
566            .saturating_sub(target.additional_npdu_header_len())
567    }
568
569    pub(super) fn advertised_max_apdu_length_for_target(
570        &self,
571        target: ConfirmedTarget<'_>,
572    ) -> Result<u16, Error> {
573        cap_max_apdu_to_transport(
574            self.config.max_apdu_length,
575            self.target_transport_max_apdu_length(target),
576        )
577    }
578
579    /// Send an unconfirmed request (fire-and-forget) to a specific destination.
580    pub async fn unconfirmed_request(
581        &self,
582        destination_mac: &[u8],
583        service_choice: UnconfirmedServiceChoice,
584        service_data: &[u8],
585    ) -> Result<(), Error> {
586        let pdu = Apdu::UnconfirmedRequest(bacnet_encoding::apdu::UnconfirmedRequest {
587            service_choice,
588            service_request: Bytes::copy_from_slice(service_data),
589        });
590
591        let mut buf = BytesMut::with_capacity(2 + service_data.len());
592        encode_apdu(&mut buf, &pdu)?;
593
594        self.network
595            .send_apdu(&buf, destination_mac, false, NetworkPriority::NORMAL)
596            .await
597    }
598
599    /// Broadcast an unconfirmed request on the local network.
600    pub async fn broadcast_unconfirmed(
601        &self,
602        service_choice: UnconfirmedServiceChoice,
603        service_data: &[u8],
604    ) -> Result<(), Error> {
605        let pdu = Apdu::UnconfirmedRequest(bacnet_encoding::apdu::UnconfirmedRequest {
606            service_choice,
607            service_request: Bytes::copy_from_slice(service_data),
608        });
609
610        let mut buf = BytesMut::with_capacity(2 + service_data.len());
611        encode_apdu(&mut buf, &pdu)?;
612
613        self.network
614            .broadcast_apdu(&buf, false, NetworkPriority::NORMAL)
615            .await
616    }
617
618    /// Broadcast an unconfirmed request globally (DNET=0xFFFF).
619    pub async fn broadcast_global_unconfirmed(
620        &self,
621        service_choice: UnconfirmedServiceChoice,
622        service_data: &[u8],
623    ) -> Result<(), Error> {
624        let pdu = Apdu::UnconfirmedRequest(bacnet_encoding::apdu::UnconfirmedRequest {
625            service_choice,
626            service_request: Bytes::copy_from_slice(service_data),
627        });
628
629        let mut buf = BytesMut::with_capacity(2 + service_data.len());
630        encode_apdu(&mut buf, &pdu)?;
631
632        self.network
633            .broadcast_global_apdu(&buf, false, NetworkPriority::NORMAL)
634            .await
635    }
636
637    /// Broadcast an unconfirmed request to a specific remote network.
638    pub async fn broadcast_network_unconfirmed(
639        &self,
640        service_choice: UnconfirmedServiceChoice,
641        service_data: &[u8],
642        dest_network: u16,
643    ) -> Result<(), Error> {
644        let pdu = Apdu::UnconfirmedRequest(bacnet_encoding::apdu::UnconfirmedRequest {
645            service_choice,
646            service_request: Bytes::copy_from_slice(service_data),
647        });
648
649        let mut buf = BytesMut::with_capacity(2 + service_data.len());
650        encode_apdu(&mut buf, &pdu)?;
651
652        self.network
653            .broadcast_to_network(&buf, dest_network, false, NetworkPriority::NORMAL)
654            .await
655    }
656}
657
658#[cfg(test)]
659#[path = "requests_tests.rs"]
660mod transmittable_length_tests;
bacnet_client/client/requests.rs

bacnet_client/client/
requests.rs
