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
327/// Retire Windows recovery coordination that no live watchdog is maintaining.
328///
329/// A watchdog keeps its recovery generation in memory only. When it exits
330/// mid-recovery -- the Windows service or the companion task restarting while
331/// WSL is being recovered -- its durable fence claim and drain request outlive
332/// it, and its successor starts with a new generation. [`clear_recovery`] only
333/// clears a matching generation, so the successor never clears them and every
334/// guest launch stays fenced forever while WSL itself is healthy.
335///
336/// A live watchdog rewrites its drain request on every unhealthy observation,
337/// so a request newer than `abandoned_after` means recovery is still being
338/// driven and nothing is touched. A Windows-owned claim younger than that is
339/// likewise left alone, because its owner may be between claiming the fence
340/// and its next request write. A guest launch claim is never removed.
341///
342/// Returns `true` when abandoned coordination was retired.
343pub fn retire_abandoned_windows_recovery(
344    root: &Path,
345    now: DateTime<Utc>,
346    abandoned_after: std::time::Duration,
347) -> Result<bool, FenceError> {
348    let abandoned = |at: DateTime<Utc>| (now - at).to_std().is_ok_and(|age| age >= abandoned_after);
349    let request = DrainRequest::read(root)?;
350    if request
351        .as_ref()
352        .is_some_and(|request| !abandoned(request.requested_at))
353    {
354        return Ok(false);
355    }
356    let windows_owner =
357        FenceClaim::owner(root)?.filter(|owner| owner.kind == FenceOwnerKind::WindowsRecovery);
358    if windows_owner
359        .as_ref()
360        .is_some_and(|owner| !abandoned(owner.acquired_at))
361    {
362        return Ok(false);
363    }
364    if request.is_none() && windows_owner.is_none() {
365        return Ok(false);
366    }
367    retire_windows_recovery(root)?;
368    Ok(true)
369}
370
371fn remove_claim(directory: &Path) -> Result<(), FenceError> {
372    match fs::remove_dir_all(directory) {
373        Ok(()) => Ok(()),
374        Err(source) if source.kind() == std::io::ErrorKind::NotFound => Ok(()),
375        Err(source) => Err(io("release", directory, source)),
376    }
377}
378
379fn atomic_write(path: &Path, bytes: &[u8]) -> Result<(), FenceError> {
380    let parent = path.parent().unwrap_or_else(|| Path::new("."));
381    fs::create_dir_all(parent).map_err(|source| io("create", parent, source))?;
382    let mut temporary =
383        tempfile::NamedTempFile::new_in(parent).map_err(|source| io("write", path, source))?;
384    temporary
385        .write_all(bytes)
386        .and_then(|()| temporary.as_file().sync_all())
387        .map_err(|source| io("write", path, source))?;
388    temporary
389        .persist(path)
390        .map(|_| ())
391        .map_err(|error| io("replace", path, error.error))
392}
393
394fn write_json<T: Serialize>(path: &Path, value: &T) -> Result<(), FenceError> {
395    let bytes = serde_json::to_vec_pretty(value).map_err(|source| FenceError::Decode {
396        path: path.to_path_buf(),
397        source,
398    })?;
399    atomic_write(path, &bytes)
400}
401
402fn read_json<T: for<'de> Deserialize<'de>>(path: &Path) -> Result<Option<T>, FenceError> {
403    let bytes = match fs::read(path) {
404        Ok(bytes) => bytes,
405        Err(source) if source.kind() == std::io::ErrorKind::NotFound => return Ok(None),
406        Err(source) => return Err(io("read", path, source)),
407    };
408    let value: T = serde_json::from_slice(&bytes).map_err(|source| FenceError::Decode {
409        path: path.to_path_buf(),
410        source,
411    })?;
412    Ok(Some(value))
413}
414
415/// Persistent runner services are outside runner-manager's attempt journal.
416/// Their presence blocks automated WSL termination even while they look idle.
417#[must_use]
418pub fn unmanaged_runner_service_count() -> Option<u32> {
419    if !cfg!(target_os = "linux") {
420        return Some(0);
421    }
422    let mut names = std::collections::BTreeSet::new();
423    for directory in [
424        "/etc/systemd/system",
425        "/usr/lib/systemd/system",
426        "/lib/systemd/system",
427    ] {
428        let entries = match fs::read_dir(directory) {
429            Ok(entries) => entries,
430            Err(error) if error.kind() == std::io::ErrorKind::NotFound => continue,
431            Err(_) => return None,
432        };
433        for entry in entries.flatten() {
434            let name = entry.file_name().to_string_lossy().into_owned();
435            if name.starts_with("actions.runner.") && name.ends_with(".service") {
436                names.insert(name);
437            }
438        }
439    }
440    u32::try_from(names.len()).ok()
441}
442
443#[cfg(test)]
444mod tests {
445    use super::*;
446
447    #[test]
448    fn one_directory_has_exactly_one_owner_and_drop_releases_guest_claim() {
449        let root = tempfile::tempdir().unwrap();
450        let first = FenceClaim::try_claim(root.path(), FenceOwnerKind::GuestLaunch, None)
451            .unwrap()
452            .unwrap();
453        assert!(
454            FenceClaim::try_claim(root.path(), FenceOwnerKind::WindowsRecovery, Some(1))
455                .unwrap()
456                .is_none()
457        );
458        drop(first);
459        assert!(
460            FenceClaim::try_claim(root.path(), FenceOwnerKind::WindowsRecovery, Some(1))
461                .unwrap()
462                .is_some()
463        );
464    }
465
466    #[test]
467    fn only_matching_recovery_generation_is_cleared() {
468        let root = tempfile::tempdir().unwrap();
469        let claim = FenceClaim::try_claim(root.path(), FenceOwnerKind::WindowsRecovery, Some(9))
470            .unwrap()
471            .unwrap();
472        claim.make_durable();
473        DrainRequest::new(9, Utc::now()).write(root.path()).unwrap();
474        clear_recovery(root.path(), 8).unwrap();
475        assert!(root.path().join(FENCE_DIRECTORY).exists());
476        assert_eq!(
477            DrainRequest::read(root.path()).unwrap().unwrap().generation,
478            9
479        );
480        clear_recovery(root.path(), 9).unwrap();
481        assert!(!root.path().join(FENCE_DIRECTORY).exists());
482        assert!(!root.path().join(REQUEST_FILE).exists());
483    }
484
485    #[test]
486    fn retiring_windows_recovery_handles_restarted_generations_but_keeps_guest_claims() {
487        let root = tempfile::tempdir().unwrap();
488        let claim = FenceClaim::try_claim(root.path(), FenceOwnerKind::WindowsRecovery, Some(7))
489            .unwrap()
490            .unwrap();
491        claim.make_durable();
492        DrainRequest::new(8, Utc::now()).write(root.path()).unwrap();
493        RecoveryStatus {
494            schema_version: SCHEMA_VERSION,
495            observed_at: Utc::now(),
496            phase: RecoveryPhase::RecoveryBlocked,
497            consecutive_probe_failures: 9,
498            reason: Some("stale".into()),
499            last_recovered_at: None,
500        }
501        .write(root.path())
502        .unwrap();
503
504        retire_windows_recovery(root.path()).unwrap();
505
506        assert!(!root.path().join(FENCE_DIRECTORY).exists());
507        assert!(!root.path().join(REQUEST_FILE).exists());
508        assert!(!root.path().join(RECOVERY_STATUS_FILE).exists());
509
510        let guest = FenceClaim::try_claim(root.path(), FenceOwnerKind::GuestLaunch, None)
511            .unwrap()
512            .unwrap();
513        guest.make_durable();
514        DrainRequest::new(10, Utc::now())
515            .write(root.path())
516            .unwrap();
517        retire_windows_recovery(root.path()).unwrap();
518        assert!(root.path().join(FENCE_DIRECTORY).exists());
519        assert!(!root.path().join(REQUEST_FILE).exists());
520    }
521
522    #[test]
523    fn a_restarted_watchdog_retires_only_recovery_nobody_is_still_driving() {
524        let root = tempfile::tempdir().unwrap();
525        let now = Utc::now();
526        let window = std::time::Duration::from_secs(300);
527        let old = now - chrono::Duration::minutes(10);
528        let claim = FenceClaim::try_claim(root.path(), FenceOwnerKind::WindowsRecovery, Some(1))
529            .unwrap()
530            .unwrap();
531        claim.make_durable();
532        let owner_path = root.path().join(FENCE_DIRECTORY).join(OWNER_FILE);
533        let backdate_owner = || {
534            let mut owner = FenceClaim::owner(root.path()).unwrap().unwrap();
535            owner.acquired_at = old;
536            write_json(&owner_path, &owner).unwrap();
537        };
538
539        // A live watchdog keeps its request fresh: nothing is touched even
540        // though the claim itself is old and of another generation.
541        backdate_owner();
542        DrainRequest::new(2, now - chrono::Duration::seconds(10))
543            .write(root.path())
544            .unwrap();
545        assert!(!retire_abandoned_windows_recovery(root.path(), now, window).unwrap());
546        assert!(root.path().join(FENCE_DIRECTORY).exists());
547
548        // The incident: a dead generation's claim and request, WSL healthy.
549        DrainRequest::new(2, old).write(root.path()).unwrap();
550        assert!(retire_abandoned_windows_recovery(root.path(), now, window).unwrap());
551        assert!(!root.path().join(FENCE_DIRECTORY).exists());
552        assert!(!root.path().join(REQUEST_FILE).exists());
553        assert!(!retire_abandoned_windows_recovery(root.path(), now, window).unwrap());
554
555        // A freshly claimed fence whose request has not been rewritten yet.
556        let claim = FenceClaim::try_claim(root.path(), FenceOwnerKind::WindowsRecovery, Some(3))
557            .unwrap()
558            .unwrap();
559        claim.make_durable();
560        assert!(!retire_abandoned_windows_recovery(root.path(), now, window).unwrap());
561        assert!(root.path().join(FENCE_DIRECTORY).exists());
562        remove_claim(&root.path().join(FENCE_DIRECTORY)).unwrap();
563
564        // A guest launch claim is never retired, but an abandoned request is.
565        let guest = FenceClaim::try_claim(root.path(), FenceOwnerKind::GuestLaunch, None)
566            .unwrap()
567            .unwrap();
568        guest.make_durable();
569        backdate_owner();
570        DrainRequest::new(4, old).write(root.path()).unwrap();
571        assert!(retire_abandoned_windows_recovery(root.path(), now, window).unwrap());
572        assert!(root.path().join(FENCE_DIRECTORY).exists());
573        assert!(!root.path().join(REQUEST_FILE).exists());
574    }
575
576    #[test]
577    fn heartbeat_and_status_documents_replace_atomically() {
578        let root = tempfile::tempdir().unwrap();
579        let first = GuestHeartbeat {
580            schema_version: SCHEMA_VERSION,
581            observed_at: Utc::now(),
582            acknowledged_generation: None,
583            local_active_attempts: Some(1),
584            local_busy_attempts: Some(1),
585            managed_targets: Vec::new(),
586            unmanaged_runner_services: Some(0),
587        };
588        let mut second = first.clone();
589        second.local_active_attempts = Some(0);
590        first.write(root.path()).unwrap();
591        second.write(root.path()).unwrap();
592        assert_eq!(GuestHeartbeat::read(root.path()).unwrap(), Some(second));
593
594        let first = RecoveryStatus {
595            schema_version: SCHEMA_VERSION,
596            observed_at: Utc::now(),
597            phase: RecoveryPhase::Degraded,
598            consecutive_probe_failures: 1,
599            reason: None,
600            last_recovered_at: None,
601        };
602        let mut second = first.clone();
603        second.phase = RecoveryPhase::Healthy;
604        first.write(root.path()).unwrap();
605        second.write(root.path()).unwrap();
606        assert_eq!(RecoveryStatus::read(root.path()).unwrap(), Some(second));
607    }
608}
runner_manager_platform/wsl/fence.rs

runner_manager_platform/wsl/
fence.rs
