Refactor prefixed path

src/backends/mod.rs

@@ -22,7 +22,7 @@ use thiserror::Error;
 use crate::{
     config::{Config, ConfigError},
     instance::{Instance, InstanceClient, InstanceError},
-    util::Path,
+    path::PrefixedPath,
 };
 
 /// One virtio-scsi IOThread-to-virtqueue mapping.
@@ -159,7 +159,7 @@ pub struct BackendRegistration {
     /// Stable backend name used in logs and diagnostics.
     pub name: &'static str,
     /// Build the backend from `Config::backend_config_dir`.
-    pub build: fn(&Path) -> Result<Box<dyn Backend>, BackendClientError>,
+    pub build: fn(&PrefixedPath) -> Result<Box<dyn Backend>, BackendClientError>,
 }
 
 /// Linked-in backend factories. Feature-gated modules append themselves here.
@@ -180,7 +180,7 @@ pub trait Backend: Send + Sync {
     /// Inotify identifies the changed entry and event kind, but it does not
     /// describe the backend-level inventory delta. The daemon therefore uses
     /// any event only as a prompt to run a complete discovery pass.
-    fn watch_paths(&self) -> Vec<Path> {
+    fn watch_paths(&self) -> Vec<PrefixedPath> {
         Vec::new()
     }
 
@@ -224,12 +224,12 @@ pub fn registered_backends(cfg: &Config) -> Result<Vec<Box<dyn Backend>>, Backen
 /// supplied factory keeps the control protocol and concrete client
 /// backend-owned.
 pub async fn discover_via_control_socket<F>(
-    socket_dir: impl AsRef<Path>,
+    socket_dir: impl AsRef<PrefixedPath>,
     socket_name: &str,
     make_client: F,
 ) -> io::Result<Vec<Arc<Instance>>>
 where
-    F: Fn(Path) -> Box<dyn InstanceClient>,
+    F: Fn(PrefixedPath) -> Box<dyn InstanceClient>,
 {
     let socket_dir = socket_dir.as_ref();
     tracing::debug!("discovering in {}", socket_dir.display());
@@ -262,7 +262,8 @@ where
         let Ok(id) = entry.file_name().into_string() else {
             continue;
         };
-        let sock_path = Path::new(&socket_dir.join(entry.path().join(socket_name)));
+        // FIXME double prefix bug
+        let sock_path = PrefixedPath::new(&socket_dir.join(entry.path().join(socket_name)));
         let pid = match cached_control_socket_peer_pid(&sock_path).await {
             Ok(pid) => pid,
             Err(error) if error.kind() == io::ErrorKind::WouldBlock => {
@@ -320,7 +321,7 @@ static PEER_CACHE: LazyLock<Mutex<HashMap<PathBuf, CachedPeer>>> = LazyLock::new
 
 /// Return the control-socket peer PID, connecting only when the socket file
 /// or the previously identified peer process changed.
-async fn cached_control_socket_peer_pid(sock_path: &Path) -> io::Result<i32> {
+async fn cached_control_socket_peer_pid(sock_path: &PrefixedPath) -> io::Result<i32> {
     let key: &std::path::Path = sock_path;
     let metadata = tokio::fs::metadata(sock_path).await?;
     let cached = PEER_CACHE.lock().expect("lock poisoned").get(key).copied();
@@ -348,7 +349,7 @@ async fn cached_control_socket_peer_pid(sock_path: &Path) -> io::Result<i32> {
 
 /// Return the start time of `pid`, or `None` if it no longer exists.
 fn process_start_time(pid: i32) -> Option<u64> {
-    let path = Path::new(&format!("/proc/{pid}"));
+    let path = PrefixedPath::new(&format!("/proc/{pid}"));
     let path = std::path::PathBuf::from(std::path::Path::as_os_str(&path));
     procfs::process::Process::new_with_root(path)
         .and_then(|process| process.stat())
@@ -357,7 +358,7 @@ fn process_start_time(pid: i32) -> Option<u64> {
 }
 
 /// Return the process ID at the far end of a UNIX control socket.
-async fn control_socket_peer_pid(sock_path: &Path) -> io::Result<i32> {
+async fn control_socket_peer_pid(sock_path: &PrefixedPath) -> io::Result<i32> {
     let stream = tokio::net::UnixStream::connect(sock_path).await?;
     stream.peer_cred()?.pid().ok_or_else(|| {
         io::Error::new(
@@ -392,7 +393,7 @@ mod tests {
     }
 
     async fn discover(socket_dir: &std::path::Path) -> Vec<Arc<Instance>> {
-        discover_via_control_socket(Path::new(socket_dir), "ctl", |_| {
+        discover_via_control_socket(PrefixedPath::new(socket_dir), "ctl", |_| {
             Box::new(NullClient) as Box<dyn InstanceClient>
         })
         .await

src/backends/qemu/client.rs

@@ -26,6 +26,7 @@ use tokio::sync::Mutex;
 use uuid::Uuid;
 
 use crate::{
+    ProcessExt,
     backends::{
         BackendClientError, IoThreadProperties, VqMapping,
         qemu::{
@@ -36,7 +37,6 @@ use crate::{
         },
     },
     instance::{InstanceClient, ThreadPoolSnapshot},
-    util::ProcessExt,
 };
 
 /// Wire client for an QEMU-managed VM.

src/backends/qemu/helpers.rs

@@ -30,8 +30,8 @@ use rustix::param;
 use thiserror::Error;
 
 use crate::{
+    ProcessExt,
     backends::{BackendClientError, VqMapping},
-    util::ProcessExt,
 };
 
 /// Errors produced while querying or reconfiguring a QEMU vm.

src/backends/qemu/mod.rs

@@ -33,7 +33,7 @@ use crate::{
     backends::{Backend, BackendClientError, BackendRegistration},
     config::{self, ConfigError},
     instance::{Instance, InstanceClient},
-    util::Path,
+    path::PrefixedPath,
 };
 
 pub mod client;
@@ -42,7 +42,7 @@ pub mod libvirt;
 
 const BACKEND_NAME: &str = "qemu";
 
-fn build_qemu_backend(dir: &Path) -> Result<Box<dyn Backend>, BackendClientError> {
+fn build_qemu_backend(dir: &PrefixedPath) -> Result<Box<dyn Backend>, BackendClientError> {
     Ok(Box::new(QemuBackend::from_config_dir(dir)?))
 }
 
@@ -81,8 +81,8 @@ impl Default for QemuConfig {
 impl QemuConfig {
     /// Read `<dir>/qemu.json` if present, otherwise return the
     /// built-in defaults.
-    pub fn from_dir(dir: &Path) -> Result<Self, ConfigError> {
-        let path = Path::new(&dir.join("qemu.json"));
+    pub fn from_dir(dir: &PrefixedPath) -> Result<Self, ConfigError> {
+        let path = PrefixedPath::new(&dir.join("qemu.json"));
         config::load_or_default(&path)
     }
 }
@@ -109,7 +109,7 @@ impl QemuBackend {
 
     /// Convenience: read `<dir>/qemu.json` (or fall back to
     /// defaults) and construct.
-    pub fn from_config_dir(dir: &Path) -> Result<Self, ConfigError> {
+    pub fn from_config_dir(dir: &PrefixedPath) -> Result<Self, ConfigError> {
         Ok(Self::new(QemuConfig::from_dir(dir)?))
     }
 
@@ -207,7 +207,7 @@ impl Backend for QemuBackend {
             // `Instance` model; for the libvirt-backed backend
             // it is purely informational (log context), so we
             // stuff the libvirt URI in there.
-            let sock_path = Path::new(&self.cfg.libvirt_uri);
+            let sock_path = PrefixedPath::new(&self.cfg.libvirt_uri);
             out.push(Arc::new(Instance::new(
                 uuid.to_string(),
                 sock_path,

src/config.rs

@@ -10,7 +10,7 @@ use std::{io, time::Duration};
 use serde::{Deserialize, Deserializer, Serialize, Serializer};
 use thiserror::Error;
 
-use crate::util::Path;
+use crate::path::PrefixedPath;
 
 #[derive(Error, Debug)]
 pub enum ConfigError {
@@ -39,14 +39,14 @@ pub struct Config {
     pub engine: String,
     /// Directory to look for `<engine>.json` engine-specific
     /// config files.
-    pub engine_config_dir: Path,
+    pub engine_config_dir: PrefixedPath,
     /// Directory containing backend-owned configuration files.
-    pub backend_config_dir: Path,
+    pub backend_config_dir: PrefixedPath,
     /// JSON file containing the managed and unmanaged VM lists.
     ///
     /// The default is below `/run`, so state survives daemon restarts but not a
     /// host reboot.
-    pub vm_state_path: Path,
+    pub vm_state_path: PrefixedPath,
     /// Lowest worker count accepted from a scaling engine.
     #[serde(default = "default_min_thread_count")]
     pub min_thread_count: u32,
@@ -235,15 +235,15 @@ impl Default for Config {
     fn default() -> Self {
         Self {
             engine: "threshold".to_string(),
-            engine_config_dir: Path::new("/etc/io-thread-controller.d/engines"),
-            backend_config_dir: Path::new("/etc/io-thread-controller.d/backends"),
+            engine_config_dir: PrefixedPath::new("/etc/io-thread-controller.d/engines"),
+            backend_config_dir: PrefixedPath::new("/etc/io-thread-controller.d/backends"),
             scale_poll_secs: 10,
             round_up_durations: false,
             min_thread_count: 2,
             max_thread_count: 8,
             host_cpu_scale_up_ceiling: 0.9,
             cooldown_secs: default_cooldown_secs(),
-            vm_state_path: Path::new("/run/io-thread-controller/vm-ownership.json"),
+            vm_state_path: PrefixedPath::new("/run/io-thread-controller/vm-ownership.json"),
             enable_per_vm_status_line: false,
             enable_per_vm_main_log: false,
             enable_aggregate_status_line: true,
@@ -490,7 +490,7 @@ mod tests {
         )
         .unwrap();
 
-        let cfg: Config = load_or_default(Path::new(path.to_str().unwrap())).unwrap();
+        let cfg: Config = load_or_default(PrefixedPath::new(path.to_str().unwrap())).unwrap();
         assert_eq!(cfg.engine, "threshold");
         assert_eq!(cfg.scale_poll_secs, 7);
         assert_eq!(
@@ -517,7 +517,7 @@ mod tests {
         )
         .unwrap();
 
-        let err = load_or_default::<Config>(Path::new(path.to_str().unwrap())).unwrap_err();
+        let err = load_or_default::<Config>(PrefixedPath::new(path.to_str().unwrap())).unwrap_err();
         assert!(matches!(err, ConfigError::SerdeJson(_)));
     }
 

src/controller.rs

@@ -27,9 +27,9 @@ use crate::{
         AppliedOutcome, BlockedReason, EngineTickContext, PsiSample, ScaleAction, ScalingEngine,
     },
     instance::{CgroupError, Instance, InstanceStatus, SnapshotLatency},
+    path::PrefixedPath,
     per_vm_log,
     state::{StateError, VmOwnership, VmStateStore},
-    util::Path,
 };
 /// Wire-facing container for the D-Bus `GetSnapshot` reply.
 #[derive(Debug, Clone, Default, Serialize, Deserialize)]
@@ -1012,7 +1012,7 @@ fn read_psi_sample() -> PsiSample {
 /// Read aggregate host CPU counters.
 fn read_host_cpu_sample() -> Result<HostCpuSample, ControllerError> {
     let current_si = procfs::current_system_info();
-    let total = KernelStats::from_file(Path::new("/proc/stat"), current_si)?.total;
+    let total = KernelStats::from_file(PrefixedPath::new("/proc/stat"), current_si)?.total;
 
     let idle_ticks = total.idle.saturating_add(total.iowait.unwrap_or(0));
     let total_ticks = total
@@ -1086,7 +1086,6 @@ mod tests {
         engines::{EngineTickContext, ScalingEngine},
         instance::{InstanceClient, ThreadPoolSnapshot},
         per_vm_log::PerVmLog,
-        util::Path,
     };
     use tracing_subscriber::layer::SubscriberExt;
 
@@ -1147,14 +1146,14 @@ mod tests {
     async fn actuation_enforces_vcpu_cap() {
         let state_dir = tempfile::tempdir().unwrap();
         let cfg = Config {
-            vm_state_path: Path::new(&state_dir.path().join("ownership.json")),
+            vm_state_path: PrefixedPath::new(&state_dir.path().join("ownership.json")),
             cooldown_secs: 0,
             ..Default::default()
         };
         let client = Arc::new(VcpuLimitedClient::new(2));
         let instance = Arc::new(Instance::new(
             "some-vcpu-limited-test-instance".to_string(),
-            Path::new(""),
+            PrefixedPath::new(""),
             0,
             Arc::clone(&client),
         ));
@@ -1210,7 +1209,7 @@ mod tests {
     async fn actuation_enforces_controller_bounds_and_host_ceiling() {
         let state_dir = tempfile::tempdir().unwrap();
         let cfg = Config {
-            vm_state_path: Path::new(&state_dir.path().join("ownership.json")),
+            vm_state_path: PrefixedPath::new(&state_dir.path().join("ownership.json")),
             min_thread_count: 2,
             max_thread_count: 3,
             host_cpu_scale_up_ceiling: 0.5,
@@ -1220,7 +1219,7 @@ mod tests {
         let client = Arc::new(VcpuLimitedClient::new(2));
         let instance = Arc::new(Instance::new(
             "policy-guarded".to_string(),
-            Path::new(""),
+            PrefixedPath::new(""),
             0,
             Arc::clone(&client),
         ));
@@ -1280,7 +1279,7 @@ mod tests {
     async fn revert_arms_cooldown_blocking_immediate_retry() {
         let state_dir = tempfile::tempdir().unwrap();
         let cfg = Config {
-            vm_state_path: Path::new(&state_dir.path().join("ownership.json")),
+            vm_state_path: PrefixedPath::new(&state_dir.path().join("ownership.json")),
             cooldown_secs: 30,
             min_thread_count: 1,
             max_thread_count: 8,
@@ -1289,7 +1288,7 @@ mod tests {
         let client = Arc::new(VcpuLimitedClient::new(4));
         let instance = Arc::new(Instance::new(
             "revert-cooldown".to_string(),
-            Path::new(""),
+            PrefixedPath::new(""),
             0,
             Arc::clone(&client),
         ));
@@ -1337,7 +1336,7 @@ mod tests {
     async fn revert_bypasses_existing_cooldown() {
         let state_dir = tempfile::tempdir().unwrap();
         let cfg = Config {
-            vm_state_path: Path::new(&state_dir.path().join("ownership.json")),
+            vm_state_path: PrefixedPath::new(&state_dir.path().join("ownership.json")),
             cooldown_secs: 60,
             min_thread_count: 1,
             max_thread_count: 8,
@@ -1346,7 +1345,7 @@ mod tests {
         let client = Arc::new(VcpuLimitedClient::new(3));
         let instance = Arc::new(Instance::new(
             "revert-bypass".to_string(),
-            Path::new(""),
+            PrefixedPath::new(""),
             0,
             Arc::clone(&client),
         ));
@@ -1414,7 +1413,7 @@ mod tests {
     async fn test_set_all_attempts_every_vm_and_reports_partial_failure() {
         let state_dir = tempfile::tempdir().unwrap();
         let cfg = Config {
-            vm_state_path: Path::new(&state_dir.path().join("ownership.json")),
+            vm_state_path: PrefixedPath::new(&state_dir.path().join("ownership.json")),
             ..Default::default()
         };
         let engine = Box::new(default_threshold_engine());
@@ -1429,8 +1428,18 @@ mod tests {
             fail: true,
             target: Arc::clone(&bad_target),
         };
-        let good_instance = Arc::new(Instance::new("visible".to_string(), Path::new(""), 0, good));
-        let bad_instance = Arc::new(Instance::new("hidden".to_string(), Path::new(""), 0, bad));
+        let good_instance = Arc::new(Instance::new(
+            "visible".to_string(),
+            PrefixedPath::new(""),
+            0,
+            good,
+        ));
+        let bad_instance = Arc::new(Instance::new(
+            "hidden".to_string(),
+            PrefixedPath::new(""),
+            0,
+            bad,
+        ));
         good_instance.status.write().await.vcpu_count = 16;
         bad_instance.status.write().await.vcpu_count = 16;
         controller
@@ -1529,7 +1538,7 @@ mod tests {
     fn instance(id: &str, threads: u32, closed: Arc<AtomicUsize>) -> Arc<Instance> {
         Arc::new(Instance::new(
             id.to_string(),
-            Path::new(""),
+            PrefixedPath::new(""),
             1,
             SnapshotClient { threads, closed },
         ))
@@ -1564,7 +1573,7 @@ mod tests {
         let evaluated = Arc::new(AtomicUsize::new(0));
         let state_dir = tempfile::tempdir().unwrap();
         let cfg = Config {
-            vm_state_path: Path::new(&state_dir.path().join("ownership.json")),
+            vm_state_path: PrefixedPath::new(&state_dir.path().join("ownership.json")),
             ..Default::default()
         };
         let mut controller = Controller::new(
@@ -1613,7 +1622,7 @@ mod tests {
     ) {
         let dir = tempfile::tempdir().unwrap();
         let per_vm_log = PerVmLog::default();
-        per_vm_log.set_dir(Some(Path::new(dir.path())));
+        per_vm_log.set_dir(Some(PrefixedPath::new(dir.path())));
         let guard = tracing::subscriber::set_default(
             tracing_subscriber::registry().with(per_vm_log.clone()),
         );
@@ -1641,7 +1650,7 @@ mod tests {
         let evaluated = Arc::new(AtomicUsize::new(0));
         let state_dir = tempfile::tempdir().unwrap();
         let cfg = Config {
-            vm_state_path: Path::new(&state_dir.path().join("ownership.json")),
+            vm_state_path: PrefixedPath::new(&state_dir.path().join("ownership.json")),
             ..Default::default()
         };
         let mut controller = Controller::new(
@@ -1659,7 +1668,7 @@ mod tests {
         let ok = instance("ok", 4, Arc::clone(&closed));
         let bad = Arc::new(Instance::new(
             "bad".to_string(),
-            Path::new(""),
+            PrefixedPath::new(""),
             2,
             FailingClient,
         ));