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