bacnet_objects/notification_forwarder/
persistence.rs

1//! Keeping a forwarder's Recipient_List and Subscribed_Recipients across a
2//! restart (Clauses 12.51.8 and 12.51.9).
3//!
4//! Like [`AuditLogPersistence`](crate::audit::AuditLogPersistence), the
5//! storage belongs to the application. The forwarder loads it once, when it is
6//! built, and saves both lists whenever either changes. Saves run on the
7//! forwarder's own writer thread. The bundled server waits for them with the
8//! database guard dropped; a write nobody staged waits where it is
9//! ([`crate::durable`] lists those paths). Each saved Subscribed_Recipients
10//! entry carries the whole minutes it had left at the save.
11
12use std::path::Path;
13
14use bacnet_encoding::constructed::{
15    decode_destination, decode_event_notification_subscription, encode_destination_list,
16    encode_event_notification_subscription_list,
17};
18use bacnet_types::constructed::{BACnetDestination, BACnetEventNotificationSubscription};
19use bacnet_types::error::Error;
20use bacnet_types::primitives::ObjectIdentifier;
21use bytes::BytesMut;
22
23use super::MAX_RECIPIENT_LIST_DESTINATIONS;
24use crate::durable::file::ObjectFile;
25use crate::subscribed_recipients::MAX_SUBSCRIBED_RECIPIENTS;
26
27/// What a forwarder keeps across a restart: both of its lists.
28#[derive(Clone, Debug, Default, PartialEq)]
29pub struct ForwarderSnapshot {
30    /// Recipient_List, in list order, once a write has set it. `None` until
31    /// then: the destinations the application configures apply at each
32    /// start.
33    pub recipient_list: Option<Vec<BACnetDestination>>,
34    /// The live Subscribed_Recipients entries, in list order, each with the
35    /// whole minutes it had left when saved.
36    pub subscribed_recipients: Vec<BACnetEventNotificationSubscription>,
37}
38
39/// Application-owned storage for one Notification Forwarder's lists.
40///
41/// The forwarder calls it from its writer thread, one call at a time. A
42/// written list lands in storage a moment before the forwarder serves it: see
43/// [storage leads the served
44/// state](crate::durable#storage-leads-the-served-state).
45pub trait NotificationForwarderPersistence: Send + Sync {
46    /// The lists last saved for `forwarder`, or `None` when nothing was
47    /// saved.
48    fn load(&self, forwarder: ObjectIdentifier) -> Result<Option<ForwarderSnapshot>, Error>;
49
50    /// Durably replace what is saved for `forwarder` with `snapshot`.
51    fn save(&self, forwarder: ObjectIdentifier, snapshot: &ForwarderSnapshot) -> Result<(), Error>;
52}
53
54const MAGIC: &[u8; 8] = b"RBNFWD01";
55/// After the shared header (the magic tag and the object identifier): one
56/// octet, 1 when the file holds a Recipient_List and 0 when not, then the
57/// Recipient_List's length.
58const BODY_HEADER_LEN: usize = 1 + 4;
59/// A full Recipient_List of the longest destinations is 1,504 octets and a
60/// full Subscribed_Recipients 1,184; anything much larger is not a file this
61/// backend wrote.
62pub(super) const MAX_FILE_BYTES: u64 = 64 * 1024;
63
64/// A file holding one forwarder's lists, replaced whole on each save.
65///
66/// A save writes a sibling `.tmp` file, synchronizes it, renames it over the
67/// old one and then synchronizes the directory, so a failed save leaves the
68/// previous lists in place and a completed one survives a power loss. Once
69/// the rename succeeds the save has landed: a directory sync that fails is
70/// logged, not returned. (On Windows the directory is not synchronized: see
71/// the [`durable`](crate::durable) module.) The file holds a magic tag, the
72/// forwarder's object identifier, whether a Recipient_List follows, the
73/// encoded Recipient_List with its length, and the encoded
74/// Subscribed_Recipients. It does not coordinate between processes.
75#[derive(Clone, Debug)]
76pub struct FileNotificationForwarderPersistence {
77    file: ObjectFile,
78}
79
80impl FileNotificationForwarderPersistence {
81    /// Keep the lists in the file at `path`.
82    pub fn new(path: impl AsRef<Path>) -> Result<Self, Error> {
83        Ok(Self {
84            file: ObjectFile::new(
85                path.as_ref(),
86                MAGIC,
87                MAX_FILE_BYTES,
88                "Notification Forwarder",
89            )?,
90        })
91    }
92
93    /// The file the lists are kept in.
94    pub fn path(&self) -> &Path {
95        self.file.path()
96    }
97}
98
99impl NotificationForwarderPersistence for FileNotificationForwarderPersistence {
100    fn load(&self, forwarder: ObjectIdentifier) -> Result<Option<ForwarderSnapshot>, Error> {
101        let Some(body) = self.file.load(forwarder)? else {
102            return Ok(None);
103        };
104        if body.len() < BODY_HEADER_LEN {
105            return Err(self.file.corrupt("has no valid header"));
106        }
107        let length = u32::from_be_bytes(body[1..BODY_HEADER_LEN].try_into().unwrap());
108        let split = usize::try_from(length)
109            .ok()
110            .and_then(|length| BODY_HEADER_LEN.checked_add(length))
111            .filter(|split| *split <= body.len())
112            .ok_or_else(|| self.file.corrupt("has a Recipient_List past its end"))?;
113        let recipient_list = &body[BODY_HEADER_LEN..split];
114        let recipient_list = match body[0] {
115            0 if recipient_list.is_empty() => None,
116            1 => Some(self.file.decode_capped(
117                recipient_list,
118                MAX_RECIPIENT_LIST_DESTINATIONS,
119                decode_destination,
120            )?),
121            _ => return Err(self.file.corrupt("has no valid header")),
122        };
123        Ok(Some(ForwarderSnapshot {
124            recipient_list,
125            subscribed_recipients: self.file.decode_capped(
126                &body[split..],
127                MAX_SUBSCRIBED_RECIPIENTS,
128                decode_event_notification_subscription,
129            )?,
130        }))
131    }
132
133    fn save(&self, forwarder: ObjectIdentifier, snapshot: &ForwarderSnapshot) -> Result<(), Error> {
134        let mut recipient_list = BytesMut::new();
135        if let Some(list) = &snapshot.recipient_list {
136            encode_destination_list(&mut recipient_list, list)?;
137        }
138        let mut body = BytesMut::with_capacity(BODY_HEADER_LEN + recipient_list.len() + 1024);
139        body.extend_from_slice(&[u8::from(snapshot.recipient_list.is_some())]);
140        let length = u32::try_from(recipient_list.len())
141            .map_err(|_| Error::OutOfRange("Recipient_List too long to save".into()))?;
142        body.extend_from_slice(&length.to_be_bytes());
143        body.extend_from_slice(&recipient_list);
144        encode_event_notification_subscription_list(&mut body, &snapshot.subscribed_recipients)?;
145        self.file.save(forwarder, &body)
146    }
147}
bacnet_objects/notification_forwarder/persistence.rs

bacnet_objects/notification_forwarder/
persistence.rs
