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}
28
29impl std::fmt::Debug for DatabaseWriter {
30    fn fmt(&self, formatter: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
31        formatter
32            .debug_struct("DatabaseWriter")
33            .field("id", &self.id)
34            .finish_non_exhaustive()
35    }
36}
37
38impl DatabaseWriter {
39    pub(super) fn execute<T, F>(&self, label: &'static str, operation: F) -> Result<T>
40    where
41        T: Send + 'static,
42        F: FnOnce(&mut Connection) -> Result<T> + Send + 'static,
43    {
44        let (reply_tx, reply_rx) = sync_channel(1);
45        self.sender
46            .send(DatabaseWriterMessage::Run {
47                label,
48                job: Box::new(move |connection| {
49                    let reply = match connection {
50                        Ok(connection) => operation(connection),
51                        // The mismatch travels as the operation's own failure,
52                        // so a refused write reports why rather than the
53                        // writer-stopped message a dropped reply would give.
54                        Err(error) => Err(error),
55                    };
56                    let _ = reply_tx.send(reply);
57                }),
58            })
59            .map_err(|_| {
60                anyhow::anyhow!("submit database writer operation {label}: writer stopped")
61            })?;
62        reply_rx
63            .recv()
64            .with_context(|| format!("database writer stopped during {label}"))?
65    }
66}
67
68/// Owns the daemon's writer thread and persistent SQLite connection.
69///
70/// The owner is deliberately not cloneable. Dropping it removes the global
71/// submission handle, drains accepted work in FIFO order, and joins the
72/// thread before releasing the connection.
73pub struct DatabaseWriterOwner {
74    pub(super) writer: DatabaseWriter,
75    pub(super) thread: Option<JoinHandle<()>>,
76    pub(super) stopped: Receiver<Result<()>>,
77}
78
79impl std::fmt::Debug for DatabaseWriterOwner {
80    fn fmt(&self, formatter: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
81        formatter
82            .debug_struct("DatabaseWriterOwner")
83            .field("writer", &self.writer)
84            .finish_non_exhaustive()
85    }
86}
87
88impl DatabaseWriterOwner {
89    pub fn shutdown(mut self) -> Result<()> {
90        self.shutdown_inner()
91    }
92
93    pub(super) fn shutdown_inner(&mut self) -> Result<()> {
94        if self.thread.is_none() {
95            return Ok(());
96        }
97        clear_database_writer(self.writer.id);
98        let send_result = self.writer.sender.send(DatabaseWriterMessage::Shutdown);
99        let worker_result = self
100            .stopped
101            .recv()
102            .context("database writer stopped without reporting its result")?;
103        let join_result = self
104            .thread
105            .take()
106            .expect("database writer thread checked above")
107            .join();
108        if let Err(panic) = join_result {
109            std::panic::resume_unwind(panic);
110        }
111        match (send_result, worker_result) {
112            (_, Err(error)) => Err(error),
113            (Err(_), Ok(())) => bail!("request database writer shutdown: writer stopped"),
114            (Ok(()), Ok(())) => Ok(()),
115        }
116    }
117}
118
119impl Drop for DatabaseWriterOwner {
120    fn drop(&mut self) {
121        if let Err(error) = self.shutdown_inner() {
122            tracing::error!(%error, "database writer did not shut down cleanly");
123        }
124    }
125}
126
127pub(super) fn database_writer_slot() -> &'static Mutex<Option<DatabaseWriter>> {
128    static WRITER: OnceLock<Mutex<Option<DatabaseWriter>>> = OnceLock::new();
129    WRITER.get_or_init(|| Mutex::new(None))
130}
131
132pub(super) fn clear_database_writer(id: u64) {
133    let mut installed = database_writer_slot()
134        .lock()
135        .unwrap_or_else(PoisonError::into_inner);
136    if installed.as_ref().is_some_and(|writer| writer.id == id) {
137        *installed = None;
138    }
139}
140
141/// Install the process-wide writer for a test that owns its data directory.
142///
143/// Production installs this once, in the daemon, after `ControllerStoreGuard`
144/// establishes exclusivity, and the daemon is then the only process that
145/// writes. A test may do the same only because it re-execs itself with its own
146/// `MJ_DATA_DIR` and is therefore alone in its process — which is exactly why
147/// the tests that need this are shaped that way.
148///
149/// The returned owner has to be held for the rest of the test: dropping it
150/// stops the writer, and the next write fails with the message above.
151///
152/// This fixture is compiled unconditionally and hidden from the documentation
153/// because the controller crate's tests need it and a `#[cfg(test)]` item is
154/// invisible to another crate. It is a thin wrapper over
155/// [`start_database_writer`], so nothing test-only leaks into the library.
156#[doc(hidden)]
157#[must_use = "the writer stops when this owner is dropped"]
158pub fn install_isolated_test_writer() -> DatabaseWriterOwner {
159    start_database_writer().expect("install the writer for an isolated test child")
160}
161
162pub fn start_database_writer() -> Result<DatabaseWriterOwner> {
163    start_database_writer_at(&database_path(), true)
164}
165
166pub(super) fn start_database_writer_at(
167    path: &Path,
168    install_globally: bool,
169) -> Result<DatabaseWriterOwner> {
170    static NEXT_WRITER_ID: AtomicU64 = AtomicU64::new(1);
171
172    let connection = schema::open_writer(path)?;
173    let mut observed_revision = schema::read_schema_state(&connection)?.revision;
174    let path = path.to_owned();
175    let (sender, receiver) = sync_channel(DATABASE_WRITE_QUEUE_CAPACITY);
176    let (stopped_tx, stopped) = sync_channel(1);
177    let id = NEXT_WRITER_ID.fetch_add(1, Ordering::Relaxed);
178    let writer = DatabaseWriter { id, sender };
179    if install_globally {
180        let mut installed = database_writer_slot()
181            .lock()
182            .unwrap_or_else(PoisonError::into_inner);
183        ensure!(installed.is_none(), "database writer is already running");
184        *installed = Some(writer.clone());
185    }
186    let thread = match thread::Builder::new()
187        .name("hel-database-writer".to_owned())
188        .spawn(move || {
189            let mut connection = connection;
190            let result = std::panic::catch_unwind(std::panic::AssertUnwindSafe(|| {
191                loop {
192                    match receiver.recv() {
193                        Ok(DatabaseWriterMessage::Run { label, job }) => {
194                            tracing::trace!(operation = label, "running database writer operation");
195                            // Recheck compatibility even after startup, and
196                            // remember forward progress to detect rollback.
197                            match writer_schema_state(
198                                &path,
199                                &connection,
200                                label,
201                                &mut observed_revision,
202                            ) {
203                                Ok(()) => job(Ok(&mut connection)),
204                                Err(error) => job(Err(error)),
205                            }
206                        }
207                        Ok(DatabaseWriterMessage::Shutdown) => break Ok(()),
208                        Err(error) => {
209                            break Err(error).context("database writer queue disconnected");
210                        }
211                    }
212                }
213            }))
214            .unwrap_or_else(|panic| {
215                let detail = panic
216                    .downcast_ref::<&str>()
217                    .copied()
218                    .or_else(|| panic.downcast_ref::<String>().map(String::as_str))
219                    .unwrap_or("unknown panic payload");
220                Err(anyhow::anyhow!("database writer thread panicked: {detail}"))
221            });
222            clear_database_writer(id);
223            let _ = stopped_tx.send(result);
224        }) {
225        Ok(thread) => thread,
226        Err(error) => {
227            if install_globally {
228                clear_database_writer(id);
229            }
230            return Err(error).context("spawn database writer thread");
231        }
232    };
233    Ok(DatabaseWriterOwner {
234        writer,
235        thread: Some(thread),
236        stopped,
237    })
238}
239
240/// Refuse incompatible stores, rollback, and unreadable metadata before a job.
241pub(super) fn writer_schema_state(
242    path: &Path,
243    connection: &Connection,
244    label: &'static str,
245    observed_revision: &mut i64,
246) -> Result<()> {
247    let result: Result<()> = (|| {
248        let state = schema::read_schema_state(connection)?;
249        if state.revision < *observed_revision {
250            return Err(StoreSchemaMismatch {
251                found: state.revision,
252                supported: SCHEMA_VERSION,
253                reason: StoreSchemaMismatchReason::Rollback {
254                    previous: *observed_revision,
255                },
256            }
257            .into());
258        }
259        *observed_revision = state.revision;
260        state.ensure_supported()
261    })();
262    if let Err(error) = &result {
263        tracing::error!(
264            operation = label,
265            path = %path.display(),
266            error = %error,
267            "could not establish store compatibility; refusing the operation"
268        );
269    }
270    result.with_context(|| {
271        format!(
272            "check database compatibility before {label} at {}",
273            path.display()
274        )
275    })
276}
277
278pub(super) fn submit_database_write<T, F>(label: &'static str, operation: F) -> Result<T>
279where
280    T: Send + 'static,
281    F: FnOnce(&mut Connection) -> Result<T> + Send + 'static,
282{
283    let writer = database_writer_slot()
284        .lock()
285        .unwrap_or_else(PoisonError::into_inner)
286        .clone();
287    if let Some(writer) = writer {
288        writer.execute(label, operation)
289    } else {
290        // There is one way to write, and this is not it. In production the
291        // daemon installs the writer after `ControllerStoreGuard` establishes
292        // exclusivity, and it is the only process that writes; a caller
293        // reaching here has no exclusivity and would be competing with
294        // whatever does. This used to open `database_path()` directly, which
295        // meant any process without a writer silently wrote to — and migrated
296        // — the real user database as a side effect of doing something else.
297        bail!("database writer is not available for operation {label}")
298    }
299}
mj_controller/database/writer.rs

mj_controller/database/
writer.rs
