changes

src/rust/lqos_overrides/src/file_lock.rs

@@ -5,92 +5,548 @@ use nix::{
 };
 use std::{
     ffi::CString,
-    fs::{File, remove_file},
-    io::{Read, Write},
-    path::Path,
+    fs::{File, OpenOptions, hard_link, remove_file},
+    io::{ErrorKind, Read, Write},
+    os::fd::AsRawFd,
+    path::{Path, PathBuf},
+    sync::atomic::{AtomicU64, Ordering},
+    time::{SystemTime, UNIX_EPOCH},
 };
 
 const LOCK_PATH: &str = "/run/lqos/lqos_overrides.lock";
 const LOCK_DIR: &str = "/run/lqos";
 const LOCK_DIR_PERMS: &str = "/run/lqos";
+const LOCK_GUARD_PATH: &str = "/run/lqos/lqos_overrides.lock.guard";
+const STALE_LOCK_REPLACE_ATTEMPTS: usize = 3;
+const LOCK_TEMP_CREATE_ATTEMPTS: usize = 8;
+const LOCK_CONTENTION_CODE: &str = "LQOS_OVERRIDES_LOCKED";
+
+static TEMP_LOCK_SEQUENCE: AtomicU64 = AtomicU64::new(0);
 
 /// Cross-process lock used while mutating operator-owned override files.
-pub struct FileLock {}
+#[derive(Debug)]
+pub struct FileLock {
+    lock_path: PathBuf,
+}
+
+struct AcquisitionGuard {
+    file: File,
+}
+
+#[derive(Clone, Debug, PartialEq, Eq)]
+struct LockMetadata {
+    pid: i32,
+    process: Option<String>,
+    operation: Option<String>,
+    created_unix: Option<u64>,
+}
+
+impl LockMetadata {
+    fn current(operation: &str) -> Self {
+        Self {
+            pid: unsafe { getpid() },
+            process: current_process_name(),
+            operation: Some(sanitize_lock_field(operation)),
+            created_unix: SystemTime::now()
+                .duration_since(UNIX_EPOCH)
+                .ok()
+                .map(|duration| duration.as_secs()),
+        }
+    }
+
+    fn parse(contents: &str) -> Result<Self> {
+        let trimmed = contents.trim();
+        if let Ok(pid) = trimmed.parse::<i32>() {
+            return Ok(Self {
+                pid,
+                process: None,
+                operation: None,
+                created_unix: None,
+            });
+        }
+
+        let mut pid = None;
+        let mut process = None;
+        let mut operation = None;
+        let mut created_unix = None;
+        for line in contents.lines() {
+            let Some((key, value)) = line.split_once('=') else {
+                continue;
+            };
+            match key.trim() {
+                "pid" => pid = Some(value.trim().parse::<i32>()?),
+                "process" => process = Some(value.trim().to_string()),
+                "operation" => operation = Some(value.trim().to_string()),
+                "created_unix" => created_unix = Some(value.trim().parse::<u64>()?),
+                _ => {}
+            }
+        }
+
+        let Some(pid) = pid else {
+            return Err(Error::msg(
+                "The LibreQoS overrides lock file does not contain a process id.",
+            ));
+        };
+
+        Ok(Self {
+            pid,
+            process,
+            operation,
+            created_unix,
+        })
+    }
+
+    fn serialize(&self) -> String {
+        let process = self.process.as_deref().unwrap_or("unknown");
+        let operation = self.operation.as_deref().unwrap_or("unknown");
+        let created_unix = self
+            .created_unix
+            .map(|seconds| seconds.to_string())
+            .unwrap_or_else(|| "unknown".to_string());
+        format!(
+            "pid={}\nprocess={}\noperation={}\ncreated_unix={}\n",
+            self.pid, process, operation, created_unix
+        )
+    }
+
+    fn describe_holder(&self) -> String {
+        let mut parts = vec![format!("pid={}", self.pid)];
+        if let Some(process) = self.process.as_deref().filter(|value| !value.is_empty()) {
+            parts.push(format!("process={process}"));
+        }
+        if let Some(operation) = self.operation.as_deref().filter(|value| !value.is_empty()) {
+            parts.push(format!("operation={operation}"));
+        }
+        if let Some(created_unix) = self.created_unix {
+            parts.push(format!("created_unix={created_unix}"));
+        }
+        parts.join(", ")
+    }
+}
 
 impl FileLock {
-    /// Acquires the lock or returns an error if another live process holds it.
-    pub fn new() -> Result<Self> {
-        Self::check_directory()?;
-        let lock_path = Path::new(LOCK_PATH);
-        if lock_path.exists() {
-            if Self::is_lock_valid()? {
-                return Err(Error::msg(
-                    "The LibreQoS overrides files are locked by another process.",
-                ));
+    /// Acquires the lock and records the caller operation for diagnostics.
+    pub fn new_for_operation(operation: &str) -> Result<Self> {
+        Self::new_at(
+            operation,
+            Path::new(LOCK_PATH),
+            Path::new(LOCK_DIR),
+            Path::new(LOCK_DIR_PERMS),
+            Path::new(LOCK_GUARD_PATH),
+        )
+    }
+
+    fn new_at(
+        operation: &str,
+        lock_path: &Path,
+        lock_dir: &Path,
+        lock_dir_perms: &Path,
+        guard_path: &Path,
+    ) -> Result<Self> {
+        Self::check_directory_at(lock_dir, lock_dir_perms)?;
+        let _guard = Self::acquire_guard(guard_path)?;
+        for _ in 0..STALE_LOCK_REPLACE_ATTEMPTS {
+            if Self::create_lock(lock_path, lock_dir, operation)? {
+                return Ok(Self {
+                    lock_path: lock_path.to_path_buf(),
+                });
             }
 
-            Self::create_lock()?;
-            Ok(Self {})
-        } else {
-            Self::create_lock()?;
-            Ok(Self {})
+            let metadata = match Self::read_lock_metadata(lock_path) {
+                Ok(metadata) => metadata,
+                Err(err) if Self::is_not_found_error(&err) => continue,
+                Err(err) => {
+                    return Err(Error::msg(format!(
+                        "{LOCK_CONTENTION_CODE}: The LibreQoS overrides files are locked by another process (lock metadata unreadable: {err})."
+                    )));
+                }
+            };
+            if Self::is_lock_valid(&metadata) {
+                return Err(Error::msg(format!(
+                    "{LOCK_CONTENTION_CODE}: The LibreQoS overrides files are locked by another process ({}).",
+                    metadata.describe_holder()
+                )));
+            }
+
+            match remove_file(lock_path) {
+                Ok(()) => {}
+                Err(err) if err.kind() == ErrorKind::NotFound => {}
+                Err(err) => return Err(err.into()),
+            }
         }
+
+        Err(Error::msg(
+            "Unable to acquire the LibreQoS overrides lock after replacing a stale lock.",
+        ))
     }
 
-    fn is_lock_valid() -> Result<bool> {
-        let mut f = File::open(LOCK_PATH)?;
-        let mut contents = String::new();
-        f.read_to_string(&mut contents)?;
-        let pid: i32 = contents.trim().parse()?;
-
-        let ret = unsafe { nix::libc::kill(pid, 0) };
-        if ret == 0 {
-            return Ok(true);
-        }
-        let err = Errno::last();
-        if err == Errno::ESRCH {
-            Ok(false)
-        } else {
-            Ok(true)
-        }
-    }
-
-    fn create_lock() -> Result<()> {
-        let pid = unsafe { getpid() };
-        let pid_format = format!("{pid}");
-        {
-            let mut f = File::create(LOCK_PATH)?;
-            f.write_all(pid_format.as_bytes())?;
-        }
-        let unix_path = CString::new(LOCK_PATH)?;
+    fn acquire_guard(guard_path: &Path) -> Result<AcquisitionGuard> {
+        let file = OpenOptions::new()
+            .read(true)
+            .write(true)
+            .create(true)
+            .truncate(false)
+            .open(guard_path)?;
+        let unix_path = CString::new(guard_path.to_string_lossy().as_bytes())?;
         unsafe {
             nix::libc::chmod(unix_path.as_ptr(), mode_t::from_le(0o666));
         }
-        Ok(())
+        let ret = unsafe { nix::libc::flock(file.as_raw_fd(), nix::libc::LOCK_EX) };
+        if ret == 0 {
+            Ok(AcquisitionGuard { file })
+        } else {
+            Err(Error::msg(format!(
+                "Unable to acquire LibreQoS overrides lock guard: {}",
+                Errno::last()
+            )))
+        }
     }
 
-    fn check_directory() -> Result<()> {
-        let dir_path = std::path::Path::new(LOCK_DIR);
-        if dir_path.exists() && dir_path.is_dir() {
+    fn read_lock_metadata(lock_path: &Path) -> Result<LockMetadata> {
+        let mut f = File::open(lock_path)?;
+        let mut contents = String::new();
+        f.read_to_string(&mut contents)?;
+        LockMetadata::parse(&contents)
+    }
+
+    fn is_not_found_error(err: &Error) -> bool {
+        err.downcast_ref::<std::io::Error>()
+            .is_some_and(|io_err| io_err.kind() == ErrorKind::NotFound)
+    }
+
+    fn is_lock_valid(metadata: &LockMetadata) -> bool {
+        let ret = unsafe { nix::libc::kill(metadata.pid, 0) };
+        if ret == 0 {
+            return true;
+        }
+        let err = Errno::last();
+        err != Errno::ESRCH
+    }
+
+    fn create_lock(lock_path: &Path, lock_dir: &Path, operation: &str) -> Result<bool> {
+        let metadata = LockMetadata::current(operation);
+        let (mut f, temp_path) = Self::create_temp_lock_file(lock_dir)?;
+        f.write_all(metadata.serialize().as_bytes())?;
+        f.sync_all()?;
+        drop(f);
+        let link_result = hard_link(&temp_path, lock_path);
+        let _ = remove_file(&temp_path);
+        match link_result {
+            Ok(()) => {}
+            Err(err) if err.kind() == ErrorKind::AlreadyExists => return Ok(false),
+            Err(err) => return Err(err.into()),
+        }
+        let unix_path = CString::new(lock_path.to_string_lossy().as_bytes())?;
+        unsafe {
+            nix::libc::chmod(unix_path.as_ptr(), mode_t::from_le(0o666));
+        }
+        Ok(true)
+    }
+
+    fn create_temp_lock_file(lock_dir: &Path) -> Result<(File, PathBuf)> {
+        let nanos = SystemTime::now()
+            .duration_since(UNIX_EPOCH)
+            .ok()
+            .map(|duration| duration.as_nanos())
+            .unwrap_or_default();
+        for _ in 0..LOCK_TEMP_CREATE_ATTEMPTS {
+            let sequence = TEMP_LOCK_SEQUENCE.fetch_add(1, Ordering::Relaxed);
+            let temp_path = lock_dir.join(format!(
+                ".lqos_overrides.lock.{}.{}.{}.tmp",
+                std::process::id(),
+                nanos,
+                sequence
+            ));
+            match OpenOptions::new()
+                .write(true)
+                .create_new(true)
+                .open(&temp_path)
+            {
+                Ok(file) => return Ok((file, temp_path)),
+                Err(err) if err.kind() == ErrorKind::AlreadyExists => continue,
+                Err(err) => return Err(err.into()),
+            }
+        }
+
+        Err(Error::msg(format!(
+            "Unable to create a unique temporary LibreQoS overrides lock file after {LOCK_TEMP_CREATE_ATTEMPTS} attempts."
+        )))
+    }
+
+    fn check_directory_at(lock_dir: &Path, lock_dir_perms: &Path) -> Result<()> {
+        if lock_dir.exists() && lock_dir.is_dir() {
             Ok(())
         } else {
-            std::fs::create_dir(dir_path)?;
-            let unix_path = CString::new(LOCK_DIR_PERMS)?;
+            std::fs::create_dir(lock_dir)?;
+            let unix_path = CString::new(lock_dir_perms.to_string_lossy().as_bytes())?;
             unsafe {
                 nix::libc::chmod(unix_path.as_ptr(), 0o777);
             }
             Ok(())
         }
     }
-
-    /// Removes the current lock file, if present.
-    pub fn remove_lock() {
-        let _ = remove_file(LOCK_PATH);
-    }
 }
 
 impl Drop for FileLock {
     fn drop(&mut self) {
-        Self::remove_lock();
+        let _ = remove_file(&self.lock_path);
+    }
+}
+
+impl Drop for AcquisitionGuard {
+    fn drop(&mut self) {
+        let _ = unsafe { nix::libc::flock(self.file.as_raw_fd(), nix::libc::LOCK_UN) };
+    }
+}
+
+fn current_process_name() -> Option<String> {
+    std::fs::read_to_string("/proc/self/comm")
+        .ok()
+        .and_then(|name| {
+            let trimmed = sanitize_lock_field(name.trim());
+            if trimmed.is_empty() {
+                None
+            } else {
+                Some(trimmed)
+            }
+        })
+        .or_else(|| {
+            std::env::args().next().and_then(|arg| {
+                let trimmed = sanitize_lock_field(&arg);
+                if trimmed.is_empty() {
+                    None
+                } else {
+                    Some(trimmed)
+                }
+            })
+        })
+}
+
+fn sanitize_lock_field(value: &str) -> String {
+    value
+        .chars()
+        .filter(|character| *character != '\n' && *character != '\r')
+        .collect()
+}
+
+#[cfg(test)]
+mod tests {
+    use super::{FileLock, LockMetadata};
+    use std::{
+        fs::{create_dir_all, read_to_string, remove_dir_all, write},
+        path::PathBuf,
+        time::{SystemTime, UNIX_EPOCH},
+    };
+
+    fn unique_test_dir() -> PathBuf {
+        let nanos = SystemTime::now()
+            .duration_since(UNIX_EPOCH)
+            .expect("system clock before UNIX_EPOCH")
+            .as_nanos();
+        std::env::temp_dir().join(format!(
+            "lqos-overrides-lock-test-{}-{nanos}",
+            std::process::id()
+        ))
+    }
+
+    #[test]
+    fn parses_legacy_pid_only_lock_file() {
+        let metadata = LockMetadata::parse("12345\n").expect("legacy pid lock should parse");
+
+        assert_eq!(metadata.pid, 12345);
+        assert_eq!(metadata.process, None);
+        assert_eq!(metadata.operation, None);
+        assert_eq!(metadata.created_unix, None);
+    }
+
+    #[test]
+    fn parses_metadata_lock_file() {
+        let metadata = LockMetadata::parse(
+            "pid=12345\nprocess=lqosd\noperation=load effective overrides\ncreated_unix=1800000000\n",
+        )
+        .expect("metadata lock should parse");
+
+        assert_eq!(metadata.pid, 12345);
+        assert_eq!(metadata.process.as_deref(), Some("lqosd"));
+        assert_eq!(
+            metadata.operation.as_deref(),
+            Some("load effective overrides")
+        );
+        assert_eq!(metadata.created_unix, Some(1_800_000_000));
+    }
+
+    #[test]
+    fn holder_description_includes_available_metadata() {
+        let metadata = LockMetadata::parse(
+            "pid=12345\nprocess=node_manager\noperation=save overrides\ncreated_unix=1800000000\n",
+        )
+        .expect("metadata lock should parse");
+
+        assert_eq!(
+            metadata.describe_holder(),
+            "pid=12345, process=node_manager, operation=save overrides, created_unix=1800000000"
+        );
+    }
+
+    #[test]
+    fn lock_file_records_operation_metadata_and_cleans_up_on_drop() {
+        let dir = unique_test_dir();
+        create_dir_all(&dir).expect("failed to create temp test dir");
+        let path = dir.join("lqos_overrides.lock");
+
+        {
+            let _lock = FileLock::new_at(
+                "load effective overrides",
+                &path,
+                &dir,
+                &dir,
+                &dir.join("lqos_overrides.lock.guard"),
+            )
+            .expect("failed to create lock in temp dir");
+            let contents = read_to_string(&path).expect("failed to read temp lock");
+
+            assert!(contents.contains(&format!("pid={}", std::process::id())));
+            assert!(contents.contains("operation=load effective overrides"));
+            assert!(contents.contains("created_unix="));
+        }
+
+        assert!(!path.exists());
+        remove_dir_all(&dir).expect("failed to clean up temp test dir");
+    }
+
+    #[test]
+    fn live_lock_error_reports_existing_holder_metadata() {
+        let dir = unique_test_dir();
+        create_dir_all(&dir).expect("failed to create temp test dir");
+        let path = dir.join("lqos_overrides.lock");
+        write(
+            &path,
+            format!(
+                "pid={}\nprocess=test-holder\noperation=save overrides\ncreated_unix=1800000000\n",
+                std::process::id()
+            ),
+        )
+        .expect("failed to write temp lock");
+
+        let error = FileLock::new_at(
+            "load effective overrides",
+            &path,
+            &dir,
+            &dir,
+            &dir.join("lqos_overrides.lock.guard"),
+        )
+        .expect_err("live lock should reject a second holder");
+        let message = error.to_string();
+
+        assert!(message.contains("pid="));
+        assert!(message.contains("process=test-holder"));
+        assert!(message.contains("operation=save overrides"));
+        assert!(message.contains("created_unix=1800000000"));
+
+        remove_dir_all(&dir).expect("failed to clean up temp test dir");
+    }
+
+    #[test]
+    fn malformed_lock_file_is_reported_as_lock_contention() {
+        let dir = unique_test_dir();
+        create_dir_all(&dir).expect("failed to create temp test dir");
+        let path = dir.join("lqos_overrides.lock");
+        write(&path, "pid=").expect("failed to write malformed temp lock");
+
+        let error = FileLock::new_at(
+            "load effective overrides",
+            &path,
+            &dir,
+            &dir,
+            &dir.join("lqos_overrides.lock.guard"),
+        )
+        .expect_err("malformed lock metadata should be reported as contention");
+        let message = error.to_string();
+
+        assert!(message.contains("locked by another process"));
+        assert!(message.contains("lock metadata unreadable"));
+
+        remove_dir_all(&dir).expect("failed to clean up temp test dir");
+    }
+
+    #[test]
+    fn truncated_metadata_lock_missing_pid_is_reported_as_contention() {
+        let dir = unique_test_dir();
+        create_dir_all(&dir).expect("failed to create temp test dir");
+        let path = dir.join("lqos_overrides.lock");
+        write(
+            &path,
+            "process=test-holder\noperation=save overrides\ncreated_unix=1800000000\n",
+        )
+        .expect("failed to write truncated temp lock");
+
+        let error = FileLock::new_at(
+            "load effective overrides",
+            &path,
+            &dir,
+            &dir,
+            &dir.join("lqos_overrides.lock.guard"),
+        )
+        .expect_err("truncated lock metadata should be reported as contention");
+        let message = error.to_string();
+
+        assert!(message.contains("locked by another process"));
+        assert!(message.contains("lock metadata unreadable"));
+
+        remove_dir_all(&dir).expect("failed to clean up temp test dir");
+    }
+
+    #[test]
+    fn stale_legacy_lock_is_replaced_with_metadata_lock() {
+        let dir = unique_test_dir();
+        create_dir_all(&dir).expect("failed to create temp test dir");
+        let path = dir.join("lqos_overrides.lock");
+        write(&path, "2147483647\n").expect("failed to write stale temp lock");
+
+        {
+            let _lock = FileLock::new_at(
+                "load effective overrides",
+                &path,
+                &dir,
+                &dir,
+                &dir.join("lqos_overrides.lock.guard"),
+            )
+            .expect("stale legacy lock should be replaceable");
+            let contents = read_to_string(&path).expect("failed to read temp lock");
+
+            assert!(contents.contains(&format!("pid={}", std::process::id())));
+            assert!(contents.contains("operation=load effective overrides"));
+        }
+
+        remove_dir_all(&dir).expect("failed to clean up temp test dir");
+    }
+
+    #[test]
+    fn stale_structured_lock_is_replaced_with_metadata_lock() {
+        let dir = unique_test_dir();
+        create_dir_all(&dir).expect("failed to create temp test dir");
+        let path = dir.join("lqos_overrides.lock");
+        write(
+            &path,
+            "pid=2147483647\nprocess=old-holder\noperation=save overrides\ncreated_unix=1800000000\n",
+        )
+        .expect("failed to write stale structured temp lock");
+
+        {
+            let _lock = FileLock::new_at(
+                "load effective overrides",
+                &path,
+                &dir,
+                &dir,
+                &dir.join("lqos_overrides.lock.guard"),
+            )
+            .expect("stale structured lock should be replaceable");
+            let contents = read_to_string(&path).expect("failed to read temp lock");
+
+            assert!(contents.contains(&format!("pid={}", std::process::id())));
+            assert!(contents.contains("operation=load effective overrides"));
+        }
+
+        remove_dir_all(&dir).expect("failed to clean up temp test dir");
     }
 }

src/rust/lqos_overrides/src/overrides_file.rs

@@ -228,11 +228,14 @@ fn save_to_path(path: &Path, overrides: &OverrideFile) -> Result<()> {
 
 fn write_atomic(path: &Path, contents: &[u8]) -> Result<()> {
     let parent = path.parent().ok_or_else(|| {
-        anyhow::anyhow!("Overrides path '{}' has no parent directory", path.display())
-    })?;
-    let file_name = path.file_name().ok_or_else(|| {
-        anyhow::anyhow!("Overrides path '{}' has no file name", path.display())
+        anyhow::anyhow!(
+            "Overrides path '{}' has no parent directory",
+            path.display()
+        )
     })?;
+    let file_name = path
+        .file_name()
+        .ok_or_else(|| anyhow::anyhow!("Overrides path '{}' has no file name", path.display()))?;
     let pid = std::process::id();
     let nanos = std::time::SystemTime::now()
         .duration_since(std::time::UNIX_EPOCH)
@@ -604,7 +607,7 @@ impl OverrideFile {
 
     /// Loads the operator-owned overrides file, creating an empty file if it does not exist.
     pub fn load() -> Result<Self> {
-        let lock = FileLock::new()?;
+        let lock = FileLock::new_for_operation("load operator overrides")?;
         let config = lqos_config::load_config()?;
         let path = overrides_path(&config, OverrideLayer::Operator);
         ensure_exists_default(&path)?;
@@ -615,7 +618,7 @@ impl OverrideFile {
 
     /// Saves this value to the operator-owned overrides file.
     pub fn save(&self) -> Result<()> {
-        let lock = FileLock::new()?;
+        let lock = FileLock::new_for_operation("save operator overrides")?;
         let config = lqos_config::load_config()?;
         let path = overrides_path(&config, OverrideLayer::Operator);
         save_to_path(&path, self)?;
@@ -1080,7 +1083,7 @@ impl OverrideStore {
     ///
     /// Side effects: acquires the global overrides lock and may create the operator overrides file.
     pub fn load_layer(layer: OverrideLayer) -> Result<OverrideFile> {
-        let lock = FileLock::new()?;
+        let lock = FileLock::new_for_operation(&format!("load {layer:?} overrides"))?;
         let config = lqos_config::load_config()?;
         let path = Self::path_for_layer_config(&config, layer);
         let overrides = match layer {
@@ -1104,7 +1107,7 @@ impl OverrideStore {
     ///
     /// Side effects: acquires the global overrides lock and writes the selected overrides file.
     pub fn save_layer(layer: OverrideLayer, overrides: &OverrideFile) -> Result<()> {
-        let lock = FileLock::new()?;
+        let lock = FileLock::new_for_operation(&format!("save {layer:?} overrides"))?;
         let config = lqos_config::load_config()?;
         let path = overrides_path(&config, layer);
         save_to_path(&path, overrides)?;
@@ -1118,7 +1121,7 @@ impl OverrideStore {
     ///
     /// Side effects: acquires the global overrides lock and may create the operator overrides file.
     pub fn load_effective(apply_stormguard: bool, apply_treeguard: bool) -> Result<OverrideFile> {
-        let lock = FileLock::new()?;
+        let lock = FileLock::new_for_operation("load effective overrides")?;
         let config = lqos_config::load_config()?;
         let merged =
             Self::load_effective_for_config_inner(&config, apply_stormguard, apply_treeguard)?;

src/rust/lqos_overrides/src/overrides_file/file_lock.rs

@@ -1,98 +0,0 @@
-use anyhow::{Error, Result};
-use nix::{
-    errno::Errno,
-    libc::{getpid, mode_t},
-};
-use std::{
-    ffi::CString,
-    fs::{File, remove_file},
-    io::{Read, Write},
-    path::Path,
-};
-
-const LOCK_PATH: &str = "/run/lqos/lqos_overrides.lock";
-const LOCK_DIR: &str = "/run/lqos";
-const LOCK_DIR_PERMS: &str = "/run/lqos";
-
-pub struct FileLock {}
-
-impl FileLock {
-    pub fn new() -> Result<Self> {
-        Self::check_directory()?;
-        let lock_path = Path::new(LOCK_PATH);
-        if lock_path.exists() {
-            if Self::is_lock_valid()? {
-                return Err(Error::msg(
-                    "The lqos_overrides file is locked by another process.",
-                ));
-            }
-
-            // It's a stale pid, so we need to replace it
-            Self::create_lock()?;
-
-            Ok(Self {})
-        } else {
-            Self::create_lock()?;
-            Ok(Self {})
-        }
-    }
-
-    fn is_lock_valid() -> Result<bool> {
-        let mut f = File::open(LOCK_PATH)?;
-        let mut contents = String::new();
-        f.read_to_string(&mut contents)?;
-        let pid: i32 = contents.trim().parse()?;
-
-        // Check if the process exists by sending signal 0.
-        // Returns 0 if the process exists; -1 with ESRCH if it doesn't;
-        // other errors (e.g., EPERM) mean the process exists but we lack permission.
-        let ret = unsafe { nix::libc::kill(pid, 0) };
-        if ret == 0 {
-            return Ok(true);
-        }
-        let err = Errno::last();
-        if err == Errno::ESRCH {
-            Ok(false)
-        } else {
-            Ok(true)
-        }
-    }
-
-    fn create_lock() -> Result<()> {
-        let pid = unsafe { getpid() };
-        let pid_format = format!("{pid}");
-        {
-            let mut f = File::create(LOCK_PATH)?;
-            f.write_all(pid_format.as_bytes())?;
-        }
-        let unix_path = CString::new(LOCK_PATH)?;
-        unsafe {
-            nix::libc::chmod(unix_path.as_ptr(), mode_t::from_le(666));
-        }
-        Ok(())
-    }
-
-    fn check_directory() -> Result<()> {
-        let dir_path = std::path::Path::new(LOCK_DIR);
-        if dir_path.exists() && dir_path.is_dir() {
-            Ok(())
-        } else {
-            std::fs::create_dir(dir_path)?;
-            let unix_path = CString::new(LOCK_DIR_PERMS)?;
-            unsafe {
-                nix::libc::chmod(unix_path.as_ptr(), 777);
-            }
-            Ok(())
-        }
-    }
-
-    pub fn remove_lock() {
-        let _ = remove_file(LOCK_PATH); // Ignore result
-    }
-}
-
-impl Drop for FileLock {
-    fn drop(&mut self) {
-        Self::remove_lock();
-    }
-}

src/rust/lqos_overrides/src/topology_overrides.rs

@@ -200,7 +200,7 @@ impl TopologyOverridesFile {
 
     /// Saves this value to the operator-owned topology-manager overrides file.
     pub fn save(&self) -> Result<()> {
-        let lock = FileLock::new()?;
+        let lock = FileLock::new_for_operation("save topology overrides")?;
         let config = lqos_config::load_config()?;
         let path = topology_overrides_path(&config);
         save_to_path(&path, self)?;

src/rust/lqos_python/src/lib.rs

@@ -1067,6 +1067,7 @@ fn liblqos_python(m: &Bound<'_, PyModule>) -> PyResult<()> {
         overrides_network_adjustments_materialized,
         m
     )?)?;
+    m.add_function(wrap_pyfunction!(overrides_materialized, m)?)?;
     m.add_function(wrap_pyfunction!(is_network_flat, m)?)?;
     m.add_function(wrap_pyfunction!(blackboard_finish, m)?)?;
     m.add_function(wrap_pyfunction!(blackboard_submit, m)?)?;
@@ -1564,13 +1565,17 @@ fn overrides_persistent_devices_effective(py: Python<'_>) -> PyResult<Vec<PyObje
 /// overwrite the source-of-truth CSV.
 #[pyfunction]
 fn overrides_persistent_devices_materialized(py: Python<'_>) -> PyResult<Vec<PyObject>> {
-    let overrides = match lqos_overrides::OverrideStore::load_effective(false, false) {
-        Ok(o) => o,
-        Err(e) => return Err(PyOSError::new_err(e.to_string())),
-    };
+    let overrides = load_materialized_overrides()?;
 
+    persistent_devices_to_py(py, overrides.persistent_devices())
+}
+
+fn persistent_devices_to_py(
+    py: Python<'_>,
+    devices: &[lqos_config::ShapedDevice],
+) -> PyResult<Vec<PyObject>> {
     let mut out: Vec<PyObject> = Vec::new();
-    for dev in overrides.persistent_devices().iter() {
+    for dev in devices.iter() {
         let ipv4s: Vec<String> = dev
             .ipv4
             .iter()
@@ -1813,13 +1818,17 @@ fn overrides_circuit_adjustments_effective(py: Python<'_>) -> PyResult<Vec<PyObj
 /// overwrite the source-of-truth CSV.
 #[pyfunction]
 fn overrides_circuit_adjustments_materialized(py: Python<'_>) -> PyResult<Vec<PyObject>> {
-    let overrides = match lqos_overrides::OverrideStore::load_effective(false, false) {
-        Ok(o) => o,
-        Err(e) => return Err(PyOSError::new_err(e.to_string())),
-    };
+    let overrides = load_materialized_overrides()?;
 
+    circuit_adjustments_to_py(py, overrides.circuit_adjustments())
+}
+
+fn circuit_adjustments_to_py(
+    py: Python<'_>,
+    adjustments: &[lqos_overrides::CircuitAdjustment],
+) -> PyResult<Vec<PyObject>> {
     let mut out: Vec<PyObject> = Vec::new();
-    for adj in overrides.circuit_adjustments().iter() {
+    for adj in adjustments.iter() {
         let d = PyDict::new(py);
         match adj {
             lqos_overrides::CircuitAdjustment::CircuitAdjustSpeed {
@@ -1940,14 +1949,41 @@ fn overrides_network_adjustments_effective(py: Python<'_>) -> PyResult<Vec<PyObj
 /// file.
 #[pyfunction]
 fn overrides_network_adjustments_materialized(py: Python<'_>) -> PyResult<Vec<PyObject>> {
-    let overrides = match lqos_overrides::OverrideStore::load_effective(false, false) {
-        Ok(o) => o,
-        Err(e) => return Err(PyOSError::new_err(e.to_string())),
-    };
+    let overrides = load_materialized_overrides()?;
 
     network_adjustments_to_py(py, overrides.network_adjustments())
 }
 
+/// Returns all operator-owned override sections that scheduler materializes into
+/// source-of-truth compatibility files.
+///
+/// Side effects: acquires the overrides file lock once and reads the effective
+/// operator-only override snapshot.
+#[pyfunction]
+fn overrides_materialized(py: Python<'_>) -> PyResult<PyObject> {
+    let overrides = load_materialized_overrides()?;
+
+    let out = PyDict::new(py);
+    out.set_item(
+        "persistent_devices",
+        persistent_devices_to_py(py, overrides.persistent_devices())?,
+    )?;
+    out.set_item(
+        "circuit_adjustments",
+        circuit_adjustments_to_py(py, overrides.circuit_adjustments())?,
+    )?;
+    out.set_item(
+        "network_adjustments",
+        network_adjustments_to_py(py, overrides.network_adjustments())?,
+    )?;
+    Ok(out.unbind().into())
+}
+
+fn load_materialized_overrides() -> PyResult<lqos_overrides::OverrideFile> {
+    lqos_overrides::OverrideStore::load_effective(false, false)
+        .map_err(|e| PyOSError::new_err(e.to_string()))
+}
+
 fn network_adjustments_to_py(
     py: Python<'_>,
     adjustments: &[lqos_overrides::NetworkAdjustment],

src/rust/lqos_topology/src/lib.rs

@@ -10,13 +10,14 @@ use lqos_config::{
     TopLevelPlannerItem, TopLevelPlannerMode, TopLevelPlannerParams, TopologyAllowedParent,
     TopologyAttachmentHealthStateFile, TopologyAttachmentHealthStatus, TopologyAttachmentOption,
     TopologyAttachmentRateSource, TopologyAttachmentRole, TopologyCanonicalIngressKind,
-    TopologyCanonicalNode, TopologyCanonicalStateFile, TopologyEditorNode, TopologyEditorStateFile,
-    TopologyEffectiveAttachmentState, TopologyEffectiveNodeState, TopologyEffectiveStateFile,
-    TopologyQueueVisibilityPolicy, TopologyRuntimeStatusFile, TopologyShapingCircuitInput,
-    TopologyShapingDeviceInput, TopologyShapingInputsFile, TopologyShapingResolutionSource,
-    circuit_anchors_path, compute_effective_network_generation, detect_shaping_cpus,
-    plan_top_level_assignments, topology_effective_network_path, topology_effective_state_path,
-    topology_runtime_status_path, topology_shaping_inputs_path,
+    TopologyCanonicalNode, TopologyCanonicalRateInputSource, TopologyCanonicalStateFile,
+    TopologyEditorNode, TopologyEditorStateFile, TopologyEffectiveAttachmentState,
+    TopologyEffectiveNodeState, TopologyEffectiveStateFile, TopologyQueueVisibilityPolicy,
+    TopologyRuntimeStatusFile, TopologyShapingCircuitInput, TopologyShapingDeviceInput,
+    TopologyShapingInputsFile, TopologyShapingResolutionSource, circuit_anchors_path,
+    compute_effective_network_generation, detect_shaping_cpus, plan_top_level_assignments,
+    topology_effective_network_path, topology_effective_state_path, topology_runtime_status_path,
+    topology_shaping_inputs_path,
 };
 use lqos_overrides::{
     CircuitAdjustment, NetworkAdjustment, OverrideStore, TopologyAttachmentMode,
@@ -3269,7 +3270,16 @@ fn rate_pair_from_value(node: &Map<String, Value>) -> CompiledRatePair {
     )
 }
 
-fn rate_pair_from_canonical_node(node: &TopologyCanonicalNode) -> CompiledRatePair {
+fn rate_pair_from_canonical_node(
+    node: &TopologyCanonicalNode,
+    use_compatibility_export_rates: bool,
+) -> CompiledRatePair {
+    if node.rate_input.source == TopologyCanonicalRateInputSource::CompatibilityExport
+        && !use_compatibility_export_rates
+    {
+        return CompiledRatePair::default();
+    }
+
     compiled_rate_pair(
         node.rate_input
             .intrinsic_download_mbps
@@ -3352,16 +3362,21 @@ fn recompile_effective_bandwidths_for_value(
     canonical_nodes: &HashMap<&str, &TopologyCanonicalNode>,
     selected_attachment_caps: &HashMap<String, CompiledRatePair>,
     inherited_parent_rates: Option<CompiledRatePair>,
+    use_compatibility_export_rates: bool,
 ) {
     let Some(node) = value.as_object_mut() else {
         return;
     };
     let existing_rates = rate_pair_from_value(node);
     let node_id = node.get("id").and_then(Value::as_str);
-    let mut compiled = node_id
-        .and_then(|node_id| canonical_nodes.get(node_id).copied())
-        .map(rate_pair_from_canonical_node)
+    let canonical_node = node_id.and_then(|node_id| canonical_nodes.get(node_id).copied());
+    let mut compiled = canonical_node
+        .map(|node| rate_pair_from_canonical_node(node, use_compatibility_export_rates))
         .unwrap_or(existing_rates);
+    let skip_existing_compatibility_rates = canonical_node.is_some_and(|node| {
+        node.rate_input.source == TopologyCanonicalRateInputSource::CompatibilityExport
+            && !use_compatibility_export_rates
+    });
     if let Some(node_id) = node_id
         && let Some(attachment_rates) = selected_attachment_caps.get(node_id)
     {
@@ -3371,14 +3386,22 @@ fn recompile_effective_bandwidths_for_value(
         compiled = intersect_rate_pairs(compiled, parent_rates);
     }
     if compiled.download.is_none() {
-        compiled.download = existing_rates
-            .download
-            .or(inherited_parent_rates.and_then(|pair| pair.download));
+        compiled.download = if skip_existing_compatibility_rates {
+            inherited_parent_rates.and_then(|pair| pair.download)
+        } else {
+            existing_rates
+                .download
+                .or(inherited_parent_rates.and_then(|pair| pair.download))
+        };
     }
     if compiled.upload.is_none() {
-        compiled.upload = existing_rates
-            .upload
-            .or(inherited_parent_rates.and_then(|pair| pair.upload));
+        compiled.upload = if skip_existing_compatibility_rates {
+            inherited_parent_rates.and_then(|pair| pair.upload)
+        } else {
+            existing_rates
+                .upload
+                .or(inherited_parent_rates.and_then(|pair| pair.upload))
+        };
     }
     write_rate_pair(node, compiled);
     let next_parent_rates = Some(rate_pair_from_value(node));
@@ -3389,6 +3412,7 @@ fn recompile_effective_bandwidths_for_value(
                 canonical_nodes,
                 selected_attachment_caps,
                 next_parent_rates,
+                use_compatibility_export_rates,
             );
         }
     }
@@ -3406,12 +3430,15 @@ fn recompile_effective_network_bandwidths(
         .map(|node| (node.node_id.as_str(), node))
         .collect::<HashMap<_, _>>();
     let selected_attachment_caps = selected_attachment_rate_caps(ui_state, effective);
+    let use_compatibility_export_rates =
+        canonical.ingress_kind != TopologyCanonicalIngressKind::NativeIntegration;
     for node in root.values_mut() {
         recompile_effective_bandwidths_for_value(
             node,
             &canonical_nodes,
             &selected_attachment_caps,
             None,
+            use_compatibility_export_rates,
         );
     }
 }
@@ -4376,10 +4403,11 @@ mod tests {
         CircuitAnchor, CircuitAnchorsFile, Config, ConfigShapedDevices, ShapedDevice,
         TopologyAllowedParent, TopologyAttachmentHealthStateFile, TopologyAttachmentHealthStatus,
         TopologyAttachmentOption, TopologyAttachmentRateSource, TopologyAttachmentRole,
-        TopologyCanonicalIngressKind, TopologyCanonicalNode, TopologyCanonicalStateFile,
-        TopologyEditorNode, TopologyEditorStateFile, TopologyEffectiveAttachmentState,
-        TopologyEffectiveNodeState, TopologyEffectiveStateFile, TopologyQueueVisibilityPolicy,
-        TopologyRuntimeStatusFile, topology_effective_network_path, topology_effective_state_path,
+        TopologyCanonicalIngressKind, TopologyCanonicalNode, TopologyCanonicalRateInput,
+        TopologyCanonicalRateInputSource, TopologyCanonicalStateFile, TopologyEditorNode,
+        TopologyEditorStateFile, TopologyEffectiveAttachmentState, TopologyEffectiveNodeState,
+        TopologyEffectiveStateFile, TopologyQueueVisibilityPolicy, TopologyRuntimeStatusFile,
+        topology_effective_network_path, topology_effective_state_path,
         topology_runtime_status_path, topology_shaping_inputs_path,
     };
     use lqos_overrides::{TopologyAttachmentMode, TopologyOverridesFile};
@@ -4399,6 +4427,150 @@ mod tests {
         path
     }
 
+    fn canonical_node_with_rate_source(
+        node_id: &str,
+        node_name: &str,
+        download: u64,
+        upload: u64,
+        source: TopologyCanonicalRateInputSource,
+    ) -> TopologyCanonicalNode {
+        TopologyCanonicalNode {
+            node_id: node_id.to_string(),
+            node_name: node_name.to_string(),
+            latitude: None,
+            longitude: None,
+            node_kind: "Site".to_string(),
+            is_virtual: false,
+            current_parent_node_id: None,
+            current_parent_node_name: None,
+            current_attachment_id: None,
+            current_attachment_name: None,
+            can_move: true,
+            allowed_parents: Vec::new(),
+            queue_visibility_policy: TopologyQueueVisibilityPolicy::QueueVisible,
+            rate_input: TopologyCanonicalRateInput {
+                intrinsic_download_mbps: Some(download),
+                intrinsic_upload_mbps: Some(upload),
+                legacy_imported_download_mbps: Some(download),
+                legacy_imported_upload_mbps: Some(upload),
+                source,
+            },
+        }
+    }
+
+    fn sample_effective_bandwidth_tree() -> serde_json::Map<String, Value> {
+        json!({
+            "Hoodoo Hill": {
+                "children": {
+                    "HoodooHill-Thunderhill": {
+                        "children": {},
+                        "downloadBandwidthMbps": 214,
+                        "id": "ap-thunderhill",
+                        "name": "HoodooHill-Thunderhill",
+                        "type": "AP",
+                        "uploadBandwidthMbps": 773
+                    }
+                },
+                "downloadBandwidthMbps": 774,
+                "id": "site-hoodoo",
+                "name": "Hoodoo Hill",
+                "type": "Site",
+                "uploadBandwidthMbps": 774
+            }
+        })
+        .as_object()
+        .expect("sample tree should be an object")
+        .clone()
+    }
+
+    #[test]
+    fn native_compatibility_export_rates_do_not_clamp_auto_child_nodes() {
+        let mut root = sample_effective_bandwidth_tree();
+        let canonical = TopologyCanonicalStateFile {
+            ingress_kind: TopologyCanonicalIngressKind::NativeIntegration,
+            nodes: vec![
+                canonical_node_with_rate_source(
+                    "site-hoodoo",
+                    "Hoodoo Hill",
+                    774,
+                    774,
+                    TopologyCanonicalRateInputSource::AttachmentMax,
+                ),
+                canonical_node_with_rate_source(
+                    "ap-thunderhill",
+                    "HoodooHill-Thunderhill",
+                    214,
+                    773,
+                    TopologyCanonicalRateInputSource::CompatibilityExport,
+                ),
+            ],
+            ..Default::default()
+        };
+
+        super::recompile_effective_network_bandwidths(
+            &mut root,
+            &canonical,
+            &TopologyEditorStateFile::default(),
+            &TopologyEffectiveStateFile::default(),
+        );
+
+        let ap = root["Hoodoo Hill"]["children"]["HoodooHill-Thunderhill"]
+            .as_object()
+            .expect("AP node should exist");
+        assert_eq!(
+            super::node_bandwidth_mbps(ap, "downloadBandwidthMbps"),
+            Some(774)
+        );
+        assert_eq!(
+            super::node_bandwidth_mbps(ap, "uploadBandwidthMbps"),
+            Some(774)
+        );
+    }
+
+    #[test]
+    fn legacy_imported_rates_still_cap_child_nodes() {
+        let mut root = sample_effective_bandwidth_tree();
+        let canonical = TopologyCanonicalStateFile {
+            ingress_kind: TopologyCanonicalIngressKind::LegacyNetworkJson,
+            nodes: vec![
+                canonical_node_with_rate_source(
+                    "site-hoodoo",
+                    "Hoodoo Hill",
+                    774,
+                    774,
+                    TopologyCanonicalRateInputSource::ImportedNetworkJson,
+                ),
+                canonical_node_with_rate_source(
+                    "ap-thunderhill",
+                    "HoodooHill-Thunderhill",
+                    214,
+                    773,
+                    TopologyCanonicalRateInputSource::ImportedNetworkJson,
+                ),
+            ],
+            ..Default::default()
+        };
+
+        super::recompile_effective_network_bandwidths(
+            &mut root,
+            &canonical,
+            &TopologyEditorStateFile::default(),
+            &TopologyEffectiveStateFile::default(),
+        );
+
+        let ap = root["Hoodoo Hill"]["children"]["HoodooHill-Thunderhill"]
+            .as_object()
+            .expect("AP node should exist");
+        assert_eq!(
+            super::node_bandwidth_mbps(ap, "downloadBandwidthMbps"),
+            Some(214)
+        );
+        assert_eq!(
+            super::node_bandwidth_mbps(ap, "uploadBandwidthMbps"),
+            Some(773)
+        );
+    }
+
     fn write_runtime_json_fixture(path: PathBuf, value: &Value, label: &str) {
         let parent = path
             .parent()

src/scheduler.py

@@ -14,8 +14,8 @@ from liblqos_python import automatic_import_uisp, automatic_import_splynx, queue
     blackboard_finish, blackboard_submit, automatic_import_wispgate, enable_insight_topology, insight_topology_role, \
     automatic_import_netzur, automatic_import_visp, calculate_shaping_runtime_hash, efficiency_core_ids, scheduler_alive as _scheduler_alive_native, scheduler_error as _scheduler_error_native, \
     calculate_topology_source_generation, topology_import_ingress_enabled, \
-    scheduler_progress as _scheduler_progress_native, overrides_persistent_devices_materialized, overrides_circuit_adjustments_materialized, \
-    overrides_network_adjustments_materialized, \
+    scheduler_progress as _scheduler_progress_native, overrides_network_adjustments_materialized, \
+    overrides_materialized, \
     scheduler_output as _scheduler_output_native, wait_for_bus_ready
 
 from apscheduler.schedulers.background import BlockingScheduler
@@ -37,6 +37,13 @@ STARTUP_TOPOLOGY_RUNTIME_MAX_WAIT_SECONDS = 120
 PARTIAL_TOPOLOGY_RUNTIME_MAX_WAIT_SECONDS = 120
 SCHEDULER_STARTUP_STEP_COUNT = 5
 SCHEDULER_REFRESH_STEP_COUNT = 4
+OVERRIDE_SECTION_READ_ATTEMPTS = 5
+OVERRIDE_SECTION_READ_RETRY_SECONDS = 0.25
+OVERRIDE_LOCK_ERROR_MARKERS = (
+    "lqos_overrides_locked",
+    "overrides files are locked by another process",
+    "locked by another process",
+)
 scheduler_status_bus_enabled = True
 startup_topology_runtime_pending = False
 startup_topology_runtime_generation = None
@@ -49,6 +56,10 @@ partial_topology_runtime_last_report_state = None
 last_integration_failure_message = ""
 
 
+class RequiredOverrideReadError(RuntimeError):
+    """Raised when scheduler reloads cannot safely materialize operator overrides."""
+
+
 def configure_scheduler_stdio():
     """
     Enable line-buffered scheduler logs when running under systemd.
@@ -493,8 +504,19 @@ def importFromCRM(
         )
         apply_lqos_overrides()
     except Exception as e:
-        scheduler_error(f"Failed to apply lqos_overrides: {e}")
-        print(f"Failed to apply lqos_overrides: {e}")
+        message = f"Failed to apply lqos_overrides; preserving last-known-good topology: {e}"
+        scheduler_error(message)
+        scheduler_output(message)
+        publish_scheduler_progress(
+            False,
+            "degraded",
+            "Scheduler blocked by override failure",
+            overrides_step,
+            progress_step_count,
+            percent=100,
+        )
+        print(message)
+        raise
 
 
 # --------------- Overrides Handling ---------------
@@ -704,23 +726,23 @@ def operator_requires_sqm_column(devices, adjustments):
 
 def apply_lqos_overrides():
     """Apply overrides to source topology files without mutating integration-owned shaping CSV."""
+    materialized = read_required_override_section(
+        "materialized overrides",
+        overrides_materialized,
+    )
+    persistent_devices = materialized.get("persistent_devices", [])
+    circuit_adjustments = materialized.get("circuit_adjustments", [])
+    network_adjustments = materialized.get("network_adjustments", [])
+
     if not topology_import_ingress_enabled():
         path = shaped_devices_csv_path()
         header, rows = read_shaped_devices_csv(path)
 
         # 1) Persistent devices: replace by device_id or append
-        try:
-            extra = overrides_persistent_devices_materialized()
-        except Exception as e:
-            print(f"Skipping persistent device overrides: {e}")
-            extra = []
+        extra = persistent_devices
 
         # 2) Circuit adjustments: speed changes, removals, reparenting
-        try:
-            adjustments = overrides_circuit_adjustments_materialized()
-        except Exception as e:
-            print(f"Failed to read circuit adjustments: {e}")
-            adjustments = []
+        adjustments = circuit_adjustments
 
         need_sqm_column = header_has_sqm(header) or operator_requires_sqm_column(extra, adjustments)
         if need_sqm_column and not header_has_sqm(header):
@@ -806,11 +828,7 @@ def apply_lqos_overrides():
             print("Updated ShapedDevices.csv with overrides")
 
     # 3) Load, adjust, and optionally save topology compatibility state
-    try:
-        adjustments = overrides_network_adjustments_materialized()
-    except Exception as e:
-        print(f"Failed to read network adjustments: {e}")
-        adjustments = []
+    adjustments = network_adjustments
     canonical_path = topology_canonical_state_path()
     canonical_state = load_topology_canonical_state(canonical_path)
     if topology_import_ingress_enabled():
@@ -870,6 +888,35 @@ def write_topology_canonical_state(path: str, canonical_state: dict):
         f.write(json.dumps(canonical_state, indent=4))
 
 
+def _format_attempt_count(attempts: int) -> str:
+    if attempts == 1:
+        return "1 attempt"
+    return f"{attempts} attempts"
+
+
+def _is_override_lock_error(exc: Exception) -> bool:
+    message = str(exc or "").lower()
+    return any(marker in message for marker in OVERRIDE_LOCK_ERROR_MARKERS)
+
+
+def read_required_override_section(section_name: str, reader):
+    attempts = max(1, int(OVERRIDE_SECTION_READ_ATTEMPTS))
+    last_error = None
+    for attempt_index in range(attempts):
+        try:
+            return reader()
+        except Exception as exc:
+            last_error = exc
+            if not _is_override_lock_error(exc):
+                raise
+            if attempt_index + 1 < attempts:
+                time.sleep(OVERRIDE_SECTION_READ_RETRY_SECONDS)
+
+    raise RequiredOverrideReadError(
+        f"failed to read {section_name} after {_format_attempt_count(attempts)}: {last_error}"
+    ) from last_error
+
+
 def apply_network_adjustments(network: dict, adjustments=None) -> bool:
     """Apply network adjustments from overrides to the network JSON structure.
 
@@ -886,11 +933,10 @@ def apply_network_adjustments(network: dict, adjustments=None) -> bool:
     Returns True if any changes were applied.
     """
     if adjustments is None:
-        try:
-            adjustments = overrides_network_adjustments_materialized()
-        except Exception as e:
-            print(f"Failed to read network adjustments: {e}")
-            return False
+        adjustments = read_required_override_section(
+            "network adjustments",
+            overrides_network_adjustments_materialized,
+        )
 
     if not adjustments:
         return False

src/test_scheduler.py

@@ -59,11 +59,14 @@ def install_scheduler_stubs():
     lqlib.scheduler_progress = Mock()
     lqlib.wait_for_bus_ready = Mock(return_value=True)
     lqlib.overrides_persistent_devices_effective = lambda: []
-    lqlib.overrides_persistent_devices_materialized = lambda: []
     lqlib.overrides_circuit_adjustments_effective = lambda: []
-    lqlib.overrides_circuit_adjustments_materialized = lambda: []
     lqlib.overrides_network_adjustments_effective = lambda: []
     lqlib.overrides_network_adjustments_materialized = lambda: []
+    lqlib.overrides_materialized = lambda: {
+        "persistent_devices": [],
+        "circuit_adjustments": [],
+        "network_adjustments": [],
+    }
     sys.modules["liblqos_python"] = lqlib
 
     apscheduler_pkg = types.ModuleType("apscheduler")
@@ -1235,6 +1238,165 @@ class TestSchedulerOverrideMerge(unittest.TestCase):
         self.assertEqual(len(merged), 2)
         self.assertEqual(merged[1][2], "legacy_device_1")
 
+    def test_required_override_section_retries_transient_lock_failure(self):
+        reader = Mock(side_effect=[
+            RuntimeError("locked by another process"),
+            RuntimeError("still locked by another process"),
+            [{"type": "adjust_site_speed"}],
+        ])
+
+        with patch.object(scheduler, "OVERRIDE_SECTION_READ_ATTEMPTS", 5):
+            with patch.object(scheduler, "OVERRIDE_SECTION_READ_RETRY_SECONDS", 0.01):
+                with patch.object(scheduler.time, "sleep") as mock_sleep:
+                    adjustments = scheduler.read_required_override_section(
+                        "network adjustments",
+                        reader,
+                    )
+
+        self.assertEqual(adjustments, [{"type": "adjust_site_speed"}])
+        self.assertEqual(reader.call_count, 3)
+        self.assertEqual(mock_sleep.call_count, 2)
+
+    def test_required_override_section_raises_after_retry_budget(self):
+        reader = Mock(side_effect=RuntimeError("locked by another process: pid 42"))
+
+        with patch.object(scheduler, "OVERRIDE_SECTION_READ_ATTEMPTS", 2):
+            with patch.object(scheduler.time, "sleep"):
+                with self.assertRaises(scheduler.RequiredOverrideReadError) as raised:
+                    scheduler.read_required_override_section(
+                        "network adjustments",
+                        reader,
+                    )
+
+        self.assertEqual(reader.call_count, 2)
+        self.assertIn("failed to read network adjustments after 2 attempts", str(raised.exception))
+        self.assertIn("locked by another process: pid 42", str(raised.exception))
+
+    def test_required_override_section_does_not_retry_non_lock_failure(self):
+        reader = Mock(side_effect=RuntimeError("invalid overrides json"))
+
+        with patch.object(scheduler, "OVERRIDE_SECTION_READ_ATTEMPTS", 5):
+            with patch.object(scheduler.time, "sleep") as mock_sleep:
+                with self.assertRaises(RuntimeError) as raised:
+                    scheduler.read_required_override_section(
+                        "network adjustments",
+                        reader,
+                    )
+
+        self.assertEqual(reader.call_count, 1)
+        mock_sleep.assert_not_called()
+        self.assertIn("invalid overrides json", str(raised.exception))
+
+    def test_apply_lqos_overrides_aborts_when_network_adjustments_are_unavailable(self):
+        with patch.object(scheduler, "topology_import_ingress_enabled", return_value=True):
+            with patch.object(
+                scheduler,
+                "overrides_materialized",
+                side_effect=RuntimeError("locked by another process: pid 42"),
+            ):
+                with patch.object(scheduler, "OVERRIDE_SECTION_READ_ATTEMPTS", 2):
+                    with patch.object(scheduler.time, "sleep"):
+                        with patch.object(scheduler, "load_topology_canonical_state") as mock_load_canonical:
+                            with patch.object(scheduler, "write_topology_canonical_state") as mock_write_canonical:
+                                with self.assertRaises(scheduler.RequiredOverrideReadError):
+                                    scheduler.apply_lqos_overrides()
+
+        mock_load_canonical.assert_not_called()
+        mock_write_canonical.assert_not_called()
+
+    def test_apply_lqos_overrides_reads_all_sections_before_writing_non_ingress_files(self):
+        with patch.object(scheduler, "topology_import_ingress_enabled", return_value=False):
+            with patch.object(
+                scheduler,
+                "overrides_materialized",
+                side_effect=RuntimeError("locked by another process: pid 42"),
+            ):
+                with patch.object(scheduler, "OVERRIDE_SECTION_READ_ATTEMPTS", 2):
+                    with patch.object(scheduler.time, "sleep"):
+                        with patch.object(scheduler, "read_shaped_devices_csv") as mock_read_sd:
+                            with patch.object(scheduler, "write_shaped_devices_csv") as mock_write_sd:
+                                with patch.object(scheduler, "load_network_json") as mock_load_network:
+                                    with patch.object(scheduler, "write_network_json") as mock_write_network:
+                                        with self.assertRaises(scheduler.RequiredOverrideReadError):
+                                            scheduler.apply_lqos_overrides()
+
+        mock_read_sd.assert_not_called()
+        mock_write_sd.assert_not_called()
+        mock_load_network.assert_not_called()
+        mock_write_network.assert_not_called()
+
+    def test_apply_lqos_overrides_uses_single_materialized_snapshot(self):
+        header = [
+            "Circuit ID", "Circuit Name", "Device ID", "Device Name", "Parent Node", "MAC",
+            "IPv4", "IPv6", "Download Min Mbps", "Upload Min Mbps", "Download Max Mbps",
+            "Upload Max Mbps", "Comment",
+        ]
+        rows = []
+        materialized = {
+            "persistent_devices": [],
+            "circuit_adjustments": [],
+            "network_adjustments": [],
+        }
+
+        with patch.object(scheduler, "topology_import_ingress_enabled", return_value=False):
+            with patch.object(scheduler, "overrides_materialized", return_value=materialized) as mock_materialized:
+                with patch.object(
+                    scheduler,
+                    "overrides_persistent_devices_materialized",
+                    side_effect=AssertionError("legacy persistent reader should not be used"),
+                    create=True,
+                ) as mock_persistent:
+                    with patch.object(
+                        scheduler,
+                        "overrides_circuit_adjustments_materialized",
+                        side_effect=AssertionError("legacy circuit reader should not be used"),
+                        create=True,
+                    ) as mock_circuit:
+                        with patch.object(
+                            scheduler,
+                            "overrides_network_adjustments_materialized",
+                            side_effect=AssertionError("legacy network reader should not be used"),
+                        ) as mock_network:
+                            with patch.object(scheduler, "read_shaped_devices_csv", return_value=(header, rows)):
+                                with patch.object(scheduler, "load_network_json", return_value={}):
+                                    with patch.object(scheduler, "load_topology_canonical_state", return_value=None):
+                                        scheduler.apply_lqos_overrides()
+
+        mock_materialized.assert_called_once_with()
+        mock_persistent.assert_not_called()
+        mock_circuit.assert_not_called()
+        mock_network.assert_not_called()
+
+    def test_import_from_crm_does_not_continue_after_override_failure(self):
+        with patch.object(
+            scheduler,
+            "apply_lqos_overrides",
+            side_effect=scheduler.RequiredOverrideReadError("locked by pid 42"),
+        ):
+            with patch.object(scheduler.os.path, "isfile", return_value=False):
+                with patch.object(scheduler, "scheduler_error") as mock_scheduler_error:
+                    with patch.object(scheduler, "scheduler_output") as mock_scheduler_output:
+                        with patch.object(scheduler, "publish_scheduler_progress") as mock_progress:
+                            with patch("builtins.print"):
+                                with self.assertRaises(scheduler.RequiredOverrideReadError):
+                                    scheduler.importFromCRM()
+
+        self.assertIn(
+            "preserving last-known-good topology",
+            mock_scheduler_error.call_args.args[0],
+        )
+        self.assertEqual(mock_scheduler_output.call_args.args, mock_scheduler_error.call_args.args)
+        self.assertTrue(
+            any(
+                call.args[:3] == (
+                    False,
+                    "degraded",
+                    "Scheduler blocked by override failure",
+                )
+                for call in mock_progress.call_args_list
+            )
+        )
+
     def test_apply_lqos_overrides_device_adjust_sqm_only_updates_sqm_column(self):
         header = [
             "Circuit ID", "Circuit Name", "Device ID", "Device Name", "Parent Node", "MAC",
@@ -1249,18 +1411,21 @@ class TestSchedulerOverrideMerge(unittest.TestCase):
         # Test-only fake csv path.
         with patch.object(scheduler, "shaped_devices_csv_path", return_value="/tmp/ShapedDevices.csv"):  # nosec B108
             with patch.object(scheduler, "read_shaped_devices_csv", return_value=(header, rows)):
-                with patch.object(scheduler, "overrides_persistent_devices_materialized", return_value=[]):
-                    with patch.object(
-                        scheduler,
-                        "overrides_circuit_adjustments_materialized",
-                        return_value=[{
+                with patch.object(
+                    scheduler,
+                    "overrides_materialized",
+                    return_value={
+                        "persistent_devices": [],
+                        "circuit_adjustments": [{
                             "type": "device_adjust_sqm",
                             "device_id": "splynx_service_93",
                             "sqm_override": "fq_codel/fq_codel",
                         }],
-                    ):
-                        with patch.object(scheduler, "write_shaped_devices_csv") as mock_write:
-                            scheduler.apply_lqos_overrides()
+                        "network_adjustments": [],
+                    },
+                ):
+                    with patch.object(scheduler, "write_shaped_devices_csv") as mock_write:
+                        scheduler.apply_lqos_overrides()
 
         written_rows = mock_write.call_args.args[2]
         self.assertEqual(written_rows[0][10], "330")
@@ -1281,18 +1446,21 @@ class TestSchedulerOverrideMerge(unittest.TestCase):
 
         with patch.object(scheduler, "shaped_devices_csv_path", return_value="/tmp/ShapedDevices.csv"):  # nosec B108
             with patch.object(scheduler, "read_shaped_devices_csv", return_value=(header, rows)):
-                with patch.object(scheduler, "overrides_persistent_devices_materialized", return_value=[]):
-                    with patch.object(
-                        scheduler,
-                        "overrides_circuit_adjustments_materialized",
-                        return_value=[{
+                with patch.object(
+                    scheduler,
+                    "overrides_materialized",
+                    return_value={
+                        "persistent_devices": [],
+                        "circuit_adjustments": [{
                             "type": "reparent_circuit",
                             "circuit_id": "93",
                             "parent_node": "AP-Updated",
                         }],
-                    ):
-                        with patch.object(scheduler, "write_shaped_devices_csv") as mock_write:
-                            scheduler.apply_lqos_overrides()
+                        "network_adjustments": [],
+                    },
+                ):
+                    with patch.object(scheduler, "write_shaped_devices_csv") as mock_write:
+                        scheduler.apply_lqos_overrides()
 
         written_rows = mock_write.call_args.args[2]
         self.assertEqual(written_rows[0][4], "AP-Updated")
@@ -1330,26 +1498,28 @@ class TestSchedulerOverrideMerge(unittest.TestCase):
 
         with patch.object(scheduler, "shaped_devices_csv_path", return_value="/tmp/ShapedDevices.csv"):  # nosec B108
             with patch.object(scheduler, "read_shaped_devices_csv", return_value=(header, rows)):
-                with patch.object(scheduler, "overrides_persistent_devices_materialized", return_value=[]):
-                    with patch.object(scheduler, "overrides_circuit_adjustments_materialized", return_value=[]):
-                        with patch.object(
-                            scheduler,
-                            "overrides_network_adjustments_materialized",
-                            return_value=[{
+                with patch.object(
+                    scheduler,
+                    "overrides_materialized",
+                    return_value={
+                        "persistent_devices": [],
+                        "circuit_adjustments": [],
+                        "network_adjustments": [{
                                 "type": "adjust_site_speed",
                                 "node_id": "node-b",
                                 "site_name": "NodeB",
                                 "download_bandwidth_mbps": 80,
                                 "upload_bandwidth_mbps": 40,
                             }],
-                        ):
-                            with patch.object(scheduler, "topology_import_ingress_enabled", return_value=True):
-                                with patch.object(scheduler, "load_topology_canonical_state", return_value=canonical_state):
-                                    with patch.object(scheduler, "write_topology_canonical_state") as mock_write_canonical:
-                                        with patch.object(scheduler, "load_network_json") as mock_load_network:
-                                            with patch.object(scheduler, "write_network_json") as mock_write_network:
-                                                with patch.object(scheduler, "write_shaped_devices_csv") as mock_write_sd:
-                                                    scheduler.apply_lqos_overrides()
+                    },
+                ):
+                    with patch.object(scheduler, "topology_import_ingress_enabled", return_value=True):
+                        with patch.object(scheduler, "load_topology_canonical_state", return_value=canonical_state):
+                            with patch.object(scheduler, "write_topology_canonical_state") as mock_write_canonical:
+                                with patch.object(scheduler, "load_network_json") as mock_load_network:
+                                    with patch.object(scheduler, "write_network_json") as mock_write_network:
+                                        with patch.object(scheduler, "write_shaped_devices_csv") as mock_write_sd:
+                                            scheduler.apply_lqos_overrides()
 
         mock_load_network.assert_not_called()
         mock_write_network.assert_not_called()