Skip to main content

bacnet_objects/
durable.rs

1//! Saving object state off the object database lock (#1270).
2//!
3//! The Audit Log, the Notification Forwarder, the Notification Class and the
4//! Access Rights object keep state in storage the application provides
5//! ([`AuditLogPersistence`], [`NotificationForwarderPersistence`],
6//! [`NotificationClassPersistence`] and [`AccessRightsPersistence`]).
7//! A storage call can be slow, since the file backends write, synchronize and
8//! rename, so none of these objects makes one while its caller holds the
9//! database's write guard. They all follow the pattern this module provides.
10//! The Trend Log, Trend Log Multiple and Event Log commit to their
11//! [`LogPersistence`] through the same writer, on a cadence rather than per
12//! request (see [`crate::log_persistence`]), so they never stage.
13//!
14//! - Under the guard, the object copies the state to keep and queues the copy
15//!   on its save writer. The writer's own thread makes the storage calls, one
16//!   at a time and in the order they were queued.
17//! - A save whose outcome decides a request is *staged*. The object sets the
18//!   state the request would leave aside and keeps serving the old one. The
19//!   server drops the guard and waits for the save's [`SaveWait`] on Tokio's
20//!   blocking pool, so a paused test clock does not jump while the writer
21//!   thread works; then, holding
22//!   the guard again, it runs the request as usual, and the object takes the
23//!   saved state, or refuses the request if the save failed. So a forwarder
24//!   list write that cannot be saved is refused and leaves the old list, and an
25//!   Audit notification is stored, and a confirmed one acknowledged, only once
26//!   its commit is durable. An application's Audit Log purge (#1238) stages
27//!   the same way, though no property is written.
28//! - A save no request waits for, such as a forwarder's lapse and minute
29//!   saves, coalesces: a queued save the thread has not started is replaced by
30//!   the newer one, so a burst costs one save of the latest state.
31//!
32//! # Storage leads the served state
33//!
34//! A staged state reaches storage before anyone can read it from the object.
35//! The storage call returns on the writer's thread, and the object serves the
36//! new state only when the request that staged it holds the guard again and
37//! takes it (or a later stage, or the object's operation task, settles it).
38//! Peers never see that gap: the request answers, and a confirmed Audit
39//! notification is acknowledged, only after the object serves the saved
40//! state. Code that watches storage directly does see it. A storage call
41//! that has returned, such as an in-memory [`AuditLogPersistence`] whose
42//! `commit` has stored a record, does not mean a read or an AuditLogQuery
43//! finds the record yet; wait on the served state, such as the log's
44//! Record_Count, instead.
45//!
46//! A request that makes several writes to one object, as a
47//! WritePropertyMultiple can, stages them as one save of the state they
48//! leave together. Each write takes its own step of it as the request makes
49//! it, so the object serves the state the last write leaves only once every
50//! write has been made, and a request that stops part way keeps the steps it
51//! took while storage goes back to that served state (#1423). Under a
52//! mutation authorizer, the server asks it about such writes before staging
53//! them, and stages only the ones it allows (#1321).
54//!
55//! Some paths still wait for a save while the guard is held. They get the
56//! same outcome through the same writer; the object just queues the save and
57//! waits for it where it is:
58//!
59//! - a write nobody staged, such as application code writing through the
60//!   database;
61//! - an in-place change to an Audit Log, such as `add_record`, which first
62//!   lets a staged commit land, waiting for it if it is still running.
63//!
64//! A write nobody staged takes the place of a staged write it doesn't match
65//! only when the object will make it: a write the object refuses leaves the
66//! staged write alone (#1424).
67//!
68//! Dropping an object waits for the saves it has queued. The server's
69//! DeleteObject therefore drops a removed object on a blocking thread after
70//! releasing the guard, and a server dropped without `stop()` in async code
71//! drops its database there too (#1409); application code that removes an
72//! object, or drops the last handle on the database, should do the same.
73//! `bacnet_server::server::drop_database_off_runtime` lets go of a handle on
74//! the server's database that way, as the server's own detached tasks do
75//! (#1513).
76//! An object dropped with a write still staged for a request that never
77//! came back first saves the state it serves, so storage never keeps a
78//! state no client was told about, and waits for that save too (#1363).
79//! The server's `stop()` settles such writes once it has joined its
80//! requests ([`DurableWrites::settle_forgotten_writes`]) and waits for
81//! their saves, so a database dropped after a stop has nothing more to
82//! save. A server dropped without `stop()` returns before those saves land.
83//!
84//! The writer's thread is a plain `std` thread with no Tokio runtime, so a
85//! storage implementation that needs one brings its own handle. A storage
86//! call that panics fails that save. An object starts the thread with its
87//! first save and keeps it, parked while idle, until the object is dropped:
88//! one thread per object that has saved.
89//!
90//! # Adding an object
91//!
92//! Any other object that keeps a written state in storage reuses this module
93//! instead of saving under the guard:
94//!
95//! 1. Own a `staged::StagedSaves` over a `SaveWriter` of a snapshot of the
96//!    state to keep. It stages one save of the state a request's writes
97//!    leave (`staged::steps` works out each write's step), hands each write
98//!    its step or the save's error, and drops a staged write its request
99//!    never made, leaving the object to save its served state at once
100//!    (`StagedSaves::correct`), or as the object drops. A write nobody
101//!    staged saves with `StagedSaves::save_now`, once the object knows it
102//!    will make it, and a save nobody waits for coalesces through
103//!    `StagedSaves::submit_coalescing`.
104//! 2. Implement [`DurableWrites`] for the object, including
105//!    `settle_forgotten_writes` (`StagedSaves::drop_forgotten`), and return
106//!    it from `BACnetObject::durable_writes_internal`.
107//! 3. Add the object type, and the properties it saves, to `may_save` in the
108//!    server's `durable_writes` module. The server stages only the writes
109//!    listed there; any other write saves in place.
110//! 4. Give a file backend a `file::ObjectFile`, which tags the file with
111//!    the format and the object identifier, caps its size, and replaces it
112//!    through a synchronized temporary file, a rename and `sync_parent_dir`.
113//!
114//! The Notification Forwarder's `saving` module is the full example: a list
115//! write stages, and the operation task's saves coalesce. The Notification
116//! Class's is the smallest: a Recipient_List write stages, and nothing else
117//! saves. The Access Rights object's stages indexed array writes too, keyed
118//! by the index as well as the value.
119//!
120//! [`AuditLogPersistence`]: crate::audit::AuditLogPersistence
121//! [`NotificationForwarderPersistence`]: crate::notification_forwarder::NotificationForwarderPersistence
122//! [`NotificationClassPersistence`]: crate::notification_class::NotificationClassPersistence
123//! [`AccessRightsPersistence`]: crate::access_control::AccessRightsPersistence
124//! [`LogPersistence`]: crate::log_persistence::LogPersistence
125
126use std::collections::VecDeque;
127use std::future::Future;
128use std::panic::{catch_unwind, AssertUnwindSafe};
129use std::path::Path;
130use std::pin::Pin;
131use std::sync::{Arc, Condvar, Mutex, MutexGuard, OnceLock, PoisonError};
132use std::task::{Context, Poll, Waker};
133use std::thread::JoinHandle;
134use std::time::Instant;
135
136use bacnet_types::enums::{ErrorClass, ErrorCode, PropertyIdentifier};
137use bacnet_types::error::Error;
138use bacnet_types::primitives::PropertyValue;
139
140pub(crate) mod file;
141pub(crate) mod staged;
142
143/// How long a staged write may wait for its request once its save has run.
144/// It counts from the end of the save, so a slow save never uses it up. Past
145/// it the request is taken to be gone, and the object drops the staged state
146/// at the next stage or operation task call.
147#[cfg(not(test))]
148pub(crate) const STAGED_WRITE_LIFETIME: std::time::Duration = std::time::Duration::from_secs(10);
149/// Short in tests, so one can watch a forgotten staged write go.
150#[cfg(test)]
151pub(crate) const STAGED_WRITE_LIFETIME: std::time::Duration = std::time::Duration::from_millis(300);
152
153fn lock<T>(mutex: &Mutex<T>) -> MutexGuard<'_, T> {
154    mutex.lock().unwrap_or_else(PoisonError::into_inner)
155}
156
157/// Something that happens once, which any number of threads and tasks can
158/// wait for.
159#[derive(Default)]
160pub(crate) struct Event {
161    state: Mutex<EventState>,
162    happened: Condvar,
163}
164
165#[derive(Default)]
166struct EventState {
167    done: bool,
168    wakers: Vec<Waker>,
169}
170
171impl Event {
172    /// Mark the event as happened and wake every waiter.
173    pub(crate) fn set(&self) {
174        let wakers = {
175            let mut state = lock(&self.state);
176            state.done = true;
177            std::mem::take(&mut state.wakers)
178        };
179        self.happened.notify_all();
180        for waker in wakers {
181            waker.wake();
182        }
183    }
184
185    pub(crate) fn is_set(&self) -> bool {
186        lock(&self.state).done
187    }
188
189    fn wait_blocking(&self) {
190        let mut state = lock(&self.state);
191        while !state.done {
192            state = self
193                .happened
194                .wait(state)
195                .unwrap_or_else(PoisonError::into_inner);
196        }
197    }
198}
199
200/// A future that resolves once a save has run, or once a staged write that
201/// held an object has gone, whichever it was handed out for.
202///
203/// It holds no lock and borrows nothing, so it can be awaited after the
204/// database guard is dropped.
205#[must_use = "a SaveWait does nothing unless awaited"]
206#[derive(Clone)]
207pub struct SaveWait {
208    event: Arc<Event>,
209    /// A second wait this one also ends after, when an object that keeps
210    /// two kinds of state in storage settles both ([`and`](Self::and)).
211    then: Option<Arc<SaveWait>>,
212}
213
214impl SaveWait {
215    pub(crate) fn new(event: Arc<Event>) -> Self {
216        Self { event, then: None }
217    }
218
219    /// A wait that ends once both this wait and `other` have.
220    pub(crate) fn and(self, other: SaveWait) -> SaveWait {
221        let then = match self.then {
222            Some(then) => SaveWait::clone(&then).and(other),
223            None => other,
224        };
225        SaveWait {
226            event: self.event,
227            then: Some(Arc::new(then)),
228        }
229    }
230
231    /// Whether the wait is already over.
232    pub fn is_ready(&self) -> bool {
233        self.event.is_set() && self.then.as_ref().is_none_or(|then| then.is_ready())
234    }
235
236    /// Block this thread until the wait is over, for a caller that waits
237    /// off its async runtime, such as on Tokio's blocking pool.
238    pub fn block(&self) {
239        self.event.wait_blocking();
240        if let Some(then) = &self.then {
241            then.block();
242        }
243    }
244}
245
246impl std::fmt::Debug for SaveWait {
247    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
248        f.debug_struct("SaveWait")
249            .field("ready", &self.is_ready())
250            .finish()
251    }
252}
253
254impl Future for SaveWait {
255    type Output = ();
256
257    fn poll(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<()> {
258        {
259            let mut state = lock(&self.event.state);
260            if !state.done {
261                if !state.wakers.iter().any(|waker| waker.will_wake(cx.waker())) {
262                    state.wakers.push(cx.waker().clone());
263                }
264                return Poll::Pending;
265            }
266        }
267        match &self.then {
268            Some(then) => Pin::new(&mut SaveWait::clone(then)).poll(cx),
269            None => Poll::Ready(()),
270        }
271    }
272}
273
274/// The outcome of one queued save, filled in by the writer's thread.
275#[derive(Clone, Default)]
276pub(crate) struct SaveTicket {
277    slot: Arc<TicketSlot>,
278}
279
280#[derive(Default)]
281struct TicketSlot {
282    outcome: Mutex<Outcome>,
283    /// When the save finished; set before `done`.
284    finished_at: OnceLock<Instant>,
285    done: Arc<Event>,
286}
287
288#[derive(Default)]
289enum Outcome {
290    #[default]
291    Pending,
292    Done(Result<(), Error>),
293    /// Someone took the result; whether the save succeeded.
294    Taken(bool),
295}
296
297impl SaveTicket {
298    fn complete(&self, result: Result<(), Error>) {
299        *lock(&self.slot.outcome) = Outcome::Done(result);
300        let _ = self.slot.finished_at.set(Instant::now());
301        self.slot.done.set();
302    }
303
304    /// Whether the save has run.
305    pub(crate) fn is_done(&self) -> bool {
306        self.slot.done.is_set()
307    }
308
309    /// A future that resolves once the save has run.
310    pub(crate) fn wait(&self) -> SaveWait {
311        SaveWait::new(Arc::clone(&self.slot.done))
312    }
313
314    /// Whether `wait` is the one [`wait`](Self::wait) hands out for this
315    /// save, so a request releasing a staged write can show it staged it.
316    pub(crate) fn issued(&self, wait: &SaveWait) -> bool {
317        Arc::ptr_eq(&self.slot.done, &wait.event)
318    }
319
320    /// Whether, at `now`, a staged write waiting on this save has outlived
321    /// [`STAGED_WRITE_LIFETIME`]. The lifetime starts when the save finishes;
322    /// a save still running never outlives it.
323    pub(crate) fn outlived_at(&self, now: Instant) -> bool {
324        self.slot.finished_at.get().is_some_and(|finished| {
325            now.saturating_duration_since(*finished) >= STAGED_WRITE_LIFETIME
326        })
327    }
328
329    /// [`outlived_at`](Self::outlived_at) now.
330    pub(crate) fn outlived(&self) -> bool {
331        self.outlived_at(Instant::now())
332    }
333
334    /// Whether the save succeeded, once it has run.
335    pub(crate) fn succeeded(&self) -> Option<bool> {
336        match &*lock(&self.slot.outcome) {
337            Outcome::Pending => None,
338            Outcome::Done(result) => Some(result.is_ok()),
339            Outcome::Taken(ok) => Some(*ok),
340        }
341    }
342
343    /// Wait for the save, blocking the thread if it has not run yet, and
344    /// take its result. The result goes to one taker; a later call gets an
345    /// error that says so.
346    pub(crate) fn take_outcome(&self) -> Result<(), Error> {
347        self.slot.done.wait_blocking();
348        let mut outcome = lock(&self.slot.outcome);
349        match std::mem::take(&mut *outcome) {
350            Outcome::Done(result) => {
351                *outcome = Outcome::Taken(result.is_ok());
352                result
353            }
354            Outcome::Taken(ok) => {
355                *outcome = Outcome::Taken(ok);
356                if ok {
357                    Ok(())
358                } else {
359                    Err(Error::Encoding(
360                        "save failed; its error was reported".into(),
361                    ))
362                }
363            }
364            Outcome::Pending => unreachable!("the event is set only after the outcome"),
365        }
366    }
367}
368
369type SaveFn<S> = dyn Fn(&S) -> Result<(), Error> + Send + Sync;
370type DoneFn<S> = dyn Fn(&S, &Result<(), Error>) + Send + Sync;
371
372/// Runs one object's storage calls on a thread of its own, one at a time, in
373/// the order they were queued. See the [module documentation](self).
374///
375/// The thread starts with the first save and ends when the writer is
376/// dropped. Dropping the writer waits for the saves already queued, so state
377/// saved just before an object goes away is in storage when a new object
378/// loads it.
379pub(crate) struct SaveWriter<S: Send + 'static> {
380    shared: Arc<Shared<S>>,
381    /// Asserted unwind safe, as std's `JoinHandle` is not (#1428). That is
382    /// sound: only `enqueue`, which stores it, and the drop, which joins it,
383    /// touch the handle, both through `&mut self`, and a panic leaves it
384    /// either stored or not.
385    thread: Option<AssertUnwindSafe<JoinHandle<()>>>,
386    name: String,
387}
388
389struct Shared<S> {
390    queue: Mutex<Queue<S>>,
391    /// The thread has a job to run, or the writer is closing.
392    work: Condvar,
393    // A boxed closure is not `RefUnwindSafe`, so these would take that trait
394    // and `UnwindSafe` from the objects that own a writer (#1428). Asserting
395    // them is sound: `run_job` is their only caller and catches a panic from
396    // each call, on the writer's thread or on the caller's when `enqueue`
397    // could not start one. No panic from them reaches the caller's frames,
398    // and one that panics fails only that save.
399    save: AssertUnwindSafe<Box<SaveFn<S>>>,
400    done: AssertUnwindSafe<Box<DoneFn<S>>>,
401}
402
403struct Queue<S> {
404    jobs: VecDeque<Job<S>>,
405    running: bool,
406    closed: bool,
407    /// Waits [`SaveWriter::idle`] handed out, set once the queue next
408    /// empties with no job running.
409    idle_waits: Vec<Arc<Event>>,
410}
411
412struct Job<S> {
413    snapshot: S,
414    ticket: Option<SaveTicket>,
415}
416
417impl<S: Send + 'static> SaveWriter<S> {
418    /// A writer whose thread, named `name`, saves each snapshot with `save`
419    /// and then reports the result to `done`.
420    pub(crate) fn new(
421        name: String,
422        save: impl Fn(&S) -> Result<(), Error> + Send + Sync + 'static,
423        done: impl Fn(&S, &Result<(), Error>) + Send + Sync + 'static,
424    ) -> Self {
425        Self {
426            shared: Arc::new(Shared {
427                queue: Mutex::new(Queue {
428                    jobs: VecDeque::new(),
429                    running: false,
430                    closed: false,
431                    idle_waits: Vec::new(),
432                }),
433                work: Condvar::new(),
434                save: AssertUnwindSafe(Box::new(save)),
435                done: AssertUnwindSafe(Box::new(done)),
436            }),
437            thread: None,
438            name,
439        }
440    }
441
442    /// Queue a save whose outcome the caller wants.
443    pub(crate) fn submit(&mut self, snapshot: S) -> SaveTicket {
444        let ticket = SaveTicket::default();
445        self.enqueue(Job {
446            snapshot,
447            ticket: Some(ticket.clone()),
448        });
449        ticket
450    }
451
452    /// Queue a save nobody waits for. A save of this kind still queued and
453    /// not started is replaced, so only the latest state is saved.
454    pub(crate) fn submit_coalescing(&mut self, snapshot: S) {
455        {
456            let mut queue = lock(&self.shared.queue);
457            if let Some(last) = queue.jobs.back_mut().filter(|job| job.ticket.is_none()) {
458                last.snapshot = snapshot;
459                return;
460            }
461        }
462        self.enqueue(Job {
463            snapshot,
464            ticket: None,
465        });
466    }
467
468    /// A wait that ends once every save queued so far has run.
469    pub(crate) fn idle(&self) -> SaveWait {
470        let event = Arc::new(Event::default());
471        let mut queue = lock(&self.shared.queue);
472        if queue.running || !queue.jobs.is_empty() {
473            queue.idle_waits.push(Arc::clone(&event));
474        } else {
475            drop(queue);
476            event.set();
477        }
478        SaveWait::new(event)
479    }
480
481    /// Block until every queued save has run.
482    pub(crate) fn wait_idle(&self) {
483        self.idle().block();
484    }
485
486    /// The object is going away while storage may hold a staged state it
487    /// never served: save `served`, the state it does serve, and wait for
488    /// the save. The writer's own drop, which follows, would wait for it
489    /// anyway, so this blocks no longer than that drop does. A save that
490    /// fails is logged twice, by the done hook and here: storage then keeps
491    /// the staged state, and the next start serves it.
492    pub(crate) fn put_back(&mut self, served: S) {
493        if let Err(error) = self.submit(served).take_outcome() {
494            tracing::warn!(
495                writer = %self.name,
496                %error,
497                "Could not put storage back to the served state as the object closed; \
498                 the next start serves the state staged for a request that never finished"
499            );
500        }
501    }
502
503    fn enqueue(&mut self, job: Job<S>) {
504        if self.thread.is_none() {
505            let shared = Arc::clone(&self.shared);
506            match std::thread::Builder::new()
507                .name(self.name.clone())
508                .spawn(move || run(&shared))
509            {
510                Ok(handle) => self.thread = Some(AssertUnwindSafe(handle)),
511                Err(error) => {
512                    // Without a thread the save runs here, as it did before
513                    // saves left the caller's thread.
514                    tracing::warn!(%error, writer = %self.name, "Could not start a save thread");
515                    run_job(&self.shared, job);
516                    return;
517                }
518            }
519        }
520        lock(&self.shared.queue).jobs.push_back(job);
521        self.shared.work.notify_one();
522    }
523}
524
525impl<S: Send + 'static> Drop for SaveWriter<S> {
526    fn drop(&mut self) {
527        lock(&self.shared.queue).closed = true;
528        self.shared.work.notify_one();
529        if let Some(AssertUnwindSafe(thread)) = self.thread.take() {
530            let _ = thread.join();
531        }
532    }
533}
534
535fn run<S>(shared: &Shared<S>) {
536    loop {
537        let job = {
538            let mut queue = lock(&shared.queue);
539            loop {
540                if let Some(job) = queue.jobs.pop_front() {
541                    queue.running = true;
542                    break job;
543                }
544                if queue.closed {
545                    return;
546                }
547                queue = shared
548                    .work
549                    .wait(queue)
550                    .unwrap_or_else(PoisonError::into_inner);
551            }
552        };
553        run_job(shared, job);
554        let idle = {
555            let mut queue = lock(&shared.queue);
556            queue.running = false;
557            if queue.jobs.is_empty() {
558                std::mem::take(&mut queue.idle_waits)
559            } else {
560                Vec::new()
561            }
562        };
563        for event in idle {
564            event.set();
565        }
566    }
567}
568
569/// Make one storage call. A storage implementation that panics fails the save
570/// instead of ending the thread, so every ticket still completes. The
571/// snapshot is dropped before the ticket completes, so a caller that shares
572/// it through an `Arc` holds the only reference once the save has run.
573fn run_job<S>(shared: &Shared<S>, job: Job<S>) {
574    let Job { snapshot, ticket } = job;
575    let result = catch_unwind(AssertUnwindSafe(|| (shared.save.0)(&snapshot)))
576        .unwrap_or_else(|_| Err(Error::Encoding("persistence panicked during a save".into())));
577    let _ = catch_unwind(AssertUnwindSafe(|| (shared.done.0)(&snapshot, &result)));
578    drop(snapshot);
579    if let Some(ticket) = ticket {
580        ticket.complete(result);
581    }
582}
583
584/// Make a rename or a newly created file in `path`'s directory durable by
585/// synchronizing the directory.
586///
587/// Call it once the file is in place. By then the save has landed: memory
588/// takes the saved state, so storage and memory agree, and this never fails
589/// the save. A filesystem that cannot synchronize a directory (`EINVAL`,
590/// `ENOTSUP`, `EOPNOTSUPP` or `EBADF`, as some network and FUSE filesystems
591/// answer) is passed over quietly, as PostgreSQL does. Any other error is
592/// logged as a warning: the save stands, but a power loss right after it
593/// may bring back the previous file.
594///
595/// On Unix this opens the directory and calls `fsync` on it. Windows cannot
596/// open a directory through `std::fs`, so there the call does nothing: NTFS
597/// journals the rename, but a power loss right after it may still bring back
598/// the previous file.
599pub(crate) fn sync_parent_dir(path: &Path) {
600    #[cfg(unix)]
601    {
602        let parent = match path.parent() {
603            Some(parent) if !parent.as_os_str().is_empty() => parent,
604            _ => Path::new("."),
605        };
606        let result = std::fs::File::open(parent).and_then(|directory| directory.sync_all());
607        if let Err(error) = result {
608            if directory_sync_unsupported(&error) {
609                tracing::debug!(directory = %parent.display(), %error, "Directory sync not supported here");
610            } else {
611                tracing::warn!(
612                    directory = %parent.display(),
613                    %error,
614                    "Could not synchronize the directory after a save; the save stands"
615                );
616            }
617        }
618    }
619    #[cfg(not(unix))]
620    let _ = path;
621}
622
623/// The CRC-32 (IEEE 802.3, reflected) of `data`, which the two-slot
624/// snapshot files end with.
625pub(crate) fn crc32(data: &[u8]) -> u32 {
626    let mut crc = !0u32;
627    for byte in data {
628        crc ^= u32::from(*byte);
629        for _ in 0..8 {
630            crc = (crc >> 1) ^ (0xedb8_8320 & (0u32.wrapping_sub(crc & 1)));
631        }
632    }
633    !crc
634}
635
636/// Whether a directory sync failed only because the filesystem cannot
637/// synchronize a directory.
638#[cfg(unix)]
639fn directory_sync_unsupported(error: &std::io::Error) -> bool {
640    // ENOTSUP and EOPNOTSUPP are one value on Linux, two on macOS.
641    error.raw_os_error().is_some_and(|code| {
642        [libc::EINVAL, libc::ENOTSUP, libc::EOPNOTSUPP, libc::EBADF].contains(&code)
643    })
644}
645
646/// What [`DurableWrites::stage_write`] gave its caller.
647#[doc(hidden)]
648#[derive(Debug)]
649pub enum StageStep {
650    /// The object queued a save of the state the write would leave, and keeps
651    /// that state aside. Await the wait without the database guard, then make
652    /// the write as usual: the object takes the saved state, or refuses the
653    /// write if the save failed. Keep a clone of the wait to release the
654    /// staged write with.
655    Staged(SaveWait),
656    /// Another request's staged write holds the object. Await the wait, then
657    /// stage again.
658    Busy(SaveWait),
659    /// Nothing to stage: the object saves nothing for this write, or will
660    /// refuse it anyway. Make the write as usual.
661    Skip,
662}
663
664/// One write a request makes to an object, as
665/// [`DurableWrites::stage_writes`] receives it.
666#[doc(hidden)]
667#[derive(Debug, Clone, PartialEq)]
668pub struct PendingWrite {
669    /// The property written.
670    pub property: PropertyIdentifier,
671    /// The array index written, if any.
672    pub array_index: Option<u32>,
673    /// The value written.
674    pub value: PropertyValue,
675}
676
677/// Writes whose new state an object saves before serving it, staged so the
678/// save runs while the database guard is dropped (#1270).
679///
680/// The bundled server stages the writes each request makes to such an
681/// object: it calls [`stage_writes`](Self::stage_writes) under the guard,
682/// awaits the save after dropping it, makes the writes under the guard
683/// again, and in that same critical section calls
684/// [`release_staged_write`](Self::release_staged_write). A write made
685/// without staging still works: the object then saves in place.
686#[doc(hidden)]
687pub trait DurableWrites {
688    /// Stage a write of `value` to `property` (at `array_index`).
689    fn stage_write(
690        &mut self,
691        property: PropertyIdentifier,
692        array_index: Option<u32>,
693        value: &PropertyValue,
694    ) -> StageStep;
695
696    /// Stage the writes one request makes to this object, in the order it
697    /// makes them; a WritePropertyMultiple can make several. The default
698    /// stages the first write the object takes through
699    /// [`stage_write`](Self::stage_write), and the request's later writes to
700    /// the object save in place. Every bundled object overrides it to fold
701    /// the writes into one save (#1423): each write then takes its own step
702    /// of that save as the request makes it.
703    fn stage_writes(&mut self, writes: &[PendingWrite]) -> StageStep {
704        for write in writes {
705            match self.stage_write(write.property, write.array_index, &write.value) {
706                StageStep::Skip => continue,
707                step => return step,
708            }
709        }
710        StageStep::Skip
711    }
712
713    /// Stage a purge of the object's records, which the application asks
714    /// for through the server (an Audit Log's, #1238). The server goes on as
715    /// for a staged write, calling [`commit_purge`](Self::commit_purge) where
716    /// it would make the write. The default has nothing to purge.
717    fn stage_purge(&mut self) -> StageStep {
718        StageStep::Skip
719    }
720
721    /// Purge the object's records: take the purge staged on the state the
722    /// object serves, if there is one, or purge in place. The default
723    /// refuses with OBJECT / OPTIONAL_FUNCTIONALITY_NOT_SUPPORTED.
724    fn commit_purge(&mut self) -> Result<(), Error> {
725        Err(Error::Protocol {
726            class: ErrorClass::OBJECT,
727            code: ErrorCode::OPTIONAL_FUNCTIONALITY_NOT_SUPPORTED,
728        })
729    }
730
731    /// The request that staged a write is done, whether or not the write
732    /// reached the object. `staged` is the wait [`StageStep::Staged`] gave
733    /// it; a staged write some other request made since is left alone. A
734    /// staged state the write never took is dropped, and a save of the state
735    /// the object serves is queued at once, so storage follows the object
736    /// again.
737    fn release_staged_write(&mut self, staged: &SaveWait);
738
739    /// Whether the object holds something staged for a request that the
740    /// request has still to take or release. The default stages nothing.
741    fn has_staged_write(&self) -> bool {
742        false
743    }
744
745    /// No request is left to take or release what is staged, as once the
746    /// server's `stop()` has joined its requests (#1363). Settle it: a
747    /// staged write goes as a release would drop it, so storage goes back
748    /// to the served state. Returns a wait that ends once every save the
749    /// object has queued so far has run, to await off the guard, or `None`
750    /// when the object keeps nothing in storage. The default has nothing
751    /// staged.
752    fn settle_forgotten_writes(&mut self) -> Option<SaveWait> {
753        None
754    }
755}
756
757#[cfg(test)]
758#[path = "durable/tests.rs"]
759mod tests;