mj_controller/database/
writer.rs

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

mj_controller/database/
writer.rs
