bacnet_ops/safety/
ledger.rs

1//! The session ledger: every change still in effect, with the pre-image
2//! that undoes it. It is saved to `ledger.json` in the profile workspace
3//! after each change, atomically, so a crash leaves it behind for the next
4//! launch to list.
5//!
6//! An entry is written before its request goes out (state `sending`), so
7//! even a kill mid-request leaves a record. Repeated writes to one slot or
8//! property keep the first pre-image; a write that puts back the pre-image
9//! (relinquishing a slot that was empty, say) closes the entry.
10
11use std::path::{Path, PathBuf};
12use std::sync::atomic::{AtomicU64, Ordering};
13
14use bacnet_types::enums::{ObjectType, PropertyIdentifier};
15use bacnet_types::primitives::ObjectIdentifier;
16use serde::{Deserialize, Serialize};
17
18use super::{FrontEnd, NULL, Planned, Stored, WriteSpec};
19
20mod fold;
21
22/// The ledger's file in the workspace.
23pub const FILE: &str = "ledger.json";
24
25/// What [`Ledger::begin`] returns for an element write that a whole-property
26/// entry already covers: no entry, so settling it changes nothing.
27pub(crate) const COVERED: u64 = 0;
28
29static NEXT_TEMP: AtomicU64 = AtomicU64::new(0);
30
31/// Where an entry stands.
32#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
33#[serde(rename_all = "lowercase")]
34pub enum State {
35    /// The request is out; no answer yet.
36    Sending,
37    /// Acknowledged.
38    Applied,
39    /// No answer came, so it may be in effect.
40    Unknown,
41}
42
43/// One change in effect.
44#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
45pub struct Entry {
46    /// Unique within the file.
47    pub id: u64,
48    /// The process run that made it.
49    pub run: String,
50    /// When it was first written, UTC.
51    pub at: String,
52    /// Device instance.
53    pub device: u32,
54    /// Object type, raw.
55    pub object_type: u32,
56    /// Object instance.
57    pub instance: u32,
58    /// Property, raw.
59    pub property: u32,
60    /// Array index.
61    #[serde(default)]
62    pub index: Option<u32>,
63    /// The priority slot, for a commandable write.
64    #[serde(default)]
65    pub priority: Option<u8>,
66    /// Written to a priority slot.
67    pub commandable: bool,
68    /// What undoes it: the slot's old value (`None`: it was NULL) or the
69    /// property's old value (`None`: unknown, so it can't be undone).
70    pub before: Option<Stored>,
71    /// What was written last.
72    pub written: Stored,
73    /// Where it stands.
74    pub state: State,
75    /// An earlier session chose to keep it on exit.
76    #[serde(default)]
77    pub kept: bool,
78    /// Order of the latest write to it across the ledger: revert goes
79    /// from the highest down.
80    #[serde(default)]
81    pub seq: u64,
82    /// The front end and principal of its latest write.
83    #[serde(flatten)]
84    pub origin: Origin,
85}
86
87/// Who wrote an entry: the front end, and for MCP the principal (on stdio,
88/// the OS user running the server). Entries written before #2068 are the
89/// TUI's, with no principal; the file leaves both out for the TUI.
90#[derive(Debug, Clone, Default, PartialEq, Eq, Serialize, Deserialize)]
91pub struct Origin {
92    /// The front end.
93    #[serde(default, skip_serializing_if = "FrontEnd::is_tui")]
94    pub front_end: FrontEnd,
95    /// The principal, where the front end has one.
96    #[serde(default, skip_serializing_if = "Option::is_none")]
97    pub principal: Option<String>,
98}
99
100impl Origin {
101    /// The MCP server, for `principal`.
102    pub fn mcp(principal: impl Into<String>) -> Self {
103        Self {
104            front_end: FrontEnd::Mcp,
105            principal: Some(principal.into()),
106        }
107    }
108}
109
110impl Entry {
111    /// The object.
112    pub fn object(&self) -> Option<ObjectIdentifier> {
113        ObjectIdentifier::new(ObjectType::from_raw(self.object_type), self.instance).ok()
114    }
115
116    /// The property.
117    pub fn property(&self) -> PropertyIdentifier {
118        PropertyIdentifier::from_raw(self.property)
119    }
120
121    /// `device 100 analog-output 1 present-value @8`.
122    pub fn label(&self) -> String {
123        match self.object() {
124            Some(object) => super::label(
125                self.device,
126                object,
127                self.property(),
128                self.index,
129                self.priority,
130            ),
131            None => format!("device {} object {}", self.device, self.object_type),
132        }
133    }
134
135    /// The value and priority that undo it: the old slot value or NULL at
136    /// the slot, or the old property value. `None` when the old value is
137    /// unknown.
138    pub fn undo(&self) -> Option<Stored> {
139        match (&self.before, self.commandable) {
140            (Some(before), _) => Some(before.clone()),
141            (None, true) => Some(Stored::null()),
142            (None, false) => None,
143        }
144    }
145
146    fn same_target(&self, spec: &WriteSpec, priority: Option<u8>) -> bool {
147        self.device == spec.device
148            && self.object_type == spec.object.object_type().to_raw()
149            && self.instance == spec.object.instance_number()
150            && self.property == spec.property.to_raw()
151            && self.index == spec.index
152            && self.priority == priority
153    }
154
155    /// A non-commandable entry on the property `spec` names, at any index.
156    fn same_property(&self, spec: &WriteSpec) -> bool {
157        self.priority.is_none()
158            && self.device == spec.device
159            && self.object_type == spec.object.object_type().to_raw()
160            && self.instance == spec.object.instance_number()
161            && self.property == spec.property.to_raw()
162    }
163
164    /// Both entries are non-commandable writes to one property.
165    fn same_array(&self, other: &Entry) -> bool {
166        self.priority.is_none()
167            && other.priority.is_none()
168            && (self.device, self.object_type, self.instance, self.property)
169                == (
170                    other.device,
171                    other.object_type,
172                    other.instance,
173                    other.property,
174                )
175    }
176
177    /// The device is back at the pre-image.
178    fn undone(&self) -> bool {
179        match &self.before {
180            Some(before) => before.bytes == self.written.bytes,
181            None => self.commandable && self.written.bytes == NULL,
182        }
183    }
184}
185
186#[derive(Serialize, Deserialize)]
187struct LedgerFile {
188    version: u32,
189    entries: Vec<Entry>,
190}
191
192/// The ledger and its file.
193#[derive(Debug)]
194pub struct Ledger {
195    path: PathBuf,
196    entries: Vec<Entry>,
197    /// Who the entries [`Ledger::begin`] records are written by.
198    origin: Origin,
199}
200
201impl Ledger {
202    /// Load the ledger in `workspace`; none yet is an empty one.
203    pub fn open(workspace: &Path) -> Result<Self, String> {
204        let path = workspace.join(FILE);
205        let entries = match std::fs::read_to_string(&path) {
206            Ok(text) => {
207                let file: LedgerFile = serde_json::from_str(&text)
208                    .map_err(|e| format!("cannot read {}: {e}", path.display()))?;
209                file.entries
210            }
211            Err(e) if e.kind() == std::io::ErrorKind::NotFound => Vec::new(),
212            Err(e) => return Err(format!("cannot read {}: {e}", path.display())),
213        };
214        Ok(Self {
215            path,
216            entries,
217            origin: Origin::default(),
218        })
219    }
220
221    /// The ledger, with the entries it records marked as `origin`'s (the
222    /// TUI's otherwise).
223    pub fn written_by(mut self, origin: Origin) -> Self {
224        self.origin = origin;
225        self
226    }
227
228    /// The entries, oldest first.
229    pub fn entries(&self) -> &[Entry] {
230        &self.entries
231    }
232
233    /// The whole-property entry for the property `spec` names, if any.
234    fn whole_entry(&self, spec: &WriteSpec) -> Option<&Entry> {
235        self.entries
236            .iter()
237            .find(|e| e.index.is_none() && e.same_property(spec))
238    }
239
240    /// True when the slot `spec` names at `priority` still holds what
241    /// this ledger wrote there: `current` is the slot's encoded value now.
242    /// A slot someone has since overwritten is theirs, not ours.
243    pub(crate) fn owns(&self, spec: &WriteSpec, priority: Option<u8>, current: &[u8]) -> bool {
244        self.entries
245            .iter()
246            .any(|entry| entry.same_target(spec, priority) && entry.written.bytes == current)
247    }
248
249    /// Record `planned` as sending, merging into an entry for the same
250    /// target (whose first pre-image stays). Returns the entry's id, or
251    /// [`COVERED`] when nothing is recorded.
252    ///
253    /// Element and whole-property writes to one array (#1801): a whole
254    /// write folds the element entries before it into one entry whose
255    /// pre-image is the original array, and an element write to an array
256    /// with a whole entry records nothing, since that entry's pre-image
257    /// already puts the element back.
258    #[doc(hidden)]
259    pub fn begin(&mut self, planned: &Planned, run: &str, at: &str) -> u64 {
260        let (spec, priority) = (&planned.spec, planned.priority);
261        let seq = self.entries.iter().map(|e| e.seq).max().unwrap_or(0) + 1;
262        if priority.is_none() && spec.index.is_some() && self.whole_entry(spec).is_some() {
263            return COVERED;
264        }
265        if let Some(entry) = self
266            .entries
267            .iter_mut()
268            .find(|e| e.same_target(spec, priority))
269        {
270            entry.written = planned.value.clone();
271            entry.state = State::Sending;
272            entry.seq = seq;
273            entry.origin = self.origin.clone();
274            return entry.id;
275        }
276        let id = self.entries.iter().map(|e| e.id).max().map_or(1, |n| n + 1);
277        let spec = &planned.spec;
278        let mut before = planned.before.clone();
279        if priority.is_none() && spec.index.is_none() {
280            let elements: Vec<Entry> = self
281                .entries
282                .iter()
283                .filter(|e| e.index.is_some() && e.same_property(spec))
284                .cloned()
285                .collect();
286            if !elements.is_empty() {
287                before = fold::original(before.as_ref(), &elements);
288                self.entries
289                    .retain(|e| !(e.index.is_some() && e.same_property(spec)));
290            }
291        }
292        self.entries.push(Entry {
293            id,
294            run: run.into(),
295            at: at.into(),
296            device: spec.device,
297            object_type: spec.object.object_type().to_raw(),
298            instance: spec.object.instance_number(),
299            property: spec.property.to_raw(),
300            index: spec.index,
301            priority: planned.priority,
302            commandable: planned.commandable,
303            before,
304            written: planned.value.clone(),
305            state: State::Sending,
306            kept: false,
307            seq,
308            origin: self.origin.clone(),
309        });
310        id
311    }
312
313    /// Settle entry `id`: applied, refused (put `backup`'s version of it
314    /// back, or drop it if it is new), or unknown.
315    #[doc(hidden)]
316    pub fn finish(&mut self, id: u64, applied: Option<bool>, backup: &[Entry]) {
317        match applied {
318            Some(false) => {
319                let failed = self.entries.iter().find(|e| e.id == id).cloned();
320                let old = backup.iter().find(|e| e.id == id).cloned();
321                self.entries.retain(|e| e.id != id);
322                if let Some(old) = old {
323                    self.entries.push(old);
324                }
325                // A refused whole write gives back the element entries it folded.
326                if let Some(whole) = failed.filter(|e| e.index.is_none() && e.priority.is_none()) {
327                    let folded: Vec<Entry> = backup
328                        .iter()
329                        .filter(|e| e.index.is_some() && e.same_array(&whole))
330                        .filter(|e| !self.entries.iter().any(|kept| kept.id == e.id))
331                        .cloned()
332                        .collect();
333                    self.entries.extend(folded);
334                }
335                self.entries.sort_by_key(|e| e.id);
336            }
337            Some(true) => {
338                if let Some(entry) = self.entries.iter_mut().find(|e| e.id == id) {
339                    entry.state = State::Applied;
340                }
341                self.entries.retain(|e| e.id != id || !e.undone());
342            }
343            None => {
344                if let Some(entry) = self.entries.iter_mut().find(|e| e.id == id) {
345                    entry.state = State::Unknown;
346                }
347            }
348        }
349    }
350
351    /// The entries, latest write first: the order revert undoes them in.
352    pub(crate) fn newest_first(&self) -> Vec<Entry> {
353        let mut entries = self.entries.clone();
354        entries.sort_by(|a, b| b.seq.cmp(&a.seq).then(b.id.cmp(&a.id)));
355        entries
356    }
357
358    /// Entries left mid-request (`sending`), whatever run wrote them: they
359    /// become `unknown`. Only the holder of the workspace's lock calls
360    /// this, and a writer in flight in any process holds that lock until
361    /// it has settled (the TUI for its whole session, MCP for each write),
362    /// so an entry still `sending` here has no writer left: its session
363    /// ended, or its task failed, before the answer. Returns them.
364    pub(crate) fn settle_abandoned(&mut self) -> Vec<Entry> {
365        let mut settled = Vec::new();
366        for entry in &mut self.entries {
367            if entry.state == State::Sending {
368                entry.state = State::Unknown;
369                settled.push(entry.clone());
370            }
371        }
372        settled
373    }
374
375    /// Drop entry `id`, once undone.
376    pub(crate) fn remove(&mut self, id: u64) {
377        self.entries.retain(|e| e.id != id);
378    }
379
380    /// Mark every entry as kept on exit.
381    pub fn keep_all(&mut self) {
382        for entry in &mut self.entries {
383            entry.kept = true;
384        }
385    }
386
387    /// Write the file: a complete, synced temporary file renamed over it.
388    pub fn save(&self) -> Result<(), String> {
389        let text = serde_json::to_string_pretty(&LedgerFile {
390            version: 1,
391            entries: self.entries.clone(),
392        })
393        .map_err(|e| format!("cannot encode the ledger: {e}"))?;
394        let directory = self.path.parent().unwrap_or(Path::new("."));
395        let temp = directory.join(format!(
396            ".ledger-{}-{}.tmp",
397            std::process::id(),
398            NEXT_TEMP.fetch_add(1, Ordering::Relaxed)
399        ));
400        super::file::publish(&temp, &self.path, &text, false)
401            .map_err(|e| format!("cannot save {}: {e}", self.path.display()))
402    }
403}
404
405#[cfg(test)]
406mod tests {
407    use bacnet_types::primitives::PropertyValue;
408
409    use super::super::{Effect, Gate, Occupancy, Tier};
410    use super::*;
411
412    fn stored(value: f32) -> Stored {
413        let mut buf = bytes::BytesMut::new();
414        bacnet_encoding::primitives::encode_property_value(&mut buf, &PropertyValue::Real(value))
415            .unwrap();
416        Stored {
417            bytes: buf.to_vec(),
418            text: value.to_string(),
419        }
420    }
421
422    fn planned(value: Stored, before: Option<Stored>) -> Planned {
423        Planned {
424            spec: WriteSpec {
425                device: 100,
426                object: ObjectIdentifier::new(ObjectType::ANALOG_OUTPUT, 1).unwrap(),
427                property: PropertyIdentifier::PRESENT_VALUE,
428                index: None,
429                value: None,
430                priority: Some(8),
431            },
432            priority: Some(8),
433            value,
434            commandable: true,
435            before,
436            occupancy: Occupancy::Empty,
437            effect: Effect::NotPresentValue,
438            out_of_service: None,
439            notes: Vec::new(),
440            gate: Gate::Allowed(Tier::T2),
441        }
442    }
443
444    #[test]
445    fn writes_merge_keep_the_first_pre_image_and_close_on_relinquish() {
446        let dir = tempfile::tempdir().unwrap();
447        let mut ledger = Ledger::open(dir.path()).unwrap();
448        let backup = ledger.entries().to_vec();
449        let id = ledger.begin(&planned(stored(42.0), None), "run-a", "t");
450        ledger.finish(id, Some(true), &backup);
451        let backup = ledger.entries().to_vec();
452        let again = ledger.begin(&planned(stored(43.0), Some(stored(42.0))), "run-a", "t");
453        assert_eq!(again, id, "merged");
454        ledger.finish(again, Some(true), &backup);
455        assert_eq!(ledger.entries()[0].before, None, "the first pre-image");
456        assert_eq!(ledger.entries()[0].undo(), Some(Stored::null()));
457        ledger.save().unwrap();
458        let reloaded = Ledger::open(dir.path()).unwrap();
459        assert_eq!(reloaded.entries(), ledger.entries());
460        // A refused write leaves the entry as it was.
461        let backup = ledger.entries().to_vec();
462        let id = ledger.begin(&planned(stored(44.0), None), "run-a", "t");
463        ledger.finish(id, Some(false), &backup);
464        assert_eq!(ledger.entries(), backup.as_slice());
465        // Relinquishing the slot puts back its pre-image (NULL).
466        let id = ledger.begin(&planned(Stored::null(), Some(stored(43.0))), "run-a", "t");
467        ledger.finish(id, Some(true), &backup);
468        assert!(ledger.entries().is_empty());
469    }
470
471    fn texts(values: &[&str]) -> Stored {
472        let mut buf = bytes::BytesMut::new();
473        for value in values {
474            let value = PropertyValue::CharacterString((*value).into());
475            bacnet_encoding::primitives::encode_property_value(&mut buf, &value).unwrap();
476        }
477        Stored {
478            bytes: buf.to_vec(),
479            text: values.join(", "),
480        }
481    }
482
483    /// A non-commandable write to MSV:1's State_Text, at `index` or whole.
484    fn state_text(index: Option<u32>, value: &[&str], before: Option<&[&str]>) -> Planned {
485        let mut planned = planned(texts(value), before.map(texts));
486        planned.spec.object = ObjectIdentifier::new(ObjectType::MULTI_STATE_VALUE, 1).unwrap();
487        planned.spec.property = PropertyIdentifier::STATE_TEXT;
488        planned.spec.index = index;
489        planned.spec.priority = None;
490        (planned.priority, planned.commandable) = (None, false);
491        planned
492    }
493
494    /// The reviewer's sequence on [S1, S2, S3]: `[2]` = A, then the whole
495    /// array, then `[2]` = B. One entry is left, and it holds the original.
496    #[test]
497    fn element_and_whole_writes_to_one_array_revert_to_the_original() {
498        let dir = tempfile::tempdir().unwrap();
499        let mut ledger = Ledger::open(dir.path()).unwrap();
500        let t1 = ledger.begin(&state_text(Some(2), &["A"], Some(&["S2"])), "r", "t");
501        ledger.finish(t1, Some(true), &[]);
502        let whole = state_text(None, &["X", "Y", "Z"], Some(&["S1", "A", "S3"]));
503        let t2 = ledger.begin(&whole, "r", "t");
504        ledger.finish(t2, Some(true), &[]);
505        let t3 = ledger.begin(&state_text(Some(2), &["B"], Some(&["Y"])), "r", "t");
506        assert_eq!(t3, COVERED, "the whole entry covers it");
507        ledger.finish(t3, Some(true), &[]);
508        let [entry] = ledger.entries() else {
509            panic!("{:?}", ledger.entries());
510        };
511        assert_eq!(entry.index, None);
512        assert_eq!(
513            entry.undo().unwrap().bytes,
514            texts(&["S1", "S2", "S3"]).bytes
515        );
516        assert_eq!(entry.written.bytes, texts(&["X", "Y", "Z"]).bytes);
517    }
518
519    /// A refused whole write gives back the element entries it folded; an
520    /// element whose old value is unknown leaves the original unknown.
521    #[test]
522    fn a_refused_whole_write_unfolds_and_an_unknown_element_stays_unknown() {
523        let dir = tempfile::tempdir().unwrap();
524        let mut ledger = Ledger::open(dir.path()).unwrap();
525        let t1 = ledger.begin(&state_text(Some(2), &["A"], Some(&["S2"])), "r", "t");
526        ledger.finish(t1, Some(true), &[]);
527        let backup = ledger.entries().to_vec();
528        let whole = state_text(None, &["X", "Y", "Z"], Some(&["S1", "A", "S3"]));
529        let t2 = ledger.begin(&whole, "r", "t");
530        assert_eq!(ledger.entries().len(), 1, "folded");
531        ledger.finish(t2, Some(false), &backup);
532        assert_eq!(ledger.entries(), backup.as_slice());
533
534        let resize = state_text(Some(0), &["4"], None);
535        let t0 = ledger.begin(&resize, "r", "t");
536        ledger.finish(t0, Some(true), &[]);
537        let t2 = ledger.begin(&whole, "r", "t");
538        ledger.finish(t2, Some(true), &[]);
539        let [entry] = ledger.entries() else {
540            panic!("{:?}", ledger.entries());
541        };
542        assert_eq!(entry.undo(), None, "a resize was among them");
543    }
544
545    #[test]
546    fn an_unanswered_write_stays_as_unknown() {
547        let dir = tempfile::tempdir().unwrap();
548        let mut ledger = Ledger::open(dir.path()).unwrap();
549        let id = ledger.begin(&planned(stored(42.0), None), "run-a", "t");
550        assert_eq!(ledger.entries()[0].state, State::Sending);
551        ledger.finish(id, None, &[]);
552        assert_eq!(ledger.entries()[0].state, State::Unknown);
553    }
554
555    #[test]
556    fn a_slot_is_ours_only_while_it_holds_what_we_wrote() {
557        let dir = tempfile::tempdir().unwrap();
558        let mut ledger = Ledger::open(dir.path()).unwrap();
559        let ours = planned(stored(42.0), None);
560        let id = ledger.begin(&ours, "run-a", "t");
561        ledger.finish(id, Some(true), &[]);
562        assert!(ledger.owns(&ours.spec, Some(8), &stored(42.0).bytes));
563        assert!(
564            !ledger.owns(&ours.spec, Some(8), &stored(50.0).bytes),
565            "someone overwrote it"
566        );
567        assert!(!ledger.owns(&ours.spec, Some(9), &stored(42.0).bytes));
568    }
569
570    #[test]
571    fn revert_order_follows_the_latest_write_and_abandoned_sends_settle() {
572        let dir = tempfile::tempdir().unwrap();
573        let mut ledger = Ledger::open(dir.path()).unwrap();
574        let first = ledger.begin(&planned(stored(1.0), None), "run-a", "t");
575        let mut other = planned(stored(2.0), None);
576        other.priority = Some(9);
577        let second = ledger.begin(&other, "run-a", "t");
578        // Writing the first target again makes it the newest.
579        ledger.begin(&planned(stored(3.0), None), "run-a", "t");
580        let order: Vec<u64> = ledger.newest_first().iter().map(|e| e.id).collect();
581        assert_eq!(order, [first, second]);
582        // Whoever holds the lock settles every send left without a writer,
583        // its own run's included (#2068).
584        let settled = ledger.settle_abandoned();
585        assert_eq!(settled.len(), 2);
586        assert!(ledger.entries().iter().all(|e| e.state == State::Unknown));
587    }
588}
bacnet_ops/safety/ledger.rs

bacnet_ops/safety/
ledger.rs
