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