nodedb_lite/nodedb/core/
auto_flush.rs

1// SPDX-License-Identifier: Apache-2.0
2
3//! `NodeDbLite::start_auto_flush` — durable background flush task.
4
5use std::sync::{Arc, Weak};
6use std::time::Duration;
7
8use crate::storage::engine::StorageEngine;
9
10use super::types::NodeDbLite;
11
12impl<S: StorageEngine> NodeDbLite<S> {
13    /// Start a background task that calls the global `flush()` every
14    /// `interval_ms` milliseconds, bounding the data-loss window uniformly
15    /// across all engines (KV buffer, vector id-map, CRDT deltas, CSR graph,
16    /// spatial, FTS).
17    ///
18    /// # Durability contract
19    ///
20    /// `await`-ing a write operation (e.g. `kv_put`, `vector_insert`) returning
21    /// `Ok` does NOT guarantee on-disk durability. Durability is bounded by
22    /// `interval_ms`. For guaranteed durability, call `flush()` explicitly after
23    /// writes.
24    ///
25    /// # Usage
26    ///
27    /// The `open*` constructors already start this task from
28    /// [`LiteConfig::auto_flush_ms`](crate::config::LiteConfig::auto_flush_ms),
29    /// so calling it is only needed to change the interval afterwards:
30    ///
31    /// ```ignore
32    /// let db = NodeDbLite::open(storage).await?;
33    /// db.start_auto_flush(5_000); // slow the flusher down to five seconds
34    /// ```
35    ///
36    /// Each call spawns an additional task rather than replacing the running
37    /// one, so opening with `auto_flush_ms: 0` is the way to take full manual
38    /// control of when flushes happen.
39    ///
40    /// # Task lifecycle
41    ///
42    /// The task is registered with the database's task registry, so
43    /// [`NodeDbLite::shutdown`](crate::NodeDbLite::shutdown) stops it before
44    /// the host drops its async runtime. It also holds a `Weak` reference, so
45    /// a database dropped without a shutdown still ends the loop at its next
46    /// tick rather than keeping the database alive.
47    ///
48    /// # Disabling
49    ///
50    /// Pass `interval_ms = 0` to skip spawning entirely (auto-flush disabled).
51    pub fn start_auto_flush(self: &Arc<Self>, interval_ms: u64) {
52        if interval_ms == 0 {
53            return;
54        }
55
56        let weak: Weak<Self> = Arc::downgrade(self);
57        let period = Duration::from_millis(interval_ms);
58
59        let (stop_tx, mut stop) = crate::tasks::TaskRegistry::signal();
60        let handle = crate::runtime::spawn(async move {
61            let mut ticker = crate::runtime::interval(period);
62            // A flush that outlasts its own period must not be followed by the
63            // ticks it missed, back to back. Each pass takes the `crdt` lock,
64            // so a burst of them is a stretch of wall time in which no CRDT
65            // read can be scheduled — the shape a large store degenerated into
66            // once a single flush took longer than `interval_ms`.
67            ticker.delay_missed_ticks();
68            // Consume the first tick so the initial period elapses before the
69            // first flush (matches Tokio's immediate-first-tick semantics on
70            // native; on WASM the first tick already waits one period).
71            ticker.tick().await;
72
73            loop {
74                if !stop.tick_or_stop(&mut ticker).await {
75                    break;
76                }
77
78                let db = match weak.upgrade() {
79                    Some(db) => db,
80                    None => break,
81                };
82
83                if let Err(e) = db.flush().await {
84                    tracing::warn!(error = %e, "auto-flush failed");
85                }
86
87                // Drop the strong Arc before the next tick so the loop does
88                // not keep the database alive between ticks.
89                drop(db);
90            }
91        });
92        self.tasks
93            .track(crate::tasks::TaskKind::AutoFlush, stop_tx, handle);
94    }
95}
nodedb_lite/nodedb/core/auto_flush.rs

nodedb_lite/nodedb/core/
auto_flush.rs
