systemg 0.56.3

An agent-friendly general process composer.
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
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132

#[path = "common/mod.rs"]
mod common;

use std::{
    fs, thread,
    time::{Duration, Instant},
};

use common::HomeEnvGuard;
use systemg::{
    config::load_config,
    cron::{CronExecutionStatus, CronStateFile},
    ipc::{self, ControlCommand, ControlError},
    state_store::StateStore,
};
use tempfile::tempdir;

/// Loads the cron state file from whichever project directory holds `key`.
fn cron_state_with_key(key: &str) -> Option<CronStateFile> {
    let projects_dir = systemg::runtime::state_dir().join("projects");
    let entries = std::fs::read_dir(&projects_dir).ok()?;
    let mut fallback = None;
    for entry in entries.flatten() {
        let id = entry.file_name();
        let store = StateStore::for_project(&id.to_string_lossy());
        if let Ok(state) = CronStateFile::load(store) {
            if state.jobs().contains_key(key) {
                return Some(state);
            }
            fallback = Some(state);
        }
    }
    fallback
}

#[test]
/// Verifies that the supervisor persists completed cron execution history.
fn supervisor_persists_cron_execution_history() {
    use systemg::supervisor::Supervisor;

    let temp = tempdir().expect("failed to create tempdir");
    let dir = temp.path();
    let home = dir.join("home");
    fs::create_dir_all(&home).expect("failed to create home dir");
    let _home = HomeEnvGuard::set(&home);

    let marker = dir.join("cron-history-marker.txt");
    let config_path = dir.join("systemg.yaml");
    fs::write(
        &config_path,
        format!(
            r#"version: "2"
services:
  history_cron:
    command: "sh -c 'date +%s >> \"{marker}\"'"
    restart_policy: "never"
    cron:
      expression: "*/1 * * * * *"
"#,
            marker = marker.display()
        ),
    )
    .expect("failed to write config");

    let config =
        load_config(Some(config_path.to_string_lossy().as_ref())).expect("load config");
    let service_key = config.state_key("history_cron");

    let config_for_thread = config_path.clone();
    let supervisor_thread = thread::spawn(move || {
        let mut supervisor =
            Supervisor::new(config_for_thread, false, None).expect("create supervisor");
        supervisor.run().expect("run supervisor");
    });

    wait_for_supervisor_socket();
    wait_for_cron_history_record(&service_key);

    let content = fs::read_to_string(&marker).expect("read marker");
    assert!(
        !content.trim().is_empty(),
        "cron command should have written the marker"
    );

    let _ = ipc::send_command(&ControlCommand::Shutdown);
    supervisor_thread
        .join()
        .expect("supervisor thread should shut down cleanly");
}

/// Waits until the integration supervisor accepts control commands.
fn wait_for_supervisor_socket() {
    let deadline = Instant::now() + Duration::from_secs(5);
    loop {
        match ipc::send_command(&ControlCommand::Status { live: false }) {
            Ok(_) => return,
            Err(ControlError::NotAvailable) | Err(ControlError::Io(_)) => {}
            Err(err) => panic!("unexpected supervisor status error: {err}"),
        }

        if Instant::now() >= deadline {
            panic!("timed out waiting for supervisor socket");
        }

        thread::sleep(Duration::from_millis(100));
    }
}

/// Waits for a successful persisted cron record under `service_key`.
fn wait_for_cron_history_record(service_key: &str) {
    let deadline = Instant::now() + Duration::from_secs(10);
    loop {
        if let Some(state) = cron_state_with_key(service_key)
            && let Some(job) = state.jobs().get(service_key)
            && let Some(record) = job.execution_history.back()
            && record.completed_at.is_some()
            && matches!(record.status, Some(CronExecutionStatus::Success))
            && record.exit_code == Some(0)
        {
            return;
        }

        if Instant::now() >= deadline {
            let state = cron_state_with_key(service_key);
            panic!(
                "timed out waiting for persisted cron execution history for {service_key}; state={state:#?}"
            );
        }

        thread::sleep(Duration::from_millis(100));
    }
}