brokk-mj-controller 2.12.0

Daemon-side controller, session manager, and web server for Mjolnir
Documentation 
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100

use super::*;

/// Remove checkpoint archives installed by a process that exited before its
/// database transaction committed. Call this only while holding the
/// machine-wide controller-store guard and before starting background work.
pub fn reconcile_managed_checkpoint_archives() -> Result<usize> {
    let mut state = crate::database::load_state()?;
    // Include operation-owned recovery copies even after a ready destination
    // installs a newer ordinary checkpoint.
    for operation in crate::database::load_move_operations()? {
        if operation.retains_checkpoint()
            && let Some(checkpoint) = operation.checkpoint
            && let Some(mut session) = state.sessions.get(&operation.selection.session_id).cloned()
        {
            session.checkpoint = Some(checkpoint);
            state
                .sessions
                .insert(format!("move:{}", operation.operation_id), session);
        }
    }
    reconcile_managed_checkpoint_archives_in(&sessions_dir(), &state)
}

pub(super) fn reconcile_managed_checkpoint_archives_in(
    directory: &Path,
    state: &State,
) -> Result<usize> {
    if !directory.exists() {
        return Ok(0);
    }
    let referenced_names = state
        .sessions
        .values()
        .filter_map(|session| session.checkpoint.as_ref())
        .filter_map(|checkpoint| checkpoint.archive_path.file_name())
        .map(ToOwned::to_owned)
        .collect::<BTreeSet<_>>();
    let mut removed = 0;
    for entry in std::fs::read_dir(directory)
        .with_context(|| format!("scan checkpoint directory {}", directory.display()))?
    {
        let entry = entry?;
        let file_type = entry.file_type()?;
        if !file_type.is_file()
            || !is_managed_checkpoint_archive_name(&entry.file_name())
            || referenced_names.contains(&entry.file_name())
        {
            continue;
        }
        std::fs::remove_file(entry.path()).with_context(|| {
            format!(
                "remove unreferenced managed checkpoint {}",
                entry.path().display()
            )
        })?;
        removed += 1;
    }
    Ok(removed)
}

/// The session and checkpoint generation a managed archive file name names.
///
/// Managed checkpoints are named `<session_id>-<frontier>-archive-<32 hex>`,
/// so the file name alone says which session a checkpoint belongs to and which
/// of that session's checkpoints is the newest.
#[derive(Debug, Clone, PartialEq, Eq)]
pub(crate) struct ManagedCheckpointArchiveName {
    pub session_id: String,
    pub frontier: u64,
}

/// Parse a managed checkpoint archive file name, or `None` when the name is
/// not one Mjolnir wrote.
pub(crate) fn managed_checkpoint_archive_name(
    name: &OsStr,
) -> Option<ManagedCheckpointArchiveName> {
    let stem = name
        .to_str()
        .and_then(|name| name.strip_suffix(".hel.zip"))?;
    let (frontier_prefix, nonce) = stem.rsplit_once("-archive-")?;
    if nonce.len() != 32
        || !nonce
            .bytes()
            .all(|byte| byte.is_ascii_hexdigit() && !byte.is_ascii_uppercase())
    {
        return None;
    }
    let (session_id, frontier) = frontier_prefix.rsplit_once('-')?;
    if session_id.is_empty() || mj_core::config::validate_id("session", session_id).is_err() {
        return None;
    }
    Some(ManagedCheckpointArchiveName {
        session_id: session_id.to_owned(),
        frontier: frontier.parse::<u64>().ok()?,
    })
}

pub(super) fn is_managed_checkpoint_archive_name(name: &OsStr) -> bool {
    managed_checkpoint_archive_name(name).is_some()
}