asched_core/routine/
store.rs

1use super::{
2    Routine, RoutineError, RunCause, RunRecord, MAX_EVENT_RECEIPTS, PROJECT_CONFIG_VERSION,
3    RUNTIME_STATE_VERSION,
4};
5use serde::{Deserialize, Serialize};
6use std::collections::{BTreeMap, BTreeSet};
7use std::fs::{self, File, OpenOptions};
8use std::io::{Read, Write};
9use std::os::unix::fs::PermissionsExt;
10use std::path::{Path, PathBuf};
11use std::sync::{Mutex, OnceLock};
12
13static RUNTIME_WRITE_LOCK: OnceLock<Mutex<()>> = OnceLock::new();
14pub(crate) const MAX_STATE_FILE_BYTES: u64 = 64 * 1024 * 1024;
15
16#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
17pub struct ProjectRoutines {
18    pub version: u32,
19    pub revision: u64,
20    pub project_path: PathBuf,
21    #[serde(default)]
22    pub routines: Vec<Routine>,
23}
24
25#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
26pub struct EventReceipt {
27    pub kind: String,
28    pub event_id: String,
29}
30
31#[derive(Debug, Clone, Default, Serialize, Deserialize)]
32pub struct RuntimeState {
33    pub version: u32,
34    /// Greatest claimed epoch minute per routine. One entry per routine bounds
35    /// restart-safe scheduling state and rejects duplicate slots after rollback.
36    #[serde(default, deserialize_with = "deserialize_claims")]
37    pub claims: BTreeMap<String, i64>,
38    #[serde(default)]
39    pub runs: BTreeMap<String, Vec<RunRecord>>,
40    #[serde(default)]
41    pub event_receipts: Vec<EventReceipt>,
42}
43
44impl RuntimeState {
45    pub fn has_event_receipt(&self, kind: &str, event_id: &str) -> bool {
46        self.event_receipts
47            .iter()
48            .any(|receipt| receipt.kind == kind && receipt.event_id == event_id)
49    }
50
51    pub fn record_event_receipt(&mut self, kind: String, event_id: String) {
52        self.event_receipts.push(EventReceipt { kind, event_id });
53        if self.event_receipts.len() > MAX_EVENT_RECEIPTS {
54            self.event_receipts
55                .drain(0..self.event_receipts.len() - MAX_EVENT_RECEIPTS);
56        }
57    }
58}
59
60#[derive(Clone)]
61// ^ [[Durable Scheduler State and Process Cleanup]]
62pub struct RoutineStore {
63    root: PathBuf,
64    project: PathBuf,
65    key: String,
66}
67
68impl RoutineStore {
69    pub fn new(root: PathBuf, project: &Path) -> Result<Self, RoutineError> {
70        let project = canonical_working_dir(project)?;
71        let key = project_key(&project);
72        Ok(Self { root, project, key })
73    }
74
75    pub fn default_root() -> Result<PathBuf, RoutineError> {
76        crate::RegistryStore::default_root()
77            .map_err(|error| RoutineError::Unavailable(error.to_string()))
78    }
79
80    pub fn project(&self) -> &Path {
81        &self.project
82    }
83    pub fn key(&self) -> &str {
84        &self.key
85    }
86    pub fn project_file(&self) -> PathBuf {
87        self.root
88            .join("projects")
89            .join(format!("{}.toml", self.key))
90    }
91    pub fn runtime_file(&self) -> PathBuf {
92        self.root.join("runtime").join(format!("{}.toml", self.key))
93    }
94    pub fn logs_dir(&self) -> PathBuf {
95        self.root.join("logs").join(&self.key)
96    }
97    pub fn socket_path(&self) -> PathBuf {
98        self.root.join("daemon-v1.sock")
99    }
100    pub fn lock_path(&self) -> PathBuf {
101        self.root.join("daemon-v1.lock")
102    }
103
104    pub fn load(&self) -> Result<ProjectRoutines, RoutineError> {
105        let path = self.project_file();
106        if !path.exists() {
107            return Ok(ProjectRoutines {
108                version: PROJECT_CONFIG_VERSION,
109                revision: 0,
110                project_path: self.project.clone(),
111                routines: vec![],
112            });
113        }
114        let text = read_text_limited(&path)?;
115        let config: ProjectRoutines = toml::from_str(&text)
116            .map_err(|e| RoutineError::Corrupt(format!("{}: {e}", path.display())))?;
117        if !matches!(config.version, 1 | PROJECT_CONFIG_VERSION) {
118            return Err(RoutineError::Corrupt(format!(
119                "unsupported routine schema {}",
120                config.version
121            )));
122        }
123        if config.project_path != self.project {
124            return Err(RoutineError::ProjectCollision {
125                expected: self.project.clone(),
126                stored: config.project_path,
127            });
128        }
129        validate_unique(&config.routines)?;
130        Ok(config)
131    }
132
133    pub fn save(
134        &self,
135        mut config: ProjectRoutines,
136        expected: u64,
137    ) -> Result<ProjectRoutines, RoutineError> {
138        let current = self.load()?;
139        if current.revision != expected {
140            return Err(RoutineError::Conflict {
141                expected,
142                actual: current.revision,
143            });
144        }
145        config.version = PROJECT_CONFIG_VERSION;
146        config.project_path = self.project.clone();
147        config.revision = expected
148            .checked_add(1)
149            .ok_or_else(|| RoutineError::Corrupt("revision overflow".into()))?;
150        validate_unique(&config.routines)?;
151        atomic_toml(&self.project_file(), &config)?;
152        Ok(config)
153    }
154
155    pub fn load_runtime(&self) -> Result<RuntimeState, RoutineError> {
156        let path = self.runtime_file();
157        if !path.exists() {
158            return Ok(RuntimeState {
159                version: RUNTIME_STATE_VERSION,
160                ..Default::default()
161            });
162        }
163        let text = read_text_limited(&path)?;
164        let mut state: RuntimeState = toml::from_str(&text)
165            .map_err(|e| RoutineError::Corrupt(format!("{}: {e}", path.display())))?;
166        if !matches!(state.version, 1 | RUNTIME_STATE_VERSION) {
167            return Err(RoutineError::Corrupt("unsupported runtime schema".into()));
168        }
169        if state.version == 1 {
170            for runs in state.runs.values_mut() {
171                for run in runs {
172                    if let Some(minute) = run.scheduled_epoch_minute {
173                        run.cause = RunCause::Cron {
174                            scheduled_epoch_minute: minute,
175                        };
176                    }
177                }
178            }
179        }
180        state.version = RUNTIME_STATE_VERSION;
181        Ok(state)
182    }
183
184    pub fn save_runtime(&self, state: &RuntimeState) -> Result<(), RoutineError> {
185        let mut state = state.clone();
186        state.version = RUNTIME_STATE_VERSION;
187        atomic_toml(&self.runtime_file(), &state)
188    }
189
190    pub fn admit_event(
191        &self,
192        kind: &str,
193        event_id: &str,
194        records: &[RunRecord],
195    ) -> Result<bool, RoutineError> {
196        self.modify_runtime(|state| {
197            if state.has_event_receipt(kind, event_id) {
198                return Ok(false);
199            }
200            state.record_event_receipt(kind.to_string(), event_id.to_string());
201            for record in records {
202                state
203                    .runs
204                    .entry(record.routine.clone())
205                    .or_default()
206                    .push(record.clone());
207            }
208            Ok(true)
209        })
210    }
211
212    pub fn claim(&self, routine: &str, epoch_minute: i64) -> Result<bool, RoutineError> {
213        self.modify_runtime(|state| {
214            if state
215                .claims
216                .get(routine)
217                .is_some_and(|claimed| epoch_minute <= *claimed)
218            {
219                return Ok(false);
220            }
221            state.claims.insert(routine.to_string(), epoch_minute);
222            Ok(true)
223        })
224    }
225
226    pub fn modify_runtime<T>(
227        &self,
228        update: impl FnOnce(&mut RuntimeState) -> Result<T, RoutineError>,
229    ) -> Result<T, RoutineError> {
230        self.with_runtime_lock(|| {
231            let mut state = self.load_runtime()?;
232            let result = update(&mut state)?;
233            self.save_runtime(&state)?;
234            Ok(result)
235        })
236    }
237
238    pub(crate) fn with_runtime_lock<T>(
239        &self,
240        transaction: impl FnOnce() -> Result<T, RoutineError>,
241    ) -> Result<T, RoutineError> {
242        let lock = RUNTIME_WRITE_LOCK.get_or_init(|| Mutex::new(()));
243        let _guard = lock
244            .lock()
245            .map_err(|_| RoutineError::Io("runtime lock poisoned".into()))?;
246        transaction()
247    }
248}
249
250#[derive(Deserialize)]
251#[serde(untagged)]
252enum StoredClaims {
253    Current(BTreeMap<String, i64>),
254    Legacy(BTreeSet<String>),
255}
256
257fn deserialize_claims<'de, D>(deserializer: D) -> Result<BTreeMap<String, i64>, D::Error>
258where
259    D: serde::Deserializer<'de>,
260{
261    let stored = StoredClaims::deserialize(deserializer)?;
262    Ok(match stored {
263        StoredClaims::Current(claims) => claims,
264        StoredClaims::Legacy(claims) => {
265            let mut latest = BTreeMap::<String, i64>::new();
266            for claim in claims {
267                let Some((routine, minute)) = claim.rsplit_once("\\0") else {
268                    continue;
269                };
270                let Ok(minute) = minute.parse::<i64>() else {
271                    continue;
272                };
273                latest
274                    .entry(routine.to_string())
275                    .and_modify(|current| *current = (*current).max(minute))
276                    .or_insert(minute);
277            }
278            latest
279        }
280    })
281}
282
283pub fn canonical_working_dir(path: &Path) -> Result<PathBuf, RoutineError> {
284    let canonical = fs::canonicalize(path)?;
285    if !canonical.is_dir() {
286        return Err(RoutineError::Validation(format!(
287            "{} is not a directory",
288            path.display()
289        )));
290    }
291    Ok(canonical)
292}
293
294/// Stable FNV-1a-128 identity. This is persistence format, not `Hash` state.
295pub fn project_key(path: &Path) -> String {
296    // ^ Persistence contract: FNV-1a-128("/repo") = e3905a3dac83d94f708074314a8c762a.
297    const OFFSET: u128 = 0x6c62272e07bb014262b821756295c58d;
298    const PRIME: u128 = 0x0000000001000000000000000000013b;
299    let mut hash = OFFSET;
300    for byte in path.as_os_str().to_string_lossy().as_bytes() {
301        hash ^= *byte as u128;
302        hash = hash.wrapping_mul(PRIME);
303    }
304    format!("{hash:032x}")
305}
306
307fn validate_unique(routines: &[Routine]) -> Result<(), RoutineError> {
308    let mut names = BTreeSet::new();
309    for routine in routines {
310        let validated = routine.clone().validated()?;
311        if validated.name != routine.name {
312            return Err(RoutineError::Validation(
313                "stored routine name must be trimmed".into(),
314            ));
315        }
316        if !names.insert(validated.name.clone()) {
317            return Err(RoutineError::Duplicate(validated.name));
318        }
319    }
320    Ok(())
321}
322
323pub(crate) fn atomic_toml<T: Serialize>(path: &Path, value: &T) -> Result<(), RoutineError> {
324    let parent = path
325        .parent()
326        .ok_or_else(|| RoutineError::Io("path has no parent".into()))?;
327    fs::create_dir_all(parent)?;
328    fs::set_permissions(parent, fs::Permissions::from_mode(0o700))?;
329    let text = toml::to_string_pretty(value).map_err(|e| RoutineError::Corrupt(e.to_string()))?;
330    if text.len() as u64 > MAX_STATE_FILE_BYTES {
331        return Err(RoutineError::Validation(format!(
332            "serialized state exceeds the {MAX_STATE_FILE_BYTES} byte limit"
333        )));
334    }
335    let tmp = path.with_extension(format!("tmp-{}", std::process::id()));
336    let result = (|| {
337        let mut file = OpenOptions::new().write(true).create_new(true).open(&tmp)?;
338        file.set_permissions(fs::Permissions::from_mode(0o600))?;
339        file.write_all(text.as_bytes())?;
340        file.sync_all()?;
341        drop(file);
342        fs::rename(&tmp, path)?;
343        File::open(parent)?.sync_all()?;
344        Ok::<_, RoutineError>(())
345    })();
346    if result.is_err() {
347        let _ = fs::remove_file(&tmp);
348    }
349    result
350}
351
352pub(crate) fn atomic_create(path: &Path, bytes: &[u8]) -> Result<(), std::io::Error> {
353    let parent = path
354        .parent()
355        .ok_or_else(|| std::io::Error::other("path has no parent"))?;
356    fs::create_dir_all(parent)?;
357    fs::set_permissions(parent, fs::Permissions::from_mode(0o700))?;
358    let tmp = path.with_extension(format!("tmp-new-{}", std::process::id()));
359    let result = (|| {
360        let mut file = OpenOptions::new().write(true).create_new(true).open(&tmp)?;
361        file.set_permissions(fs::Permissions::from_mode(0o600))?;
362        file.write_all(bytes)?;
363        file.sync_all()?;
364        drop(file);
365        fs::hard_link(&tmp, path)?;
366        fs::remove_file(&tmp)?;
367        File::open(parent)?.sync_all()?;
368        Ok(())
369    })();
370    if result.is_err() {
371        let _ = fs::remove_file(&tmp);
372    }
373    result
374}
375
376pub(crate) fn read_text_limited(path: &Path) -> Result<String, std::io::Error> {
377    let mut bytes = Vec::new();
378    File::open(path)?
379        .take(MAX_STATE_FILE_BYTES + 1)
380        .read_to_end(&mut bytes)?;
381    if bytes.len() as u64 > MAX_STATE_FILE_BYTES {
382        return Err(std::io::Error::new(
383            std::io::ErrorKind::InvalidData,
384            format!(
385                "{} exceeds the {} byte state-file limit",
386                path.display(),
387                MAX_STATE_FILE_BYTES
388            ),
389        ));
390    }
391    String::from_utf8(bytes)
392        .map_err(|error| std::io::Error::new(std::io::ErrorKind::InvalidData, error))
393}
394
395#[cfg(test)]
396mod tests {
397    use super::*;
398    use crate::routine::{Trigger, SCHEMA_VERSION};
399    use std::sync::atomic::{AtomicU64, Ordering};
400
401    static NEXT: AtomicU64 = AtomicU64::new(0);
402
403    fn test_store() -> RoutineStore {
404        let root = PathBuf::from(env!("CARGO_MANIFEST_DIR"))
405            .join("../../target/routine-tests")
406            .join(format!(
407                "{}-{}",
408                std::process::id(),
409                NEXT.fetch_add(1, Ordering::Relaxed)
410            ));
411        let project =
412            fs::canonicalize(PathBuf::from(env!("CARGO_MANIFEST_DIR")).join("../..")).unwrap();
413        let key = project_key(&project);
414        RoutineStore { root, project, key }
415    }
416
417    fn routine(name: &str) -> Routine {
418        Routine {
419            name: name.into(),
420            trigger: Trigger::Cron("*/5 * * * *".into()),
421            command: vec!["echo".into()],
422            prompt: "hello".into(),
423            enabled: true,
424        }
425    }
426
427    #[test]
428    fn fnv_key_is_stable_known_vector() {
429        assert_eq!(
430            project_key(Path::new("/repo")),
431            "e3905a3dac83d94f708074314a8c762a"
432        );
433        assert_eq!(
434            project_key(Path::new("/repo")),
435            project_key(Path::new("/repo"))
436        );
437    }
438
439    #[test]
440    fn legacy_project_routines_without_enabled_load_as_enabled() {
441        #[derive(Serialize)]
442        struct LegacyRoutine<'a> {
443            name: &'a str,
444            cron: &'a str,
445            command: Vec<&'a str>,
446            prompt: &'a str,
447        }
448
449        #[derive(Serialize)]
450        struct LegacyProjectRoutines<'a> {
451            version: u32,
452            revision: u64,
453            project_path: &'a Path,
454            routines: Vec<LegacyRoutine<'a>>,
455        }
456
457        let store = test_store();
458        let legacy = LegacyProjectRoutines {
459            version: 1,
460            revision: 7,
461            project_path: store.project(),
462            routines: vec![LegacyRoutine {
463                name: "daily",
464                cron: "0 9 * * *",
465                command: vec!["echo"],
466                prompt: "hello",
467            }],
468        };
469        fs::create_dir_all(store.project_file().parent().unwrap()).unwrap();
470        let legacy_text = toml::to_string(&legacy).unwrap();
471        fs::write(store.project_file(), &legacy_text).unwrap();
472
473        let loaded = store.load().unwrap();
474        assert_eq!(loaded.revision, 7);
475        assert_eq!(loaded.routines.len(), 1);
476        assert!(loaded.routines[0].enabled);
477        assert_eq!(
478            fs::read_to_string(store.project_file()).unwrap(),
479            legacy_text
480        );
481        let _ = fs::remove_dir_all(&store.root);
482    }
483
484    #[test]
485    fn durable_store_rejects_stale_revision_and_path_collision() {
486        let store = test_store();
487        let first = store
488            .save(
489                ProjectRoutines {
490                    version: 1,
491                    revision: 0,
492                    project_path: store.project.clone(),
493                    routines: vec![routine("one")],
494                },
495                0,
496            )
497            .unwrap();
498        assert_eq!(first.revision, 1);
499        let stale = store.save(first.clone(), 0).unwrap_err();
500        assert!(matches!(
501            stale,
502            RoutineError::Conflict {
503                expected: 0,
504                actual: 1
505            }
506        ));
507
508        let mut wrong = first;
509        wrong.project_path = PathBuf::from("/another/project");
510        atomic_toml(&store.project_file(), &wrong).unwrap();
511        assert!(matches!(
512            store.load(),
513            Err(RoutineError::ProjectCollision { .. })
514        ));
515        let _ = fs::remove_dir_all(&store.root);
516    }
517
518    #[test]
519    fn durable_store_rejects_duplicate_exact_names() {
520        let store = test_store();
521        let result = store.save(
522            ProjectRoutines {
523                version: 1,
524                revision: 0,
525                project_path: store.project.clone(),
526                routines: vec![routine("same"), routine("same")],
527            },
528            0,
529        );
530        assert!(matches!(result, Err(RoutineError::Duplicate(name)) if name == "same"));
531        let _ = fs::remove_dir_all(&store.root);
532    }
533
534    #[test]
535    fn durable_store_rejects_noncanonical_names() {
536        let store = test_store();
537        let config = ProjectRoutines {
538            version: SCHEMA_VERSION,
539            revision: 1,
540            project_path: store.project.clone(),
541            routines: vec![routine(" padded ")],
542        };
543        atomic_toml(&store.project_file(), &config).unwrap();
544
545        assert!(matches!(
546            store.load(),
547            Err(RoutineError::Validation(message)) if message.contains("trimmed")
548        ));
549        let _ = fs::remove_dir_all(&store.root);
550    }
551
552    #[test]
553    fn claims_are_bounded_and_reject_duplicate_or_rolled_back_minutes() {
554        let store = test_store();
555        for minute in 0..100 {
556            assert!(store.claim("frequent", minute).unwrap());
557            assert!(!store.claim("frequent", minute).unwrap());
558        }
559        assert!(!store.claim("frequent", 50).unwrap());
560        assert!(store.claim("frequent", 100).unwrap());
561        assert!(store.claim("other", 42).unwrap());
562        let state = store.load_runtime().unwrap();
563        assert_eq!(state.claims.len(), 2);
564        assert_eq!(state.claims["frequent"], 100);
565        assert_eq!(state.claims["other"], 42);
566        let _ = fs::remove_dir_all(&store.root);
567    }
568
569    #[test]
570    fn legacy_runtime_load_derives_cron_cause_without_rewriting_file() {
571        let store = test_store();
572        let legacy = r#"version = 1
573claims = {}
574
575[runs]
576
577[[runs.daily]]
578id = "1-1"
579routine = "daily"
580started_epoch = 1
581scheduled_epoch_minute = 123
582status = "succeeded"
583exit_code = 0
584final_output = "done"
585stdout_path = "/logs/stdout"
586stderr_path = "/logs/stderr"
587"#;
588        fs::create_dir_all(store.runtime_file().parent().unwrap()).unwrap();
589        fs::write(store.runtime_file(), legacy).unwrap();
590
591        let loaded = store.load_runtime().unwrap();
592
593        assert_eq!(
594            loaded.runs["daily"][0].cause,
595            RunCause::Cron {
596                scheduled_epoch_minute: 123
597            }
598        );
599        assert_eq!(fs::read_to_string(store.runtime_file()).unwrap(), legacy);
600        let _ = fs::remove_dir_all(&store.root);
601    }
602
603    #[test]
604    fn event_receipts_retain_only_the_latest_bounded_window() {
605        let mut state = RuntimeState::default();
606        for index in 0..=MAX_EVENT_RECEIPTS {
607            state.record_event_receipt("test.changed".into(), index.to_string());
608        }
609
610        assert_eq!(state.event_receipts.len(), MAX_EVENT_RECEIPTS);
611        assert!(!state.has_event_receipt("test.changed", "0"));
612        assert!(state.has_event_receipt("test.changed", &MAX_EVENT_RECEIPTS.to_string()));
613    }
614
615    #[test]
616    fn legacy_claim_sets_load_as_latest_epoch_per_routine() {
617        let parsed: RuntimeState = toml::from_str(
618            r#"
619version = 1
620claims = ["one\\01", "one\\02", "two\\07"]
621"#,
622        )
623        .unwrap();
624        assert_eq!(parsed.claims.get("one"), Some(&2));
625        assert_eq!(parsed.claims.get("two"), Some(&7));
626    }
627}
628
629#[cfg(test)]
630#[path = "store_contract_tests.rs"]
631mod store_contract_tests;
asched_core/routine/store.rs

asched_core/routine/
store.rs
