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    /// Attempts which the guest journal still considers to be executing a
152    /// GitHub job. This lets recovery distinguish a stuck idle/listener
153    /// process from work which must never be interrupted.
154    #[serde(default)]
155    pub local_busy_attempts: Option<u32>,
156    pub managed_targets: Vec<ScaleTarget>,
157    pub unmanaged_runner_services: Option<u32>,
158}
159
160#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
161#[serde(rename_all = "snake_case")]
162pub enum RecoveryPhase {
163    Healthy,
164    Degraded,
165    Draining,
166    Recovering,
167    Backoff,
168    RecoveryBlocked,
169}
170
171#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
172#[serde(deny_unknown_fields)]
173pub struct RecoveryStatus {
174    pub schema_version: u32,
175    pub observed_at: DateTime<Utc>,
176    pub phase: RecoveryPhase,
177    pub consecutive_probe_failures: u8,
178    pub reason: Option<String>,
179    pub last_recovered_at: Option<DateTime<Utc>>,
180}
181
182impl RecoveryStatus {
183    pub fn write(&self, root: &Path) -> Result<(), FenceError> {
184        write_json(&root.join(RECOVERY_STATUS_FILE), self)
185    }
186
187    pub fn read(root: &Path) -> Result<Option<Self>, FenceError> {
188        read_json(&root.join(RECOVERY_STATUS_FILE))
189    }
190}
191
192impl GuestHeartbeat {
193    pub fn write(&self, root: &Path) -> Result<(), FenceError> {
194        write_json(&root.join(HEARTBEAT_FILE), self)
195    }
196
197    pub fn read(root: &Path) -> Result<Option<Self>, FenceError> {
198        read_json(&root.join(HEARTBEAT_FILE))
199    }
200}
201
202#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
203#[serde(rename_all = "snake_case")]
204pub enum FenceOwnerKind {
205    GuestLaunch,
206    WindowsRecovery,
207}
208
209#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
210#[serde(deny_unknown_fields)]
211pub struct FenceOwner {
212    pub schema_version: u32,
213    pub kind: FenceOwnerKind,
214    pub generation: Option<u64>,
215    pub process_id: u32,
216    pub acquired_at: DateTime<Utc>,
217}
218
219/// A successfully created cross-boundary directory claim.
220#[derive(Debug)]
221pub struct FenceClaim {
222    directory: PathBuf,
223    release_on_drop: bool,
224}
225
226impl FenceClaim {
227    /// Attempt to claim the launch boundary. `Ok(None)` means another side
228    /// owns it; absence or malformed owner metadata never makes it free.
229    pub fn try_claim(
230        root: &Path,
231        kind: FenceOwnerKind,
232        generation: Option<u64>,
233    ) -> Result<Option<Self>, FenceError> {
234        fs::create_dir_all(root).map_err(|source| io("create", root, source))?;
235        let directory = root.join(FENCE_DIRECTORY);
236        match fs::create_dir(&directory) {
237            Ok(()) => {}
238            Err(source) if source.kind() == std::io::ErrorKind::AlreadyExists => return Ok(None),
239            Err(source) => return Err(io("claim", &directory, source)),
240        }
241        let owner = FenceOwner {
242            schema_version: SCHEMA_VERSION,
243            kind,
244            generation,
245            process_id: std::process::id(),
246            acquired_at: Utc::now(),
247        };
248        if let Err(error) = write_json(&directory.join(OWNER_FILE), &owner) {
249            let _ = fs::remove_dir_all(&directory);
250            return Err(error);
251        }
252        Ok(Some(Self {
253            directory,
254            release_on_drop: true,
255        }))
256    }
257
258    /// Leave a recovery claim durable. A restarted watchdog may adopt only a
259    /// matching recovery owner; a guest-launch owner is never removed blindly.
260    pub fn make_durable(mut self) {
261        self.release_on_drop = false;
262    }
263
264    pub fn owner(root: &Path) -> Result<Option<FenceOwner>, FenceError> {
265        read_json(&root.join(FENCE_DIRECTORY).join(OWNER_FILE))
266    }
267
268    pub fn release(mut self) -> Result<(), FenceError> {
269        self.release_on_drop = false;
270        remove_claim(&self.directory)
271    }
272}
273
274impl Drop for FenceClaim {
275    fn drop(&mut self) {
276        if self.release_on_drop {
277            let _ = remove_claim(&self.directory);
278        }
279    }
280}
281
282pub fn clear_recovery(root: &Path, generation: u64) -> Result<(), FenceError> {
283    let owner = FenceClaim::owner(root)?;
284    if owner.as_ref().is_some_and(|owner| {
285        owner.kind == FenceOwnerKind::WindowsRecovery && owner.generation == Some(generation)
286    }) {
287        remove_claim(&root.join(FENCE_DIRECTORY))?;
288    }
289    let request_path = root.join(REQUEST_FILE);
290    if DrainRequest::read(root)?
291        .as_ref()
292        .is_some_and(|request| request.generation != generation)
293    {
294        return Ok(());
295    }
296    match fs::remove_file(&request_path) {
297        Ok(()) => Ok(()),
298        Err(source) if source.kind() == std::io::ErrorKind::NotFound => Ok(()),
299        Err(source) => Err(io("remove", &request_path, source)),
300    }
301}
302
303/// Retire coordination written by a Windows recovery watchdog that no longer
304/// has authority to run. A guest launch claim is deliberately preserved. The
305/// fence and request may have different generations after a watchdog restart,
306/// so each Windows-owned generation is cleared independently.
307pub fn retire_windows_recovery(root: &Path) -> Result<(), FenceError> {
308    let recovery_generation = FenceClaim::owner(root)?.and_then(|owner| {
309        (owner.kind == FenceOwnerKind::WindowsRecovery)
310            .then_some(owner.generation)
311            .flatten()
312    });
313    if let Some(generation) = recovery_generation {
314        clear_recovery(root, generation)?;
315    }
316    if let Some(request) = DrainRequest::read(root)? {
317        clear_recovery(root, request.generation)?;
318    }
319    let status_path = root.join(RECOVERY_STATUS_FILE);
320    match fs::remove_file(&status_path) {
321        Ok(()) => Ok(()),
322        Err(source) if source.kind() == std::io::ErrorKind::NotFound => Ok(()),
323        Err(source) => Err(io("remove", &status_path, source)),
324    }
325}
326
327fn remove_claim(directory: &Path) -> Result<(), FenceError> {
328    match fs::remove_dir_all(directory) {
329        Ok(()) => Ok(()),
330        Err(source) if source.kind() == std::io::ErrorKind::NotFound => Ok(()),
331        Err(source) => Err(io("release", directory, source)),
332    }
333}
334
335fn atomic_write(path: &Path, bytes: &[u8]) -> Result<(), FenceError> {
336    let parent = path.parent().unwrap_or_else(|| Path::new("."));
337    fs::create_dir_all(parent).map_err(|source| io("create", parent, source))?;
338    let mut temporary =
339        tempfile::NamedTempFile::new_in(parent).map_err(|source| io("write", path, source))?;
340    temporary
341        .write_all(bytes)
342        .and_then(|()| temporary.as_file().sync_all())
343        .map_err(|source| io("write", path, source))?;
344    temporary
345        .persist(path)
346        .map(|_| ())
347        .map_err(|error| io("replace", path, error.error))
348}
349
350fn write_json<T: Serialize>(path: &Path, value: &T) -> Result<(), FenceError> {
351    let bytes = serde_json::to_vec_pretty(value).map_err(|source| FenceError::Decode {
352        path: path.to_path_buf(),
353        source,
354    })?;
355    atomic_write(path, &bytes)
356}
357
358fn read_json<T: for<'de> Deserialize<'de>>(path: &Path) -> Result<Option<T>, FenceError> {
359    let bytes = match fs::read(path) {
360        Ok(bytes) => bytes,
361        Err(source) if source.kind() == std::io::ErrorKind::NotFound => return Ok(None),
362        Err(source) => return Err(io("read", path, source)),
363    };
364    let value: T = serde_json::from_slice(&bytes).map_err(|source| FenceError::Decode {
365        path: path.to_path_buf(),
366        source,
367    })?;
368    Ok(Some(value))
369}
370
371/// Persistent runner services are outside runner-manager's attempt journal.
372/// Their presence blocks automated WSL termination even while they look idle.
373#[must_use]
374pub fn unmanaged_runner_service_count() -> Option<u32> {
375    if !cfg!(target_os = "linux") {
376        return Some(0);
377    }
378    let mut names = std::collections::BTreeSet::new();
379    for directory in [
380        "/etc/systemd/system",
381        "/usr/lib/systemd/system",
382        "/lib/systemd/system",
383    ] {
384        let entries = match fs::read_dir(directory) {
385            Ok(entries) => entries,
386            Err(error) if error.kind() == std::io::ErrorKind::NotFound => continue,
387            Err(_) => return None,
388        };
389        for entry in entries.flatten() {
390            let name = entry.file_name().to_string_lossy().into_owned();
391            if name.starts_with("actions.runner.") && name.ends_with(".service") {
392                names.insert(name);
393            }
394        }
395    }
396    u32::try_from(names.len()).ok()
397}
398
399#[cfg(test)]
400mod tests {
401    use super::*;
402
403    #[test]
404    fn one_directory_has_exactly_one_owner_and_drop_releases_guest_claim() {
405        let root = tempfile::tempdir().unwrap();
406        let first = FenceClaim::try_claim(root.path(), FenceOwnerKind::GuestLaunch, None)
407            .unwrap()
408            .unwrap();
409        assert!(
410            FenceClaim::try_claim(root.path(), FenceOwnerKind::WindowsRecovery, Some(1))
411                .unwrap()
412                .is_none()
413        );
414        drop(first);
415        assert!(
416            FenceClaim::try_claim(root.path(), FenceOwnerKind::WindowsRecovery, Some(1))
417                .unwrap()
418                .is_some()
419        );
420    }
421
422    #[test]
423    fn only_matching_recovery_generation_is_cleared() {
424        let root = tempfile::tempdir().unwrap();
425        let claim = FenceClaim::try_claim(root.path(), FenceOwnerKind::WindowsRecovery, Some(9))
426            .unwrap()
427            .unwrap();
428        claim.make_durable();
429        DrainRequest::new(9, Utc::now()).write(root.path()).unwrap();
430        clear_recovery(root.path(), 8).unwrap();
431        assert!(root.path().join(FENCE_DIRECTORY).exists());
432        assert_eq!(
433            DrainRequest::read(root.path()).unwrap().unwrap().generation,
434            9
435        );
436        clear_recovery(root.path(), 9).unwrap();
437        assert!(!root.path().join(FENCE_DIRECTORY).exists());
438        assert!(!root.path().join(REQUEST_FILE).exists());
439    }
440
441    #[test]
442    fn retiring_windows_recovery_handles_restarted_generations_but_keeps_guest_claims() {
443        let root = tempfile::tempdir().unwrap();
444        let claim = FenceClaim::try_claim(root.path(), FenceOwnerKind::WindowsRecovery, Some(7))
445            .unwrap()
446            .unwrap();
447        claim.make_durable();
448        DrainRequest::new(8, Utc::now()).write(root.path()).unwrap();
449        RecoveryStatus {
450            schema_version: SCHEMA_VERSION,
451            observed_at: Utc::now(),
452            phase: RecoveryPhase::RecoveryBlocked,
453            consecutive_probe_failures: 9,
454            reason: Some("stale".into()),
455            last_recovered_at: None,
456        }
457        .write(root.path())
458        .unwrap();
459
460        retire_windows_recovery(root.path()).unwrap();
461
462        assert!(!root.path().join(FENCE_DIRECTORY).exists());
463        assert!(!root.path().join(REQUEST_FILE).exists());
464        assert!(!root.path().join(RECOVERY_STATUS_FILE).exists());
465
466        let guest = FenceClaim::try_claim(root.path(), FenceOwnerKind::GuestLaunch, None)
467            .unwrap()
468            .unwrap();
469        guest.make_durable();
470        DrainRequest::new(10, Utc::now())
471            .write(root.path())
472            .unwrap();
473        retire_windows_recovery(root.path()).unwrap();
474        assert!(root.path().join(FENCE_DIRECTORY).exists());
475        assert!(!root.path().join(REQUEST_FILE).exists());
476    }
477
478    #[test]
479    fn heartbeat_and_status_documents_replace_atomically() {
480        let root = tempfile::tempdir().unwrap();
481        let first = GuestHeartbeat {
482            schema_version: SCHEMA_VERSION,
483            observed_at: Utc::now(),
484            acknowledged_generation: None,
485            local_active_attempts: Some(1),
486            local_busy_attempts: Some(1),
487            managed_targets: Vec::new(),
488            unmanaged_runner_services: Some(0),
489        };
490        let mut second = first.clone();
491        second.local_active_attempts = Some(0);
492        first.write(root.path()).unwrap();
493        second.write(root.path()).unwrap();
494        assert_eq!(GuestHeartbeat::read(root.path()).unwrap(), Some(second));
495
496        let first = RecoveryStatus {
497            schema_version: SCHEMA_VERSION,
498            observed_at: Utc::now(),
499            phase: RecoveryPhase::Degraded,
500            consecutive_probe_failures: 1,
501            reason: None,
502            last_recovered_at: None,
503        };
504        let mut second = first.clone();
505        second.phase = RecoveryPhase::Healthy;
506        first.write(root.path()).unwrap();
507        second.write(root.path()).unwrap();
508        assert_eq!(RecoveryStatus::read(root.path()).unwrap(), Some(second));
509    }
510}
runner_manager_platform/wsl/fence.rs

runner_manager_platform/wsl/
fence.rs
