mj_controller/
target_storage.rs

1//! The one owner of "is this filesystem on a target full".
2//!
3//! The daemon's capacity service measures, on its usual cadence, each
4//! filesystem Mjolnir writes to on each host: worker roots, staged profile
5//! homes, project directories and managed clones, caches, container storage
6//! and `/tmp`, one record per filesystem. A target command that fails with
7//! "No space left on device" marks the filesystem of the path it names, or,
8//! when it names none, holds every filesystem on the host until the new
9//! measurement it asks for says which one is full. Everything else asks this
10//! board: a write checks the filesystem it lands on, recovery waits on the
11//! filesystem it needs, and the runtime feed, API and web viewer publish
12//! what it says.
13//!
14//! Hosts are named the way [`mj_core::targets::storage`] names them: `local`
15//! for the daemon's machine, otherwise the SSH host without its user.
16
17use std::collections::BTreeMap;
18use std::path::PathBuf;
19use std::sync::{Arc, Mutex, OnceLock, PoisonError};
20
21use anyhow::Result;
22use mj_core::refusal::Refusal;
23use mj_core::targets::TargetLocator;
24use mj_core::targets::storage::{
25    ContainerStorage, FilesystemSpace, HostStorageSample, LOCAL_STORAGE_HOST, TargetStorageView,
26    local_home_path, storage_host_of_destination,
27};
28
29/// A failed write keeps its filesystem full until a measurement taken this
30/// long after it says otherwise. A probe already in flight when the write
31/// failed may describe the disk from before the failure.
32const NO_SPACE_HOLD_SECONDS: u64 = 15;
33
34#[derive(Default)]
35struct HostEntry {
36    home: Option<String>,
37    filesystems: Vec<FilesystemSpace>,
38    sampled_at: Option<u64>,
39    /// Failures attributed to one filesystem, by mount point.
40    no_space: BTreeMap<String, (u64, String)>,
41    /// A failure that named no path this board could place.
42    unattributed: Option<(u64, String)>,
43}
44
45impl HostEntry {
46    fn view(&self, host: &str) -> TargetStorageView {
47        TargetStorageView::evaluate(
48            host,
49            self.home.clone(),
50            &self.filesystems,
51            self.sampled_at,
52            |space| {
53                self.no_space
54                    .get(&space.mount)
55                    .map(|(_, detail)| detail.clone())
56            },
57            self.unattributed.as_ref().map(|(_, detail)| detail.clone()),
58        )
59    }
60
61    fn superseded(&self, at: u64) -> bool {
62        self.sampled_at
63            .is_some_and(|sampled| sampled >= at.saturating_add(NO_SPACE_HOLD_SECONDS))
64    }
65}
66
67struct Board {
68    hosts: Mutex<BTreeMap<String, HostEntry>>,
69    refresh: Mutex<Option<tokio::sync::mpsc::Sender<()>>>,
70    /// Advances whenever a filesystem's verdict or free space changes.
71    generation: tokio::sync::watch::Sender<u64>,
72}
73
74impl Default for Board {
75    fn default() -> Self {
76        Self {
77            hosts: Mutex::default(),
78            refresh: Mutex::default(),
79            generation: tokio::sync::watch::channel(0).0,
80        }
81    }
82}
83
84impl Board {
85    fn hosts(&self) -> std::sync::MutexGuard<'_, BTreeMap<String, HostEntry>> {
86        self.hosts.lock().unwrap_or_else(PoisonError::into_inner)
87    }
88
89    fn changed(&self) {
90        self.generation.send_modify(|generation| *generation += 1);
91    }
92
93    fn request_refresh(&self) {
94        if let Some(refresh) = self
95            .refresh
96            .lock()
97            .unwrap_or_else(PoisonError::into_inner)
98            .as_ref()
99        {
100            // A full channel already holds a refresh request.
101            let _ = refresh.try_send(());
102        }
103    }
104}
105
106/// One board per data directory, so isolated tests in one process do not
107/// share hosts. A daemon has exactly one.
108fn board() -> Arc<Board> {
109    type Boards = Mutex<BTreeMap<PathBuf, Arc<Board>>>;
110    static BOARDS: OnceLock<Boards> = OnceLock::new();
111    BOARDS
112        .get_or_init(Mutex::default)
113        .lock()
114        .unwrap_or_else(PoisonError::into_inner)
115        .entry(mj_core::config::data_dir())
116        .or_default()
117        .clone()
118}
119
120fn now_epoch_seconds() -> u64 {
121    std::time::SystemTime::now()
122        .duration_since(std::time::UNIX_EPOCH)
123        .unwrap_or_default()
124        .as_secs()
125}
126
127/// Make this process the owner that executors report full disks to.
128pub(crate) fn install_no_space_observer() {
129    mj_core::targets::storage::set_no_space_observer(|host, detail| {
130        observe_no_space(host, detail);
131    });
132}
133
134/// Let a no-space observation wake the capacity service for a new reading.
135pub(crate) fn attach_refresh(refresh: tokio::sync::mpsc::Sender<()>) {
136    *board()
137        .refresh
138        .lock()
139        .unwrap_or_else(PoisonError::into_inner) = Some(refresh);
140}
141
142/// Install a measurement. It replaces each sampled host's filesystems, and
143/// supersedes the full-disk failures it was taken after.
144pub(crate) fn record_samples(samples: &[HostStorageSample], sampled_at: u64) {
145    if samples.is_empty() {
146        return;
147    }
148    let board = board();
149    let mut changed = false;
150    {
151        let mut hosts = board.hosts();
152        for sample in samples {
153            let entry = hosts.entry(sample.host.clone()).or_default();
154            let before = entry.view(&sample.host);
155            entry.home = sample.home.clone().or(entry.home.take());
156            entry.filesystems = sample.filesystems.clone();
157            entry.sampled_at = Some(sampled_at);
158            let hold = |at: &u64| sampled_at < at.saturating_add(NO_SPACE_HOLD_SECONDS);
159            entry.no_space.retain(|_, (at, _)| hold(at));
160            if entry.unattributed.as_ref().is_some_and(|(at, _)| !hold(at)) {
161                entry.unattributed = None;
162            }
163            let after = entry.view(&sample.host);
164            changed |= before.filesystems != after.filesystems
165                || before.unattributed_no_space != after.unattributed_no_space;
166        }
167    }
168    if changed {
169        board.changed();
170    }
171}
172
173/// A write on `host` failed because a disk is full. The filesystem of the
174/// path the failure names is marked; a failure that names no path holds the
175/// whole host until the measurement it asks for.
176pub(crate) fn observe_no_space(host: &str, detail: &str) {
177    let path = mj_core::targets::storage::no_space_path(detail);
178    observe_no_space_at(host, path.as_deref(), detail, now_epoch_seconds());
179}
180
181/// A write to `path` on `host` failed for lack of space at `at` (epoch
182/// seconds). A measurement taken after that already says what the disk holds
183/// now, so an old failure, such as a dead worker's exit record, does not
184/// override it.
185pub(crate) fn observe_no_space_at(host: &str, path: Option<&str>, detail: &str, at: u64) {
186    let board = board();
187    let changed = {
188        let mut hosts = board.hosts();
189        let entry = hosts.entry(host.to_owned()).or_default();
190        if entry.superseded(at) {
191            tracing::debug!(%host, %detail, "a later measurement supersedes this full-disk failure");
192            return;
193        }
194        let before = entry.view(host);
195        let mount = path.and_then(|path| {
196            before
197                .filesystem_for(path)
198                .map(|filesystem| filesystem.space.mount.clone())
199        });
200        match mount {
201            Some(mount) => {
202                entry.no_space.insert(mount, (at, detail.to_owned()));
203            }
204            None => entry.unattributed = Some((at, detail.to_owned())),
205        }
206        before != entry.view(host)
207    };
208    tracing::warn!(%host, ?path, %detail, "a write failed because the target's disk is full");
209    if changed {
210        board.changed();
211    }
212    board.request_refresh();
213}
214
215/// What the board says about `host`, if it has heard anything.
216pub fn view(host: &str) -> Option<TargetStorageView> {
217    board().hosts().get(host).map(|entry| entry.view(host))
218}
219
220/// Every host the board knows, for publication.
221pub fn views() -> Vec<TargetStorageView> {
222    board()
223        .hosts()
224        .iter()
225        .map(|(host, entry)| entry.view(host))
226        .collect()
227}
228
229/// Wakes whenever a filesystem's verdict or free space changes.
230pub fn subscribe() -> tokio::sync::watch::Receiver<u64> {
231    board().generation.subscribe()
232}
233
234/// The host a target writes to.
235pub fn storage_host(locator: &TargetLocator) -> String {
236    match locator {
237        TargetLocator::LocalBare { .. }
238        | TargetLocator::LocalPodman { .. }
239        | TargetLocator::LocalDocker { .. }
240        | TargetLocator::AppleContainer { .. } => LOCAL_STORAGE_HOST.to_owned(),
241        TargetLocator::AwsEc2 { ssh, .. }
242        | TargetLocator::SshBare { ssh, .. }
243        | TargetLocator::SshPodman { ssh, .. }
244        | TargetLocator::SshDocker { ssh, .. } => {
245            storage_host_of_destination(Some(&ssh.destination))
246        }
247    }
248}
249
250/// Where on its host a write to `path` on `locator` lands. A path inside a
251/// container lands in the engine's storage; Apple's runtime keeps its
252/// storage in a VM image this board does not measure.
253fn host_path(locator: &TargetLocator, path: &str) -> Option<String> {
254    match locator {
255        TargetLocator::LocalBare { .. }
256        | TargetLocator::AwsEc2 { .. }
257        | TargetLocator::SshBare { .. } => Some(path.to_owned()),
258        TargetLocator::LocalPodman { .. } => Some(local_home_path(ContainerStorage::Podman.path())),
259        TargetLocator::SshPodman { .. } => Some(ContainerStorage::Podman.path().to_owned()),
260        TargetLocator::LocalDocker { .. } | TargetLocator::SshDocker { .. } => {
261            Some(ContainerStorage::Docker.path().to_owned())
262        }
263        TargetLocator::AppleContainer { .. } => None,
264    }
265}
266
267/// Refuse a write of `bytes` (zero when the size is unknown) to `path` on
268/// `locator` that would leave its filesystem with less than the reserve. A
269/// filesystem the board has not measured is not refused: a write that then
270/// fails is still classified.
271pub fn ensure_room(locator: &TargetLocator, path: &str, bytes: u64, what: &str) -> Result<()> {
272    ensure_room_for(locator, [path], || Ok(bytes), what)
273}
274
275/// [`ensure_room`] for a write that spans several directories, such as a
276/// harness install that fills its cache and the staged profile home. The
277/// size is measured only when a filesystem the write lands on is known, so
278/// an unmeasured target costs nothing.
279pub fn ensure_room_for<'a>(
280    locator: &TargetLocator,
281    paths: impl IntoIterator<Item = &'a str>,
282    bytes: impl FnOnce() -> Result<u64>,
283    what: &str,
284) -> Result<()> {
285    let host = storage_host(locator);
286    let Some(view) = view(&host) else {
287        return Ok(());
288    };
289    let paths = paths
290        .into_iter()
291        .filter_map(|path| host_path(locator, path))
292        .filter(|path| view.filesystem_for(path).is_some() || view.unattributed_no_space.is_some())
293        .collect::<Vec<_>>();
294    if paths.is_empty() {
295        return Ok(());
296    }
297    let bytes = bytes()?;
298    for path in &paths {
299        if let Some(refusal) = view.refuse_write(path, bytes, what) {
300            tracing::warn!(%host, %path, bytes, what, "refused a write to a full filesystem: {refusal}");
301            return Err(Refusal::precondition(refusal)
302                .with_code(STORAGE_FULL_CODE)
303                .into());
304        }
305    }
306    Ok(())
307}
308
309/// Where the worker installs managed harnesses on a target: its XDG cache.
310pub fn harness_cache_path(locator: &TargetLocator) -> String {
311    const HARNESS_CACHE: &str = ".cache/mjolnir/harnesses";
312    match locator {
313        TargetLocator::LocalBare { .. } => local_home_path(HARNESS_CACHE),
314        _ => HARNESS_CACHE.to_owned(),
315    }
316}
317
318/// The refusal code carried by [`ensure_room`] refusals.
319pub const STORAGE_FULL_CODE: &str = "target_storage_full";
320
321/// Whether `error` is a write refused or failed because a disk is full.
322pub fn is_storage_full(error: &anyhow::Error) -> bool {
323    Refusal::of(error).is_some_and(|refusal| refusal.code() == Some(STORAGE_FULL_CODE))
324        || mj_core::targets::storage::reports_no_space(&format!("{error:#}"))
325}
326
327/// The host and sentence for a full filesystem one of `paths` on `host`
328/// lands on.
329pub fn full_problem<'a>(host: &str, paths: impl IntoIterator<Item = &'a str>) -> Option<String> {
330    view(host)?.problem_for(paths)
331}
332
333/// The host and worker root of a session's recorded target: where its
334/// worker keeps its journal, and where a restart stages and writes.
335pub fn session_worker_root(
336    locator: &mj_core::state::TargetLocator,
337    session_id: &str,
338) -> (String, String) {
339    let paths = mj_core::targets::storage::session_storage_paths(locator, session_id, None);
340    (paths.host, paths.worker_root)
341}
342
343/// The host and sentence while the filesystem a session's worker root is on
344/// is full: recovery waits on exactly this.
345pub fn worker_root_problem(
346    locator: &mj_core::state::TargetLocator,
347    session_id: &str,
348) -> Option<(String, String)> {
349    let (host, worker_root) = session_worker_root(locator, session_id);
350    let problem = full_problem(&host, [worker_root.as_str()])?;
351    Some((host, problem))
352}
353
354/// The sentence a session shows instead of "unreachable" while one of its
355/// own filesystems is full: its worker root, workspace, profile home or
356/// `/tmp`.
357pub fn session_problem(session: &mj_core::state::SessionRecord) -> Option<String> {
358    let checkout = session.checkout();
359    located_session_problem(
360        session.target.as_ref()?,
361        &session.id,
362        checkout.project_directory(),
363    )
364}
365
366fn located_session_problem(
367    locator: &mj_core::state::TargetLocator,
368    session_id: &str,
369    project_directory: Option<&std::path::Path>,
370) -> Option<String> {
371    let paths =
372        mj_core::targets::storage::session_storage_paths(locator, session_id, project_directory);
373    full_problem(&paths.host, paths.all())
374}
375
376#[cfg(test)]
377pub(crate) mod tests {
378    use super::*;
379    use mj_core::targets::storage::WRITE_RESERVE_BYTES;
380
381    /// One host with `/` holding the worker roots and `~/Projects` on a
382    /// filesystem of its own, as on precision-3260.
383    pub(crate) fn sample(host: &str, root_free: u64, projects_free: u64) -> HostStorageSample {
384        HostStorageSample {
385            host: host.to_owned(),
386            home: Some("/home/jonathan".into()),
387            filesystems: vec![
388                FilesystemSpace {
389                    mount: "/".into(),
390                    available_bytes: root_free,
391                    total_bytes: 500 << 30,
392                    reserved_bytes: 25 << 30,
393                    paths: vec![
394                        "/home/jonathan/.local/share/hel/workers".into(),
395                        "/home/jonathan/.local/share/hel/profiles".into(),
396                        "/home/jonathan/.cache/mjolnir".into(),
397                        "/tmp".into(),
398                    ],
399                },
400                FilesystemSpace {
401                    mount: "/home/jonathan/Projects".into(),
402                    available_bytes: projects_free,
403                    total_bytes: 1000 << 30,
404                    reserved_bytes: 0,
405                    paths: vec!["/home/jonathan/Projects".into()],
406                },
407            ],
408        }
409    }
410
411    pub(crate) fn ssh_bare(host: &str) -> TargetLocator {
412        TargetLocator::SshBare {
413            ssh: mj_core::targets::SshTarget {
414                destination: format!("jonathan@{host}"),
415                ssh_args: Vec::new(),
416            },
417            workspace: "/home/jonathan/Projects/app/.mj/clones/abc".into(),
418            worker_id: None,
419        }
420    }
421
422    const WORKER: &str = ".local/share/hel/workers/abc/hel.prepared-x.next";
423    const CLONE: &str = "/home/jonathan/Projects/app/.mj/clones/abc/restore.hel.zip";
424
425    // The board is shared by every test in this process, so each test names
426    // its own host.
427    #[test]
428    fn each_write_is_judged_by_the_filesystem_it_lands_on() {
429        let host = "precision-3260";
430        let target = ssh_bare(host);
431        // Not measured yet: unknown is not a reason to refuse.
432        ensure_room(&target, WORKER, 100 << 20, "stage the worker").unwrap();
433
434        // A full ~/Projects refuses a restore into a clone there, but does not
435        // block worker staging on a healthy /.
436        record_samples(&[sample(host, 40 << 30, 0)], now_epoch_seconds());
437        ensure_room(&target, WORKER, 100 << 20, "stage the worker").unwrap();
438        let error = ensure_room(&target, CLONE, 100 << 20, "restore the checkpoint").unwrap_err();
439        assert!(is_storage_full(&error));
440        let text = format!("{error:#}");
441        assert!(
442            text.contains("precision-3260 has 0 B free on /home/jonathan/Projects"),
443            "{text}"
444        );
445        assert!(
446            text.contains("free space on precision-3260 to continue"),
447            "{text}"
448        );
449
450        // And the other way round: a full / blocks staging, not the restore.
451        record_samples(&[sample(host, 0, 40 << 30)], now_epoch_seconds());
452        let error = ensure_room(&target, WORKER, 100 << 20, "stage the worker").unwrap_err();
453        let text = format!("{error:#}");
454        assert!(
455            text.contains(
456                "precision-3260 has 0 B free on / (the filesystem reserves 26.84 GB more for root)"
457            ),
458            "{text}"
459        );
460        ensure_room(&target, CLONE, 100 << 20, "restore the checkpoint").unwrap();
461
462        // Room for the write, but not for the reserve beside it.
463        record_samples(
464            &[sample(host, WRITE_RESERVE_BYTES + (50 << 20), 40 << 30)],
465            now_epoch_seconds(),
466        );
467        assert!(ensure_room(&target, WORKER, 100 << 20, "stage the worker").is_err());
468        ensure_room(&target, WORKER, 10 << 20, "write a config").unwrap();
469    }
470
471    #[test]
472    fn a_failed_write_marks_the_filesystem_it_names_until_a_later_measurement() {
473        let host = "observed-3260";
474        let target = ssh_bare(host);
475        let now = now_epoch_seconds();
476        record_samples(&[sample(host, 20 << 30, 20 << 30)], now);
477        let mut changes = subscribe();
478        changes.borrow_and_update();
479
480        observe_no_space(
481            host,
482            "scp: write remote \"/home/jonathan/Projects/app/.mj/clones/abc/x\": No space left on device",
483        );
484        assert!(changes.has_changed().unwrap());
485        assert!(ensure_room(&target, CLONE, 0, "restore").is_err());
486        ensure_room(&target, WORKER, 0, "stage the worker").unwrap();
487
488        // A probe that may have started before the failure does not clear it.
489        record_samples(&[sample(host, 20 << 30, 20 << 30)], now_epoch_seconds());
490        assert!(ensure_room(&target, CLONE, 0, "restore").is_err());
491        record_samples(
492            &[sample(host, 20 << 30, 20 << 30)],
493            now_epoch_seconds() + NO_SPACE_HOLD_SECONDS,
494        );
495        ensure_room(&target, CLONE, 0, "restore").unwrap();
496    }
497
498    #[test]
499    fn a_failure_naming_no_path_holds_the_host_until_measured() {
500        let host = "unattributed-3260";
501        let target = ssh_bare(host);
502        record_samples(&[sample(host, 20 << 30, 20 << 30)], now_epoch_seconds());
503        observe_no_space(
504            host,
505            "Error response from daemon: ENOSPC: no space left on device",
506        );
507        assert!(ensure_room(&target, WORKER, 0, "stage the worker").is_err());
508        assert!(ensure_room(&target, CLONE, 0, "restore").is_err());
509        // The measurement it asked for decides: here, both have room.
510        record_samples(
511            &[sample(host, 20 << 30, 20 << 30)],
512            now_epoch_seconds() + NO_SPACE_HOLD_SECONDS,
513        );
514        ensure_room(&target, WORKER, 0, "stage the worker").unwrap();
515    }
516
517    #[test]
518    fn a_session_is_disk_full_only_when_one_of_its_own_filesystems_is() {
519        let host = "session-3260";
520        let session = mj_core::state::TargetLocator::SshBare {
521            host: host.into(),
522            workspace: "/home/jonathan/Projects/app/.mj/clones/abc".into(),
523            worker_id: None,
524        };
525        let elsewhere = mj_core::state::TargetLocator::SshBare {
526            host: host.into(),
527            workspace: "/srv/work/abc".into(),
528            worker_id: None,
529        };
530        record_samples(&[sample(host, 40 << 30, 0)], now_epoch_seconds());
531        let problem = located_session_problem(&session, "abc", None).unwrap();
532        assert!(
533            problem.starts_with("disk full: session-3260 has 0 B free on /home/jonathan/Projects"),
534            "{problem}"
535        );
536        assert!(located_session_problem(&elsewhere, "abc", None).is_none());
537    }
538}
mj_controller/target_storage.rs

mj_controller/
target_storage.rs
