a3s_box_runtime/local_execution/
lifecycle_lock.rs

1use std::path::Path;
2
3use a3s_box_core::{ExecutionManagerError, ExecutionManagerResult};
4
5/// Cross-process lock shared with the CLI's stopped-commit/start lock.
6pub struct ExecutionLifecycleLock {
7    #[cfg(any(unix, windows))]
8    _file: std::fs::File,
9}
10
11/// Acquire the cross-process lifecycle lock from synchronous callers such as
12/// the direct SDK. The returned guard holds the lock until it is dropped.
13pub fn acquire_blocking(
14    home_dir: &Path,
15    execution_id: &str,
16) -> ExecutionManagerResult<ExecutionLifecycleLock> {
17    validate_execution_id(execution_id)?;
18    ExecutionLifecycleLock::acquire(&home_dir.join("locks"), execution_id).map_err(|error| {
19        ExecutionManagerError::Internal(format!("failed to acquire lifecycle lock: {error}"))
20    })
21}
22
23fn validate_execution_id(execution_id: &str) -> ExecutionManagerResult<()> {
24    if execution_id.is_empty()
25        || !execution_id
26            .bytes()
27            .all(|byte| byte.is_ascii_alphanumeric() || matches!(byte, b'-' | b'_'))
28    {
29        return Err(ExecutionManagerError::Internal(
30            "execution ID is not safe for a lifecycle lock filename".to_string(),
31        ));
32    }
33    Ok(())
34}
35
36impl ExecutionLifecycleLock {
37    #[cfg(unix)]
38    fn acquire(directory: &Path, execution_id: &str) -> std::io::Result<Self> {
39        use std::os::unix::io::AsRawFd;
40
41        std::fs::create_dir_all(directory)?;
42        let file = std::fs::OpenOptions::new()
43            .read(true)
44            .write(true)
45            .create(true)
46            .truncate(false)
47            .open(directory.join(format!("{execution_id}.lifecycle.lock")))?;
48        if unsafe { libc::flock(file.as_raw_fd(), libc::LOCK_EX) } != 0 {
49            return Err(std::io::Error::last_os_error());
50        }
51        Ok(Self { _file: file })
52    }
53
54    #[cfg(windows)]
55    fn acquire(directory: &Path, execution_id: &str) -> std::io::Result<Self> {
56        use std::os::windows::fs::OpenOptionsExt;
57        use std::time::Duration;
58
59        const ERROR_SHARING_VIOLATION: i32 = 32;
60        const ERROR_LOCK_VIOLATION: i32 = 33;
61
62        std::fs::create_dir_all(directory)?;
63        let path = directory.join(format!("{execution_id}.lifecycle.lock"));
64        loop {
65            match std::fs::OpenOptions::new()
66                .read(true)
67                .write(true)
68                .create(true)
69                .truncate(false)
70                .share_mode(0)
71                .open(&path)
72            {
73                Ok(file) => return Ok(Self { _file: file }),
74                Err(error)
75                    if matches!(
76                        error.raw_os_error(),
77                        Some(ERROR_SHARING_VIOLATION | ERROR_LOCK_VIOLATION)
78                    ) =>
79                {
80                    std::thread::sleep(Duration::from_millis(10));
81                }
82                Err(error) => return Err(error),
83            }
84        }
85    }
86
87    #[cfg(not(any(unix, windows)))]
88    fn acquire(_directory: &Path, _execution_id: &str) -> std::io::Result<Self> {
89        Ok(Self {})
90    }
91}
92
93pub(super) async fn acquire(
94    home_dir: &Path,
95    execution_id: &str,
96) -> ExecutionManagerResult<ExecutionLifecycleLock> {
97    validate_execution_id(execution_id)?;
98    let directory = home_dir.join("locks");
99    let execution_id = execution_id.to_string();
100    tokio::task::spawn_blocking(move || ExecutionLifecycleLock::acquire(&directory, &execution_id))
101        .await
102        .map_err(|error| {
103            ExecutionManagerError::Internal(format!("lifecycle lock task failed: {error}"))
104        })?
105        .map_err(|error| {
106            ExecutionManagerError::Internal(format!("failed to acquire lifecycle lock: {error}"))
107        })
108}
109
110#[cfg(test)]
111mod tests {
112    use super::*;
113
114    #[cfg(any(unix, windows))]
115    #[test]
116    fn same_execution_is_serialized_across_file_handles() {
117        use std::sync::mpsc;
118        use std::time::Duration;
119
120        let directory = tempfile::tempdir().unwrap();
121        let first = ExecutionLifecycleLock::acquire(directory.path(), "execution-1").unwrap();
122        let lock_dir = directory.path().to_path_buf();
123        let (started_tx, started_rx) = mpsc::channel();
124        let (acquired_tx, acquired_rx) = mpsc::channel();
125        let waiter = std::thread::spawn(move || {
126            started_tx.send(()).unwrap();
127            let _second = ExecutionLifecycleLock::acquire(&lock_dir, "execution-1").unwrap();
128            acquired_tx.send(()).unwrap();
129        });
130
131        started_rx.recv().unwrap();
132        assert!(acquired_rx.recv_timeout(Duration::from_millis(50)).is_err());
133        drop(first);
134        acquired_rx.recv_timeout(Duration::from_secs(2)).unwrap();
135        waiter.join().unwrap();
136    }
137}
a3s_box_runtime/local_execution/lifecycle_lock.rs

a3s_box_runtime/local_execution/
lifecycle_lock.rs
