bacnet_objects/log_persistence/
file.rs

1//! The two-slot file store for log snapshots.
2
3use std::fs::{File, OpenOptions};
4use std::io::{ErrorKind, Read, Write};
5use std::path::{Path, PathBuf};
6
7use bacnet_types::constructed::{BACnetEventLogRecord, BACnetLogMultipleRecord, BACnetLogRecord};
8use bacnet_types::error::Error;
9use bacnet_types::primitives::ObjectIdentifier;
10
11use super::codec::{self, Decoded, SnapshotRecord, MAX_SNAPSHOT_BYTES};
12use super::{LogPersistence, LogSnapshot};
13
14/// Two-slot, versioned and checksummed file store for one log's snapshots,
15/// as [`FileAuditLogPersistence`](crate::audit::FileAuditLogPersistence)
16/// keeps an Audit Log's.
17///
18/// Commits alternate between `base.slot0` and `base.slot1` by generation,
19/// and each one fully writes and synchronizes its slot, so a commit cut off
20/// part way leaves the other slot's snapshot to load. The first commit to a
21/// slot also synchronizes its directory (on Unix; see
22/// [`durable`](crate::durable)).
23///
24/// Loading takes the newer valid slot. A slot that is damaged (truncated,
25/// a bad checksum, or contents that don't decode or don't hold together) is
26/// passed over for the other one, with a warning; loading fails when both
27/// are damaged. It also fails, without falling back, when a slot belongs to
28/// another object or uses a schema this build can't read, when reading
29/// either slot fails, and when both slots hold different snapshots of one
30/// generation. It does not coordinate several processes.
31#[derive(Clone, Debug)]
32pub struct FileLogPersistence {
33    slot_paths: [PathBuf; 2],
34}
35
36impl FileLogPersistence {
37    /// Use `base_path.slot0` and `base_path.slot1`. An empty path is refused
38    /// with [`Error::OutOfRange`].
39    pub fn new(base_path: impl AsRef<Path>) -> Result<Self, Error> {
40        let base = base_path.as_ref().as_os_str();
41        if base.is_empty() {
42            return Err(Error::OutOfRange(
43                "log persistence base path must not be empty".into(),
44            ));
45        }
46        let slot = |suffix: &str| {
47            let mut path = base.to_os_string();
48            path.push(suffix);
49            PathBuf::from(path)
50        };
51        Ok(Self {
52            slot_paths: [slot(".slot0"), slot(".slot1")],
53        })
54    }
55
56    /// The two slot paths, for application backup and diagnostics.
57    pub fn slot_paths(&self) -> [PathBuf; 2] {
58        self.slot_paths.clone()
59    }
60
61    fn load_slots<R: SnapshotRecord>(
62        &self,
63        expected: ObjectIdentifier,
64    ) -> Result<Option<LogSnapshot<R>>, Error> {
65        let mut newest: Option<LogSnapshot<R>> = None;
66        let mut damage = None;
67        for path in &self.slot_paths {
68            let Some(data) = read_slot(path)? else {
69                continue;
70            };
71            match codec::decode::<R>(&data, expected) {
72                Decoded::Refused(error) => return Err(error),
73                Decoded::Corrupt(error) => {
74                    tracing::warn!(slot = %path.display(), %error, "Passing over a damaged log snapshot slot");
75                    damage = Some(error);
76                }
77                Decoded::Valid(snapshot) => match &newest {
78                    Some(current) if current.generation > snapshot.generation => {}
79                    Some(current) if current.generation == snapshot.generation => {
80                        if *current != snapshot {
81                            return Err(Error::Encoding(
82                                "log snapshot slots hold different snapshots of one generation"
83                                    .into(),
84                            ));
85                        }
86                    }
87                    _ => newest = Some(snapshot),
88                },
89            }
90        }
91        match (newest, damage) {
92            (Some(snapshot), _) => Ok(Some(snapshot)),
93            (None, Some(error)) => Err(error),
94            (None, None) => Ok(None),
95        }
96    }
97
98    fn commit_slot<R: SnapshotRecord>(&self, snapshot: &LogSnapshot<R>) -> Result<(), Error> {
99        let bytes = codec::encode(snapshot)?;
100        let path = &self.slot_paths[(snapshot.generation % 2) as usize];
101        if let Some(parent) = path
102            .parent()
103            .filter(|parent| !parent.as_os_str().is_empty())
104        {
105            if !parent.exists() {
106                std::fs::create_dir_all(parent)?;
107                // The new directory's own entry has to be durable too, or a
108                // power loss could take the slot with it.
109                crate::durable::sync_parent_dir(parent);
110            }
111        }
112        let (mut file, created) = match OpenOptions::new().write(true).create_new(true).open(path) {
113            Ok(file) => (file, true),
114            Err(error) if error.kind() == ErrorKind::AlreadyExists => (
115                OpenOptions::new().write(true).truncate(true).open(path)?,
116                false,
117            ),
118            Err(error) => return Err(error.into()),
119        };
120        if let Err(error) = file.write_all(&bytes).and_then(|()| file.sync_all()) {
121            // Best effort: the other slot keeps the last good snapshot.
122            let _ = file.set_len(0);
123            let _ = file.sync_all();
124            return Err(error.into());
125        }
126        if created {
127            crate::durable::sync_parent_dir(path);
128        }
129        Ok(())
130    }
131}
132
133/// A slot's octets, or `None` when it doesn't exist. A slot past the size
134/// cap is read only up to one octet past it, which then fails its length
135/// check as damage.
136fn read_slot(path: &Path) -> Result<Option<Vec<u8>>, Error> {
137    let file = match File::open(path) {
138        Ok(file) => file,
139        Err(error) if error.kind() == ErrorKind::NotFound => return Ok(None),
140        Err(error) => return Err(error.into()),
141    };
142    let mut data = Vec::new();
143    file.take(MAX_SNAPSHOT_BYTES as u64 + 1)
144        .read_to_end(&mut data)?;
145    Ok(Some(data))
146}
147
148macro_rules! file_log_persistence {
149    ($($record:ty),*) => {$(
150        impl LogPersistence<$record> for FileLogPersistence {
151            fn load(
152                &self,
153                expected_object: ObjectIdentifier,
154            ) -> Result<Option<LogSnapshot<$record>>, Error> {
155                self.load_slots(expected_object)
156            }
157
158            fn commit(&self, snapshot: &LogSnapshot<$record>) -> Result<(), Error> {
159                self.commit_slot(snapshot)
160            }
161        }
162    )*};
163}
164
165file_log_persistence!(
166    BACnetLogRecord,
167    BACnetEventLogRecord,
168    BACnetLogMultipleRecord
169);
bacnet_objects/log_persistence/file.rs

bacnet_objects/log_persistence/
file.rs
