bacnet_objects/log_persistence/
file.rs

1//! The two-slot file store for log snapshots.
2
3use crate::prelude::*;
4use std::fs::{File, OpenOptions};
5use std::io::{ErrorKind, Read, Write};
6use std::path::{Path, PathBuf};
7
8use bacnet_types::constructed::{BACnetEventLogRecord, BACnetLogMultipleRecord, BACnetLogRecord};
9use bacnet_types::error::Error;
10use bacnet_types::primitives::ObjectIdentifier;
11
12use super::codec::{self, Decoded, MAX_SNAPSHOT_BYTES, SnapshotRecord};
13use super::{LogPersistence, LogSnapshot};
14
15/// Two-slot, versioned and checksummed file store for one log's snapshots,
16/// as [`FileAuditLogPersistence`](crate::audit::FileAuditLogPersistence)
17/// keeps an Audit Log's.
18///
19/// Commits alternate between `base.slot0` and `base.slot1` by generation,
20/// and each one fully writes and synchronizes its slot, so a commit cut off
21/// part way leaves the other slot's snapshot to load. The first commit to a
22/// slot also synchronizes its directory (on Unix; see
23/// [`durable`](crate::durable)).
24///
25/// Loading takes the newer valid slot. A slot that is damaged (truncated,
26/// a bad checksum, or contents that don't decode or don't hold together) is
27/// passed over for the other one, with a warning; loading fails when both
28/// are damaged. It also fails, without falling back, when a slot belongs to
29/// another object or uses a schema this build can't read, when reading
30/// either slot fails, and when both slots hold different snapshots of one
31/// generation. It does not coordinate several processes.
32#[derive(Clone, Debug)]
33pub struct FileLogPersistence {
34    slot_paths: [PathBuf; 2],
35}
36
37impl FileLogPersistence {
38    /// Use `base_path.slot0` and `base_path.slot1`. An empty path is refused
39    /// with [`Error::OutOfRange`].
40    pub fn new(base_path: impl AsRef<Path>) -> Result<Self, Error> {
41        let base = base_path.as_ref().as_os_str();
42        if base.is_empty() {
43            return Err(Error::OutOfRange(
44                "log persistence base path must not be empty".into(),
45            ));
46        }
47        let slot = |suffix: &str| {
48            let mut path = base.to_os_string();
49            path.push(suffix);
50            PathBuf::from(path)
51        };
52        Ok(Self {
53            slot_paths: [slot(".slot0"), slot(".slot1")],
54        })
55    }
56
57    /// The two slot paths, for application backup and diagnostics.
58    pub fn slot_paths(&self) -> [PathBuf; 2] {
59        self.slot_paths.clone()
60    }
61
62    fn load_slots<R: SnapshotRecord>(
63        &self,
64        expected: ObjectIdentifier,
65    ) -> Result<Option<LogSnapshot<R>>, Error> {
66        let mut newest: Option<LogSnapshot<R>> = None;
67        let mut damage = None;
68        for path in &self.slot_paths {
69            let Some(data) = read_slot(path)? else {
70                continue;
71            };
72            match codec::decode::<R>(&data, expected) {
73                Decoded::Refused(error) => return Err(error),
74                Decoded::Corrupt(error) => {
75                    log_warn!(slot = %path.display(), %error, "Passing over a damaged log snapshot slot");
76                    damage = Some(error);
77                }
78                Decoded::Valid(snapshot) => match &newest {
79                    Some(current) if current.generation > snapshot.generation => {}
80                    Some(current) if current.generation == snapshot.generation => {
81                        if *current != snapshot {
82                            return Err(Error::Encoding(
83                                "log snapshot slots hold different snapshots of one generation"
84                                    .into(),
85                            ));
86                        }
87                    }
88                    _ => newest = Some(snapshot),
89                },
90            }
91        }
92        match (newest, damage) {
93            (Some(snapshot), _) => Ok(Some(snapshot)),
94            (None, Some(error)) => Err(error),
95            (None, None) => Ok(None),
96        }
97    }
98
99    fn commit_slot<R: SnapshotRecord>(&self, snapshot: &LogSnapshot<R>) -> Result<(), Error> {
100        let bytes = codec::encode(snapshot)?;
101        let path = &self.slot_paths[(snapshot.generation % 2) as usize];
102        if let Some(parent) = path
103            .parent()
104            .filter(|parent| !parent.as_os_str().is_empty())
105            && !parent.exists()
106        {
107            std::fs::create_dir_all(parent)?;
108            // The new directory's own entry has to be durable too, or a
109            // power loss could take the slot with it.
110            crate::durable::sync_parent_dir(parent);
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
