asched_core/
migration.rs

1//! Explicit import paths from schedulers that predate asched.
2//! ref: README.md#migrating-from-wsx
3
4use crate::routine::store::{
5    atomic_create, atomic_toml, project_key, read_text_limited, ProjectRoutines, RoutineStore,
6};
7use crate::{Project, ProjectRegistry, RegistryError, RegistryStore};
8use serde::{Deserialize, Serialize};
9use std::fs::{self, File, OpenOptions};
10use std::os::fd::AsRawFd;
11use std::os::unix::fs::PermissionsExt;
12use std::path::{Path, PathBuf};
13
14const WSX_IMPORT_TRANSACTION_VERSION: u32 = 1;
15
16pub fn default_wsx_paths() -> Result<(PathBuf, PathBuf), RegistryError> {
17    let root = dirs::config_dir()
18        .map(|path| path.join("wsx"))
19        .ok_or_else(|| RegistryError::Unavailable("no config directory".into()))?;
20    Ok((root.join("routines"), root.join("config.toml")))
21}
22
23#[derive(Debug, Clone, Serialize)]
24pub struct WsxImportPlan {
25    pub source_root: PathBuf,
26    pub source_config: PathBuf,
27    pub projects: Vec<WsxImportProject>,
28}
29
30#[derive(Debug, Clone, Serialize)]
31pub struct WsxImportProject {
32    pub project: Project,
33    pub routine_count: usize,
34    pub has_routine_file: bool,
35}
36
37#[derive(Debug, Clone, Serialize)]
38pub struct WsxImportResult {
39    pub registry: ProjectRegistry,
40    pub projects_registered: usize,
41    pub routine_files_imported: usize,
42    pub routines_imported: usize,
43}
44
45#[derive(Debug, Deserialize)]
46struct WsxConfig {
47    #[serde(default)]
48    projects: Vec<WsxProject>,
49}
50
51#[derive(Debug, Deserialize)]
52struct WsxProject {
53    name: String,
54    path: PathBuf,
55}
56
57#[derive(Debug, Serialize, Deserialize)]
58struct WsxImportTransaction {
59    version: u32,
60    registry_revision: u64,
61    projects_before: usize,
62    routines_imported: usize,
63    projects: Vec<Project>,
64    files: Vec<WsxImportFile>,
65}
66
67#[derive(Debug, Serialize, Deserialize)]
68struct WsxImportFile {
69    target: PathBuf,
70    contents: String,
71}
72
73pub fn plan_wsx_import(
74    source_root: &Path,
75    source_config: &Path,
76) -> Result<WsxImportPlan, RegistryError> {
77    let text = read_text_limited(source_config).map_err(|error| {
78        RegistryError::Io(format!("reading {}: {error}", source_config.display()))
79    })?;
80    let config: WsxConfig = toml::from_str(&text)
81        .map_err(|error| RegistryError::Corrupt(format!("{}: {error}", source_config.display())))?;
82    let mut projects = Vec::with_capacity(config.projects.len());
83    for source in config.projects {
84        let project = Project {
85            name: source.name,
86            working_dir: source.path,
87        }
88        .validated()?;
89        let routine_file = source_root
90            .join("projects")
91            .join(format!("{}.toml", project_key(&project.working_dir)));
92        let has_routine_file = routine_file.exists();
93        let routine_count = if has_routine_file {
94            load_source_routines(&routine_file, &project.working_dir)?
95                .routines
96                .len()
97        } else {
98            0
99        };
100        projects.push(WsxImportProject {
101            project,
102            routine_count,
103            has_routine_file,
104        });
105    }
106    projects.sort_by(|left, right| left.project.name.cmp(&right.project.name));
107    Ok(WsxImportPlan {
108        source_root: source_root.to_path_buf(),
109        source_config: source_config.to_path_buf(),
110        projects,
111    })
112}
113
114// ^ [[Crash-Safe wsx Import]]
115pub fn apply_wsx_import(
116    plan: &WsxImportPlan,
117    destination: &RegistryStore,
118    keep_enabled: bool,
119) -> Result<WsxImportResult, RegistryError> {
120    let _guard = destination.exclusive_lock()?;
121    let _daemon_guard = DaemonOfflineGuard::acquire(destination.root())?;
122    if let Some(result) = recover_wsx_import(destination)? {
123        return Ok(result);
124    }
125    let before = destination.load()?;
126    validate_destination(&before, plan, destination.root())?;
127
128    let mut files = Vec::new();
129    let mut routines_imported = 0;
130    for item in &plan.projects {
131        if !item.has_routine_file {
132            continue;
133        }
134        let key = project_key(&item.project.working_dir);
135        let source = plan
136            .source_root
137            .join("projects")
138            .join(format!("{key}.toml"));
139        let mut config = load_source_routines(&source, &item.project.working_dir)?;
140        if !keep_enabled {
141            for routine in &mut config.routines {
142                routine.enabled = false;
143            }
144        }
145        routines_imported += config.routines.len();
146        let target = RoutineStore::new(destination.root().to_path_buf(), &item.project.working_dir)
147            .map_err(|error| RegistryError::Validation(error.to_string()))?
148            .project_file();
149        let contents = toml::to_string_pretty(&config)
150            .map_err(|error| RegistryError::Corrupt(error.to_string()))?;
151        files.push(WsxImportFile { target, contents });
152    }
153
154    let transaction = WsxImportTransaction {
155        version: WSX_IMPORT_TRANSACTION_VERSION,
156        registry_revision: before.revision,
157        projects_before: before.projects.len(),
158        routines_imported,
159        projects: plan
160            .projects
161            .iter()
162            .map(|item| item.project.clone())
163            .collect(),
164        files,
165    };
166    atomic_toml(&wsx_transaction_path(destination.root()), &transaction)
167        .map_err(|error| RegistryError::Io(error.to_string()))?;
168
169    for file in &transaction.files {
170        if let Err(error) = atomic_create(&file.target, file.contents.as_bytes()) {
171            rollback_wsx_import(destination, &transaction)?;
172            return Err(if error.kind() == std::io::ErrorKind::AlreadyExists {
173                RegistryError::Validation(format!(
174                    "refusing to overwrite existing routine file {}",
175                    file.target.display()
176                ))
177            } else {
178                RegistryError::Io(error.to_string())
179            });
180        }
181    }
182    let registry = match destination.merge_locked(before.revision, transaction.projects.clone()) {
183        Ok(registry) => registry,
184        // ^ A registry rename may commit even when the following directory sync
185        // fails. Re-read the transaction before deciding whether rollback is safe.
186        Err(error) => match recover_wsx_import(destination)? {
187            Some(result) => return Ok(result),
188            None => return Err(error),
189        },
190    };
191    remove_transaction(destination.root())?;
192    Ok(WsxImportResult {
193        projects_registered: registry.projects.len() - before.projects.len(),
194        routine_files_imported: transaction.files.len(),
195        routines_imported,
196        registry,
197    })
198}
199
200fn recover_wsx_import(
201    destination: &RegistryStore,
202) -> Result<Option<WsxImportResult>, RegistryError> {
203    let path = wsx_transaction_path(destination.root());
204    if !path.exists() {
205        return Ok(None);
206    }
207    let text = read_text_limited(&path)
208        .map_err(|error| RegistryError::Io(format!("reading {}: {error}", path.display())))?;
209    let transaction: WsxImportTransaction = toml::from_str(&text)
210        .map_err(|error| RegistryError::Corrupt(format!("{}: {error}", path.display())))?;
211    validate_transaction(destination.root(), &transaction)?;
212    let registry = destination.load()?;
213    let committed = registry.revision > transaction.registry_revision
214        && transaction
215            .projects
216            .iter()
217            .all(|project| registry.projects.iter().any(|stored| stored == project));
218    if committed {
219        for file in &transaction.files {
220            ensure_transaction_file(file)?;
221        }
222        remove_transaction(destination.root())?;
223        Ok(Some(WsxImportResult {
224            projects_registered: registry
225                .projects
226                .len()
227                .saturating_sub(transaction.projects_before),
228            routine_files_imported: transaction.files.len(),
229            routines_imported: transaction.routines_imported,
230            registry,
231        }))
232    } else {
233        rollback_wsx_import(destination, &transaction)?;
234        Ok(None)
235    }
236}
237
238fn rollback_wsx_import(
239    destination: &RegistryStore,
240    transaction: &WsxImportTransaction,
241) -> Result<(), RegistryError> {
242    validate_transaction(destination.root(), transaction)?;
243    for file in &transaction.files {
244        if !file.target.exists() {
245            continue;
246        }
247        ensure_transaction_file(file)?;
248        fs::remove_file(&file.target).map_err(|error| {
249            RegistryError::Io(format!("removing {}: {error}", file.target.display()))
250        })?;
251    }
252    remove_transaction(destination.root())
253}
254
255fn ensure_transaction_file(file: &WsxImportFile) -> Result<(), RegistryError> {
256    let contents = read_text_limited(&file.target).map_err(|error| {
257        RegistryError::Io(format!("reading {}: {error}", file.target.display()))
258    })?;
259    if contents != file.contents {
260        return Err(RegistryError::Corrupt(format!(
261            "migration target changed after installation: {}",
262            file.target.display()
263        )));
264    }
265    Ok(())
266}
267
268fn validate_transaction(
269    root: &Path,
270    transaction: &WsxImportTransaction,
271) -> Result<(), RegistryError> {
272    if transaction.version != WSX_IMPORT_TRANSACTION_VERSION {
273        return Err(RegistryError::Corrupt(format!(
274            "unsupported wsx import transaction schema {}",
275            transaction.version
276        )));
277    }
278    for file in &transaction.files {
279        let valid = transaction.projects.iter().any(|project| {
280            file.target
281                == root
282                    .join("projects")
283                    .join(format!("{}.toml", project_key(&project.working_dir)))
284        });
285        if !valid {
286            return Err(RegistryError::Corrupt(format!(
287                "wsx import target is outside the transaction: {}",
288                file.target.display()
289            )));
290        }
291    }
292    Ok(())
293}
294
295fn wsx_transaction_path(root: &Path) -> PathBuf {
296    root.join("migrations").join("wsx-import-v1.toml")
297}
298
299pub(crate) fn ensure_no_pending_wsx_import(root: &Path) -> Result<(), RegistryError> {
300    let path = wsx_transaction_path(root);
301    if path.exists() {
302        return Err(RegistryError::Unavailable(format!(
303            "unfinished wsx import at {}; run 'asched migrate wsx' to recover it before starting the daemon",
304            path.display()
305        )));
306    }
307    Ok(())
308}
309
310fn remove_transaction(root: &Path) -> Result<(), RegistryError> {
311    let path = wsx_transaction_path(root);
312    if path.exists() {
313        fs::remove_file(&path)
314            .map_err(|error| RegistryError::Io(format!("removing {}: {error}", path.display())))?;
315        if let Some(parent) = path.parent() {
316            File::open(parent)?.sync_all()?;
317        }
318    }
319    Ok(())
320}
321
322struct DaemonOfflineGuard(File);
323
324impl DaemonOfflineGuard {
325    fn acquire(root: &Path) -> Result<Self, RegistryError> {
326        let path = root.join("daemon-v1.lock");
327        let file = OpenOptions::new()
328            .read(true)
329            .write(true)
330            .create(true)
331            .truncate(false)
332            .open(&path)
333            .map_err(|error| RegistryError::Io(format!("opening {}: {error}", path.display())))?;
334        file.set_permissions(fs::Permissions::from_mode(0o600))?;
335        let locked = unsafe { libc::flock(file.as_raw_fd(), libc::LOCK_EX | libc::LOCK_NB) };
336        if locked != 0 {
337            return Err(RegistryError::Unavailable(
338                "stop the asched daemon before importing wsx routines".into(),
339            ));
340        }
341        Ok(Self(file))
342    }
343}
344
345impl Drop for DaemonOfflineGuard {
346    fn drop(&mut self) {
347        unsafe {
348            libc::flock(self.0.as_raw_fd(), libc::LOCK_UN);
349        }
350    }
351}
352
353fn validate_destination(
354    current: &ProjectRegistry,
355    plan: &WsxImportPlan,
356    destination_root: &Path,
357) -> Result<(), RegistryError> {
358    for item in &plan.projects {
359        if let Some(existing) = current.projects.iter().find(|existing| {
360            existing.name == item.project.name || existing.working_dir == item.project.working_dir
361        }) {
362            if existing != &item.project {
363                return Err(RegistryError::Validation(format!(
364                    "imported project '{}' conflicts with '{}' ({})",
365                    item.project.name,
366                    existing.name,
367                    existing.working_dir.display()
368                )));
369            }
370        }
371        if item.has_routine_file {
372            let target = destination_root
373                .join("projects")
374                .join(format!("{}.toml", project_key(&item.project.working_dir)));
375            if target.exists() {
376                return Err(RegistryError::Validation(format!(
377                    "refusing to overwrite existing routine file {}",
378                    target.display()
379                )));
380            }
381        }
382    }
383    Ok(())
384}
385
386fn load_source_routines(path: &Path, working_dir: &Path) -> Result<ProjectRoutines, RegistryError> {
387    let text = read_text_limited(path)
388        .map_err(|error| RegistryError::Io(format!("reading {}: {error}", path.display())))?;
389    let mut config: ProjectRoutines = toml::from_str(&text)
390        .map_err(|error| RegistryError::Corrupt(format!("{}: {error}", path.display())))?;
391    if !matches!(config.version, 1 | crate::routine::PROJECT_CONFIG_VERSION) {
392        return Err(RegistryError::Corrupt(format!(
393            "{} uses unsupported routine schema {}",
394            path.display(),
395            config.version
396        )));
397    }
398    if config.project_path != working_dir {
399        return Err(RegistryError::Corrupt(format!(
400            "{} stores project {}, expected {}",
401            path.display(),
402            config.project_path.display(),
403            working_dir.display()
404        )));
405    }
406    for routine in &mut config.routines {
407        *routine = routine
408            .clone()
409            .validated()
410            .map_err(|error| RegistryError::Corrupt(error.to_string()))?;
411    }
412    Ok(config)
413}
414
415#[cfg(test)]
416#[path = "migration_contract_tests.rs"]
417mod migration_contract_tests;
asched_core/migration.rs

asched_core/
migration.rs
