mj_controller/database/
writer.rs

1use super::*;
2
3pub(super) const DATABASE_WRITE_QUEUE_CAPACITY: usize = 256;
4
5/// Writes that take at least this long are logged with their label.
6const SLOW_WRITE: std::time::Duration = std::time::Duration::from_millis(250);
7
8/// A queued write, handed either the writer's connection or the reason it
9/// must not be used. The job -- not the lane -- decides what a refusal means
10/// to its caller.
11pub(super) type DatabaseWriteJob = Box<dyn FnOnce(Result<&mut Connection>) + Send + 'static>;
12
13pub(super) enum DatabaseWriterMessage {
14    Run {
15        label: &'static str,
16        job: DatabaseWriteJob,
17    },
18    Shutdown,
19}
20
21/// Cloneable submission handle for the daemon's ordered SQLite write lane.
22///
23/// Calling [`DatabaseWriter::execute`] is synchronous and may apply bounded
24/// backpressure, so async and UI callers must invoke database mutations from
25/// their existing supervised blocking tasks.
26#[derive(Clone)]
27pub struct DatabaseWriter {
28    pub(super) id: u64,
29    pub(super) sender: SyncSender<DatabaseWriterMessage>,
30    path: Arc<PathBuf>,
31    committed: tokio::sync::watch::Sender<std::result::Result<Arc<CommittedState>, Arc<str>>>,
32}
33
34impl std::fmt::Debug for DatabaseWriter {
35    fn fmt(&self, formatter: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
36        formatter
37            .debug_struct("DatabaseWriter")
38            .field("id", &self.id)
39            .finish_non_exhaustive()
40    }
41}
42
43impl DatabaseWriter {
44    pub fn committed_state(&self) -> Result<Arc<CommittedState>> {
45        self.committed
46            .borrow()
47            .clone()
48            .map_err(|error| anyhow::anyhow!("{error}"))
49    }
50
51    pub fn committed_changes(
52        &self,
53    ) -> tokio::sync::watch::Receiver<std::result::Result<Arc<CommittedState>, Arc<str>>> {
54        self.committed.subscribe()
55    }
56
57    pub(super) fn execute<T, F>(&self, label: &'static str, operation: F) -> Result<T>
58    where
59        T: Send + 'static,
60        F: FnOnce(&mut Connection) -> Result<T> + Send + 'static,
61    {
62        let (reply_tx, reply_rx) = sync_channel(1);
63        let publication = self.clone();
64        self.sender
65            .send(DatabaseWriterMessage::Run {
66                label,
67                job: Box::new(move |connection| {
68                    let reply = match connection {
69                        Ok(connection) => (|| {
70                            let previous = publication.committed_state()?;
71                            committed::begin_operation(&publication.path);
72                            let result = operation(connection);
73                            match committed::finish_operation(connection, &previous) {
74                                Ok(Some(next)) => { drop(publication.committed.send_replace(Ok(Arc::new(next)))); }
75                                Ok(None) => {}
76                                Err(error) => {
77                                    // The write may already be durable. Refuse all
78                                    // further jobs rather than classify a readback
79                                    // failure as a retryable write failure.
80                                    let detail: Arc<str> = format!(
81                                        "database publication failed after {label}; a commit may have occurred, do not replay this mutation: {error:#}"
82                                    ).into();
83                                    drop(publication.committed.send_replace(Err(detail.clone())));
84                                    bail!("{detail}");
85                                }
86                            }
87                            result
88                        })(),
89                        // The mismatch travels as the operation's own failure,
90                        // so a refused write reports why rather than the
91                        // writer-stopped message a dropped reply would give.
92                        Err(error) => Err(error),
93                    };
94                    let _ = reply_tx.send(reply);
95                }),
96            })
97            .map_err(|_| {
98                anyhow::anyhow!("submit database writer operation {label}: writer stopped")
99            })?;
100        reply_rx
101            .recv()
102            .with_context(|| format!("database writer stopped during {label}"))?
103    }
104}
105
106/// Owns the daemon's writer thread and persistent SQLite connection.
107///
108/// The owner is deliberately not cloneable. Dropping it removes the global
109/// submission handle, drains accepted work in FIFO order, and joins the
110/// thread before releasing the connection.
111pub struct DatabaseWriterOwner {
112    pub(super) writer: DatabaseWriter,
113    pub(super) thread: Option<JoinHandle<()>>,
114    pub(super) stopped: Receiver<Result<()>>,
115}
116
117impl std::fmt::Debug for DatabaseWriterOwner {
118    fn fmt(&self, formatter: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
119        formatter
120            .debug_struct("DatabaseWriterOwner")
121            .field("writer", &self.writer)
122            .finish_non_exhaustive()
123    }
124}
125
126impl DatabaseWriterOwner {
127    pub fn shutdown(mut self) -> Result<()> {
128        self.shutdown_inner()
129    }
130
131    pub(super) fn shutdown_inner(&mut self) -> Result<()> {
132        if self.thread.is_none() {
133            return Ok(());
134        }
135        clear_database_writer(self.writer.id);
136        let send_result = self.writer.sender.send(DatabaseWriterMessage::Shutdown);
137        let worker_result = self
138            .stopped
139            .recv()
140            .context("database writer stopped without reporting its result")?;
141        let join_result = self
142            .thread
143            .take()
144            .expect("database writer thread checked above")
145            .join();
146        if let Err(panic) = join_result {
147            std::panic::resume_unwind(panic);
148        }
149        match (send_result, worker_result) {
150            (_, Err(error)) => Err(error),
151            (Err(_), Ok(())) => bail!("request database writer shutdown: writer stopped"),
152            (Ok(()), Ok(())) => Ok(()),
153        }
154    }
155}
156
157impl Drop for DatabaseWriterOwner {
158    fn drop(&mut self) {
159        if let Err(error) = self.shutdown_inner() {
160            tracing::error!(%error, "database writer did not shut down cleanly");
161        }
162    }
163}
164
165pub(super) fn database_writer_slot() -> &'static Mutex<Option<DatabaseWriter>> {
166    static WRITER: OnceLock<Mutex<Option<DatabaseWriter>>> = OnceLock::new();
167    WRITER.get_or_init(|| Mutex::new(None))
168}
169
170/// Whether this process holds the database writer. Only the daemon does, so
171/// work that merely keeps a daemon-owned record current, and that other
172/// processes also run, asks this instead of failing the write there.
173pub(crate) fn database_writer_installed() -> bool {
174    database_writer_slot()
175        .lock()
176        .unwrap_or_else(PoisonError::into_inner)
177        .is_some()
178}
179
180/// The daemon's current durable records. Client/offline readers have no writer
181/// and use explicit store reads; a failed daemon projection remains an error.
182pub fn committed_state() -> Result<Option<Arc<CommittedState>>> {
183    let writer = database_writer_slot()
184        .lock()
185        .unwrap_or_else(PoisonError::into_inner)
186        .clone();
187    writer.map(|writer| writer.committed_state()).transpose()
188}
189
190pub fn subscribe_committed_state()
191-> Result<tokio::sync::watch::Receiver<std::result::Result<Arc<CommittedState>, Arc<str>>>> {
192    let writer = database_writer_slot()
193        .lock()
194        .unwrap_or_else(PoisonError::into_inner)
195        .clone()
196        .context("database writer is not installed")?;
197    Ok(writer.committed_changes())
198}
199
200pub(super) fn clear_database_writer(id: u64) {
201    let mut installed = database_writer_slot()
202        .lock()
203        .unwrap_or_else(PoisonError::into_inner);
204    if installed.as_ref().is_some_and(|writer| writer.id == id) {
205        *installed = None;
206    }
207}
208
209/// Install the process-wide writer for a test that owns its data directory.
210///
211/// Production installs this once, in the daemon, after `ControllerStoreGuard`
212/// establishes exclusivity, and the daemon is then the only process that
213/// writes. A test may do the same only because it re-execs itself with its own
214/// `MJ_DATA_DIR` and is therefore alone in its process — which is exactly why
215/// the tests that need this are shaped that way.
216///
217/// The returned owner has to be held for the rest of the test: dropping it
218/// stops the writer, and the next write fails with the message above.
219///
220/// This fixture is compiled unconditionally and hidden from the documentation
221/// because the controller crate's tests need it and a `#[cfg(test)]` item is
222/// invisible to another crate. It is a thin wrapper over
223/// [`start_database_writer`], so nothing test-only leaks into the library.
224#[doc(hidden)]
225#[must_use = "the writer stops when this owner is dropped"]
226pub fn install_isolated_test_writer() -> DatabaseWriterOwner {
227    start_database_writer().expect("install the writer for an isolated test child")
228}
229
230pub fn start_database_writer() -> Result<DatabaseWriterOwner> {
231    start_database_writer_at(&database_path(), true)
232}
233
234pub(super) fn start_database_writer_at(
235    path: &Path,
236    install_globally: bool,
237) -> Result<DatabaseWriterOwner> {
238    static NEXT_WRITER_ID: AtomicU64 = AtomicU64::new(1);
239
240    let mut connection = schema::open_writer(path)?;
241    let mut observed_revision = schema::read_schema_state(&connection)?.revision;
242    let initial = CommittedState::bootstrap(&mut connection)?;
243    let (committed, _) = tokio::sync::watch::channel(Ok(Arc::new(initial)));
244    let path = path.to_owned();
245    let (sender, receiver) = sync_channel(DATABASE_WRITE_QUEUE_CAPACITY);
246    let (stopped_tx, stopped) = sync_channel(1);
247    let id = NEXT_WRITER_ID.fetch_add(1, Ordering::Relaxed);
248    let writer = DatabaseWriter {
249        id,
250        sender,
251        path: Arc::new(path.clone()),
252        committed,
253    };
254    if install_globally {
255        let mut installed = database_writer_slot()
256            .lock()
257            .unwrap_or_else(PoisonError::into_inner);
258        ensure!(installed.is_none(), "database writer is already running");
259        *installed = Some(writer.clone());
260    }
261    let publication = writer.clone();
262    let thread = match thread::Builder::new()
263        .name("hel-database-writer".to_owned())
264        .spawn(move || {
265            let mut connection = connection;
266            let result = std::panic::catch_unwind(std::panic::AssertUnwindSafe(|| {
267                loop {
268                    match receiver.recv() {
269                        Ok(DatabaseWriterMessage::Run { label, job }) => {
270                            tracing::trace!(operation = label, "running database writer operation");
271                            let job_started = std::time::Instant::now();
272                            // Recheck compatibility even after startup, and
273                            // remember forward progress to detect rollback.
274                            match writer_schema_state(
275                                &path,
276                                &connection,
277                                label,
278                                &mut observed_revision,
279                            ) {
280                                Ok(()) => job(Ok(&mut connection)),
281                                Err(error) => job(Err(error)),
282                            }
283                            // Every write waits behind the one before it, so a
284                            // slow one delays all of them, the daemon's exit
285                            // included.
286                            let took = job_started.elapsed();
287                            if took >= SLOW_WRITE {
288                                tracing::info!(
289                                    operation = label,
290                                    duration_ms = took.as_millis(),
291                                    "slow database write"
292                                );
293                            }
294                            if let Err(error) = publication.committed_state() {
295                                break Err(error);
296                            }
297                        }
298                        Ok(DatabaseWriterMessage::Shutdown) => break Ok(()),
299                        Err(error) => {
300                            break Err(error).context("database writer queue disconnected");
301                        }
302                    }
303                }
304            }))
305            .unwrap_or_else(|panic| {
306                let detail = panic
307                    .downcast_ref::<&str>()
308                    .copied()
309                    .or_else(|| panic.downcast_ref::<String>().map(String::as_str))
310                    .unwrap_or("unknown panic payload");
311                Err(anyhow::anyhow!("database writer thread panicked: {detail}"))
312            });
313            if let Err(error) = &result {
314                drop(
315                    publication
316                        .committed
317                        .send_replace(Err(format!("database writer stopped: {error:#}").into())),
318                );
319            }
320            if result.is_ok() {
321                clear_database_writer(id);
322            }
323            let _ = stopped_tx.send(result);
324        }) {
325        Ok(thread) => thread,
326        Err(error) => {
327            if install_globally {
328                clear_database_writer(id);
329            }
330            return Err(error).context("spawn database writer thread");
331        }
332    };
333    Ok(DatabaseWriterOwner {
334        writer,
335        thread: Some(thread),
336        stopped,
337    })
338}
339
340/// Refuse incompatible stores, rollback, and unreadable metadata before a job.
341pub(super) fn writer_schema_state(
342    path: &Path,
343    connection: &Connection,
344    label: &'static str,
345    observed_revision: &mut i64,
346) -> Result<()> {
347    let result: Result<()> = (|| {
348        let state = schema::read_schema_state(connection)?;
349        if state.revision < *observed_revision {
350            return Err(StoreSchemaMismatch {
351                found: state.revision,
352                supported: SCHEMA_VERSION,
353                reason: StoreSchemaMismatchReason::Rollback {
354                    previous: *observed_revision,
355                },
356            }
357            .into());
358        }
359        *observed_revision = state.revision;
360        state.ensure_supported()
361    })();
362    if let Err(error) = &result {
363        tracing::error!(
364            operation = label,
365            path = %path.display(),
366            error = %error,
367            "could not establish store compatibility; refusing the operation"
368        );
369    }
370    result.with_context(|| {
371        format!(
372            "check database compatibility before {label} at {}",
373            path.display()
374        )
375    })
376}
377
378pub(super) fn submit_database_write<T, F>(label: &'static str, operation: F) -> Result<T>
379where
380    T: Send + 'static,
381    F: FnOnce(&mut Connection) -> Result<T> + Send + 'static,
382{
383    let writer = database_writer_slot()
384        .lock()
385        .unwrap_or_else(PoisonError::into_inner)
386        .clone();
387    if let Some(writer) = writer {
388        writer.execute(label, operation)
389    } else {
390        // There is one way to write, and this is not it. In production the
391        // daemon installs the writer after `ControllerStoreGuard` establishes
392        // exclusivity, and it is the only process that writes; a caller
393        // reaching here has no exclusivity and would be competing with
394        // whatever does. This used to open `database_path()` directly, which
395        // meant any process without a writer silently wrote to — and migrated
396        // — the real user database as a side effect of doing something else.
397        bail!("database writer is not available for operation {label}")
398    }
399}
mj_controller/database/writer.rs

mj_controller/database/
writer.rs
