runner_manager_platform/wsl/
fence.rs

1//! Cross-boundary recovery fence shared by Windows and a managed WSL guest.
2//!
3//! Windows share locks, Unix `flock`, and SQLite byte-range locks do not
4//! interoperate reliably through DrvFS. Directory creation does: exactly one
5//! side can create the same directory. The directory is intentionally durable
6//! on process death; an abandoned owner blocks recovery instead of allowing a
7//! possibly concurrent runner launch.
8
9use std::fs;
10use std::io::Write as _;
11use std::path::{Path, PathBuf};
12
13use chrono::{DateTime, Utc};
14use serde::{Deserialize, Serialize};
15
16use super::discovery::{escaped_name_with_digest, validate_distribution_name};
17use crate::paths::AppPaths;
18use runner_manager_domain::model::ScaleTarget;
19
20pub const GUEST_CONFIG_FILE: &str = "wsl-recovery.toml";
21pub const REQUEST_FILE: &str = "drain-request.json";
22pub const HEARTBEAT_FILE: &str = "guest-heartbeat.json";
23pub const FENCE_DIRECTORY: &str = "launch-fence";
24pub const OWNER_FILE: &str = "owner.json";
25pub const RECOVERY_STATUS_FILE: &str = "recovery-status.json";
26pub const SCHEMA_VERSION: u32 = 1;
27
28/// Windows-side directory shared with one exact distribution.
29pub fn recovery_root(paths: &AppPaths, distribution: &str) -> Result<PathBuf, super::WslError> {
30    validate_distribution_name(distribution)?;
31    Ok(paths
32        .config_dir()
33        .join("wsl-recovery")
34        .join(escaped_name_with_digest(distribution)))
35}
36
37#[derive(Debug, thiserror::Error)]
38pub enum FenceError {
39    #[error("cannot {operation} WSL recovery state at {}: {source}", path.display())]
40    Io {
41        operation: &'static str,
42        path: PathBuf,
43        #[source]
44        source: std::io::Error,
45    },
46    #[error("cannot decode WSL recovery state at {}: {source}", path.display())]
47    Decode {
48        path: PathBuf,
49        #[source]
50        source: serde_json::Error,
51    },
52    #[error("WSL recovery state at {} has schema {found}, but this build supports {SCHEMA_VERSION}", path.display())]
53    Schema { path: PathBuf, found: u32 },
54}
55
56fn io(operation: &'static str, path: &Path, source: std::io::Error) -> FenceError {
57    FenceError::Io {
58        operation,
59        path: path.to_path_buf(),
60        source,
61    }
62}
63
64#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
65#[serde(deny_unknown_fields)]
66pub struct GuestRecoveryConfig {
67    pub schema_version: u32,
68    pub shared_root: PathBuf,
69}
70
71impl GuestRecoveryConfig {
72    #[must_use]
73    pub fn new(shared_root: PathBuf) -> Self {
74        Self {
75            schema_version: SCHEMA_VERSION,
76            shared_root,
77        }
78    }
79
80    #[must_use]
81    pub fn path(paths: &AppPaths) -> PathBuf {
82        paths.config_dir().join(GUEST_CONFIG_FILE)
83    }
84
85    pub fn read(paths: &AppPaths) -> Result<Option<Self>, FenceError> {
86        let path = Self::path(paths);
87        let text = match fs::read_to_string(&path) {
88            Ok(text) => text,
89            Err(source) if source.kind() == std::io::ErrorKind::NotFound => return Ok(None),
90            Err(source) => return Err(io("read", &path, source)),
91        };
92        let value: Self = toml::from_str(&text).map_err(|source| FenceError::Io {
93            operation: "decode",
94            path: path.clone(),
95            source: std::io::Error::new(std::io::ErrorKind::InvalidData, source),
96        })?;
97        if value.schema_version != SCHEMA_VERSION {
98            return Err(FenceError::Schema {
99                path,
100                found: value.schema_version,
101            });
102        }
103        Ok(Some(value))
104    }
105
106    pub fn write(&self, paths: &AppPaths) -> Result<(), FenceError> {
107        let path = Self::path(paths);
108        let text = toml::to_string_pretty(self).map_err(|source| FenceError::Io {
109            operation: "encode",
110            path: path.clone(),
111            source: std::io::Error::new(std::io::ErrorKind::InvalidData, source),
112        })?;
113        atomic_write(&path, text.as_bytes())
114    }
115}
116
117#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
118#[serde(deny_unknown_fields)]
119pub struct DrainRequest {
120    pub schema_version: u32,
121    pub generation: u64,
122    pub requested_at: DateTime<Utc>,
123}
124
125impl DrainRequest {
126    #[must_use]
127    pub fn new(generation: u64, requested_at: DateTime<Utc>) -> Self {
128        Self {
129            schema_version: SCHEMA_VERSION,
130            generation,
131            requested_at,
132        }
133    }
134
135    pub fn write(&self, root: &Path) -> Result<(), FenceError> {
136        write_json(&root.join(REQUEST_FILE), self)
137    }
138
139    pub fn read(root: &Path) -> Result<Option<Self>, FenceError> {
140        read_json(&root.join(REQUEST_FILE))
141    }
142}
143
144#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
145#[serde(deny_unknown_fields)]
146pub struct GuestHeartbeat {
147    pub schema_version: u32,
148    pub observed_at: DateTime<Utc>,
149    pub acknowledged_generation: Option<u64>,
150    pub local_active_attempts: Option<u32>,
151    pub managed_targets: Vec<ScaleTarget>,
152    pub unmanaged_runner_services: Option<u32>,
153}
154
155#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
156#[serde(rename_all = "snake_case")]
157pub enum RecoveryPhase {
158    Healthy,
159    Degraded,
160    Draining,
161    Recovering,
162    Backoff,
163    RecoveryBlocked,
164}
165
166#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
167#[serde(deny_unknown_fields)]
168pub struct RecoveryStatus {
169    pub schema_version: u32,
170    pub observed_at: DateTime<Utc>,
171    pub phase: RecoveryPhase,
172    pub consecutive_probe_failures: u8,
173    pub reason: Option<String>,
174    pub last_recovered_at: Option<DateTime<Utc>>,
175}
176
177impl RecoveryStatus {
178    pub fn write(&self, root: &Path) -> Result<(), FenceError> {
179        write_json(&root.join(RECOVERY_STATUS_FILE), self)
180    }
181
182    pub fn read(root: &Path) -> Result<Option<Self>, FenceError> {
183        read_json(&root.join(RECOVERY_STATUS_FILE))
184    }
185}
186
187impl GuestHeartbeat {
188    pub fn write(&self, root: &Path) -> Result<(), FenceError> {
189        write_json(&root.join(HEARTBEAT_FILE), self)
190    }
191
192    pub fn read(root: &Path) -> Result<Option<Self>, FenceError> {
193        read_json(&root.join(HEARTBEAT_FILE))
194    }
195}
196
197#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
198#[serde(rename_all = "snake_case")]
199pub enum FenceOwnerKind {
200    GuestLaunch,
201    WindowsRecovery,
202}
203
204#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
205#[serde(deny_unknown_fields)]
206pub struct FenceOwner {
207    pub schema_version: u32,
208    pub kind: FenceOwnerKind,
209    pub generation: Option<u64>,
210    pub process_id: u32,
211    pub acquired_at: DateTime<Utc>,
212}
213
214/// A successfully created cross-boundary directory claim.
215#[derive(Debug)]
216pub struct FenceClaim {
217    directory: PathBuf,
218    release_on_drop: bool,
219}
220
221impl FenceClaim {
222    /// Attempt to claim the launch boundary. `Ok(None)` means another side
223    /// owns it; absence or malformed owner metadata never makes it free.
224    pub fn try_claim(
225        root: &Path,
226        kind: FenceOwnerKind,
227        generation: Option<u64>,
228    ) -> Result<Option<Self>, FenceError> {
229        fs::create_dir_all(root).map_err(|source| io("create", root, source))?;
230        let directory = root.join(FENCE_DIRECTORY);
231        match fs::create_dir(&directory) {
232            Ok(()) => {}
233            Err(source) if source.kind() == std::io::ErrorKind::AlreadyExists => return Ok(None),
234            Err(source) => return Err(io("claim", &directory, source)),
235        }
236        let owner = FenceOwner {
237            schema_version: SCHEMA_VERSION,
238            kind,
239            generation,
240            process_id: std::process::id(),
241            acquired_at: Utc::now(),
242        };
243        if let Err(error) = write_json(&directory.join(OWNER_FILE), &owner) {
244            let _ = fs::remove_dir_all(&directory);
245            return Err(error);
246        }
247        Ok(Some(Self {
248            directory,
249            release_on_drop: true,
250        }))
251    }
252
253    /// Leave a recovery claim durable. A restarted watchdog may adopt only a
254    /// matching recovery owner; a guest-launch owner is never removed blindly.
255    pub fn make_durable(mut self) {
256        self.release_on_drop = false;
257    }
258
259    pub fn owner(root: &Path) -> Result<Option<FenceOwner>, FenceError> {
260        read_json(&root.join(FENCE_DIRECTORY).join(OWNER_FILE))
261    }
262
263    pub fn release(mut self) -> Result<(), FenceError> {
264        self.release_on_drop = false;
265        remove_claim(&self.directory)
266    }
267}
268
269impl Drop for FenceClaim {
270    fn drop(&mut self) {
271        if self.release_on_drop {
272            let _ = remove_claim(&self.directory);
273        }
274    }
275}
276
277pub fn clear_recovery(root: &Path, generation: u64) -> Result<(), FenceError> {
278    let owner = FenceClaim::owner(root)?;
279    if owner.as_ref().is_some_and(|owner| {
280        owner.kind == FenceOwnerKind::WindowsRecovery && owner.generation == Some(generation)
281    }) {
282        remove_claim(&root.join(FENCE_DIRECTORY))?;
283    }
284    let request_path = root.join(REQUEST_FILE);
285    if DrainRequest::read(root)?
286        .as_ref()
287        .is_some_and(|request| request.generation != generation)
288    {
289        return Ok(());
290    }
291    match fs::remove_file(&request_path) {
292        Ok(()) => Ok(()),
293        Err(source) if source.kind() == std::io::ErrorKind::NotFound => Ok(()),
294        Err(source) => Err(io("remove", &request_path, source)),
295    }
296}
297
298fn remove_claim(directory: &Path) -> Result<(), FenceError> {
299    match fs::remove_dir_all(directory) {
300        Ok(()) => Ok(()),
301        Err(source) if source.kind() == std::io::ErrorKind::NotFound => Ok(()),
302        Err(source) => Err(io("release", directory, source)),
303    }
304}
305
306fn atomic_write(path: &Path, bytes: &[u8]) -> Result<(), FenceError> {
307    let parent = path.parent().unwrap_or_else(|| Path::new("."));
308    fs::create_dir_all(parent).map_err(|source| io("create", parent, source))?;
309    let mut temporary =
310        tempfile::NamedTempFile::new_in(parent).map_err(|source| io("write", path, source))?;
311    temporary
312        .write_all(bytes)
313        .and_then(|()| temporary.as_file().sync_all())
314        .map_err(|source| io("write", path, source))?;
315    temporary
316        .persist(path)
317        .map(|_| ())
318        .map_err(|error| io("replace", path, error.error))
319}
320
321fn write_json<T: Serialize>(path: &Path, value: &T) -> Result<(), FenceError> {
322    let bytes = serde_json::to_vec_pretty(value).map_err(|source| FenceError::Decode {
323        path: path.to_path_buf(),
324        source,
325    })?;
326    atomic_write(path, &bytes)
327}
328
329fn read_json<T: for<'de> Deserialize<'de>>(path: &Path) -> Result<Option<T>, FenceError> {
330    let bytes = match fs::read(path) {
331        Ok(bytes) => bytes,
332        Err(source) if source.kind() == std::io::ErrorKind::NotFound => return Ok(None),
333        Err(source) => return Err(io("read", path, source)),
334    };
335    let value: T = serde_json::from_slice(&bytes).map_err(|source| FenceError::Decode {
336        path: path.to_path_buf(),
337        source,
338    })?;
339    Ok(Some(value))
340}
341
342/// Persistent runner services are outside runner-manager's attempt journal.
343/// Their presence blocks automated WSL termination even while they look idle.
344#[must_use]
345pub fn unmanaged_runner_service_count() -> Option<u32> {
346    if !cfg!(target_os = "linux") {
347        return Some(0);
348    }
349    let mut names = std::collections::BTreeSet::new();
350    for directory in [
351        "/etc/systemd/system",
352        "/usr/lib/systemd/system",
353        "/lib/systemd/system",
354    ] {
355        let entries = match fs::read_dir(directory) {
356            Ok(entries) => entries,
357            Err(error) if error.kind() == std::io::ErrorKind::NotFound => continue,
358            Err(_) => return None,
359        };
360        for entry in entries.flatten() {
361            let name = entry.file_name().to_string_lossy().into_owned();
362            if name.starts_with("actions.runner.") && name.ends_with(".service") {
363                names.insert(name);
364            }
365        }
366    }
367    u32::try_from(names.len()).ok()
368}
369
370#[cfg(test)]
371mod tests {
372    use super::*;
373
374    #[test]
375    fn one_directory_has_exactly_one_owner_and_drop_releases_guest_claim() {
376        let root = tempfile::tempdir().unwrap();
377        let first = FenceClaim::try_claim(root.path(), FenceOwnerKind::GuestLaunch, None)
378            .unwrap()
379            .unwrap();
380        assert!(
381            FenceClaim::try_claim(root.path(), FenceOwnerKind::WindowsRecovery, Some(1))
382                .unwrap()
383                .is_none()
384        );
385        drop(first);
386        assert!(
387            FenceClaim::try_claim(root.path(), FenceOwnerKind::WindowsRecovery, Some(1))
388                .unwrap()
389                .is_some()
390        );
391    }
392
393    #[test]
394    fn only_matching_recovery_generation_is_cleared() {
395        let root = tempfile::tempdir().unwrap();
396        let claim = FenceClaim::try_claim(root.path(), FenceOwnerKind::WindowsRecovery, Some(9))
397            .unwrap()
398            .unwrap();
399        claim.make_durable();
400        DrainRequest::new(9, Utc::now()).write(root.path()).unwrap();
401        clear_recovery(root.path(), 8).unwrap();
402        assert!(root.path().join(FENCE_DIRECTORY).exists());
403        assert_eq!(
404            DrainRequest::read(root.path()).unwrap().unwrap().generation,
405            9
406        );
407        clear_recovery(root.path(), 9).unwrap();
408        assert!(!root.path().join(FENCE_DIRECTORY).exists());
409        assert!(!root.path().join(REQUEST_FILE).exists());
410    }
411
412    #[test]
413    fn heartbeat_and_status_documents_replace_atomically() {
414        let root = tempfile::tempdir().unwrap();
415        let first = GuestHeartbeat {
416            schema_version: SCHEMA_VERSION,
417            observed_at: Utc::now(),
418            acknowledged_generation: None,
419            local_active_attempts: Some(1),
420            managed_targets: Vec::new(),
421            unmanaged_runner_services: Some(0),
422        };
423        let mut second = first.clone();
424        second.local_active_attempts = Some(0);
425        first.write(root.path()).unwrap();
426        second.write(root.path()).unwrap();
427        assert_eq!(GuestHeartbeat::read(root.path()).unwrap(), Some(second));
428
429        let first = RecoveryStatus {
430            schema_version: SCHEMA_VERSION,
431            observed_at: Utc::now(),
432            phase: RecoveryPhase::Degraded,
433            consecutive_probe_failures: 1,
434            reason: None,
435            last_recovered_at: None,
436        };
437        let mut second = first.clone();
438        second.phase = RecoveryPhase::Healthy;
439        first.write(root.path()).unwrap();
440        second.write(root.path()).unwrap();
441        assert_eq!(RecoveryStatus::read(root.path()).unwrap(), Some(second));
442    }
443}
runner_manager_platform/wsl/fence.rs

runner_manager_platform/wsl/
fence.rs
