`33bdd6d`

add restart recovery store

Authored by mfwolffe <wolffemf@dukes.jmu.edu> 2 months ago

SHA: 33bdd6dcb6dea23d9c71572f401b6eb4a660dac7
Parents: 1e77d9b
Tree: b9da39d

7 changed files

Status	File	+	-
M	`garwarp/src/config.rs`	12	1
M	`garwarp/src/daemon.rs`	90	3
M	`garwarp/src/main.rs`	1	0
M	`garwarp/src/request.rs`	64	0
A	`garwarp/src/request_store.rs`	179	0
M	`garwarp/src/runtime.rs`	7	0
M	`garwarp/src/window.rs`	9	0

garwarp/src/config.rsmodified

  #[derive(Debug, Clone)]
  pub struct Config {
      pub poll_interval: Duration,
 +    pub request_timeout: Duration,
+ }
  impl Config {
      #[must_use]
      pub fn from_env() -> Self {
          const DEFAULT_POLL_MS: u64 = 100;
 +        const DEFAULT_REQUEST_TIMEOUT_MS: u64 = 30_000;
          let poll_interval = env::var("GARWARP_POLL_MS")
              .ok()
              .and_then(|value| value.parse::<u64>().ok())
              .filter(|value| *value > 0)
              .map(Duration::from_millis)
              .unwrap_or_else(|| Duration::from_millis(DEFAULT_POLL_MS));
 -        Self { poll_interval }
 +        let request_timeout = env::var("GARWARP_REQUEST_TIMEOUT_MS")
 +            .ok()
 +            .and_then(|value| value.parse::<u64>().ok())
 +            .filter(|value| *value > 0)
 +            .map(Duration::from_millis)
 +            .unwrap_or_else(|| Duration::from_millis(DEFAULT_REQUEST_TIMEOUT_MS));
 +        Self {
 +            poll_interval,
 +            request_timeout,
 +        }
+     }
+ }

garwarp/src/daemon.rsmodified

  use crate::lock::SingleInstanceGuard;
  use crate::logging;
  use crate::request::{RequestOwner, RequestRegistry, RequestState};
 +use crate::request_store;
  use crate::runtime::RuntimePaths;
  use crate::window::parse_optional_parent_window;
      logging::info("daemon_starting");
 +    let (requests, recovered_ids) =
 +        load_registry_with_recovery(&paths.request_store, config.request_timeout)?;
 +    if !recovered_ids.is_empty() {
 +        logging::warn(&format!(
 +            "request_recovery_expired count={}",
 +            recovered_ids.len()
 +        ));
 +    }
++
      let mut state = DaemonState {
          health: HealthStatus::Healthy,
 -        requests: RequestRegistry::new(Duration::from_secs(30)),
 +        requests,
          running: true,
      };
 +    persist_registry_state(&paths.request_store, &state.requests);
      while state.running {
          let expired = state.requests.expire_stale(Instant::now());
 -        for id in expired {
 +        for id in &expired {
              logging::warn(&format!("request_expired id={id}"));
+         }
 +        if !expired.is_empty() {
 +            persist_registry_state(&paths.request_store, &state.requests);
 +        }
          match listener.accept() {
              Ok((stream, _address)) => {
                  if let Err(error) = handle_connection(stream, &mut state) {
                      logging::warn(&format!("request_error={error}"));
 +                } else {
 +                    persist_registry_state(&paths.request_store, &state.requests);
+                 }
+             }
              Err(error) if error.kind() == io::ErrorKind::WouldBlock => {
+     }
+ }
 +fn load_registry_with_recovery(
 +    request_store_path: &std::path::Path,
 +    timeout: Duration,
 +) -> io::Result<(RequestRegistry, Vec<String>)> {
 +    let mut registry = request_store::load_registry(request_store_path, timeout)?;
 +    let expired = registry.recover_after_restart(Instant::now());
 +    Ok((registry, expired))
 +}
++
 +fn persist_registry_state(path: &std::path::Path, registry: &RequestRegistry) {
 +    if let Err(error) = request_store::persist_registry(path, registry) {
 +        logging::warn(&format!("request_store_write_failed error={error}"));
 +    }
 +}
++
  #[cfg(test)]
  mod tests {
 -    use super::{DaemonState, handle_connection};
 +    use super::{DaemonState, handle_connection, load_registry_with_recovery};
      use garwarp_ipc::{ControlResponse, HealthStatus};
 +    use std::fs;
      use std::io::{BufRead, BufReader, Write};
      use std::os::unix::net::UnixStream;
 +    use std::path::PathBuf;
      use std::time::{Duration, Instant};
      use crate::request::{RequestOwner, RequestRegistry, RequestState};
 +    use crate::request_store;
      use crate::window::ParentWindowContext;
 +    fn unique_temp_file() -> PathBuf {
 +        let nanos = std::time::SystemTime::now()
 +            .duration_since(std::time::UNIX_EPOCH)
 +            .map_or(0, |duration| duration.as_nanos());
 +        std::env::temp_dir().join(format!("garwarp-daemon-recovery-{nanos}.state"))
 +    }
++
      #[test]
      fn status_request_returns_status_response() {
          let (mut client, server) = UnixStream::pair().expect("pair should be created");
          );
          assert_eq!(state.requests.state("req-1"), Some(RequestState::Pending));
+     }
++
 +    #[test]
 +    fn startup_recovery_expires_non_terminal_requests() {
 +        let path = unique_temp_file();
++
 +        let mut persisted = RequestRegistry::new(Duration::from_secs(5));
 +        persisted
 +            .begin_at(
 +                "req-pending",
 +                RequestOwner::new(":1.2", None),
 +                None,
 +                Instant::now(),
 +            )
 +            .expect("request should be created");
 +        persisted
 +            .begin_at(
 +                "req-done",
 +                RequestOwner::new(":1.3", None),
 +                None,
 +                Instant::now(),
 +            )
 +            .expect("request should be created");
 +        persisted
 +            .transition(
 +                "req-done",
 +                &RequestOwner::new(":1.3", None),
 +                RequestState::AwaitingUser,
 +            )
 +            .expect("request should transition");
 +        persisted
 +            .transition(
 +                "req-done",
 +                &RequestOwner::new(":1.3", None),
 +                RequestState::Fulfilled,
 +            )
 +            .expect("request should transition");
 +        request_store::persist_registry(&path, &persisted).expect("request store should persist");
++
 +        let (loaded, recovered) = load_registry_with_recovery(&path, Duration::from_secs(5))
 +            .expect("registry should load");
 +        assert_eq!(recovered, vec!["req-pending".to_string()]);
 +        assert_eq!(loaded.state("req-pending"), Some(RequestState::Expired));
 +        assert_eq!(loaded.state("req-done"), Some(RequestState::Fulfilled));
++
 +        let _ = fs::remove_file(path);
 +    }
+ }

garwarp/src/main.rsmodified

  mod lock;
  mod logging;
  mod request;
 +mod request_store;
  mod runtime;
  mod window;

garwarp/src/request.rsmodified

              Self::Expired => "expired",
+         }
+     }
++
 +    #[must_use]
 +    pub fn parse(input: &str) -> Option<Self> {
 +        match input {
 +            "pending" => Some(Self::Pending),
 +            "awaiting_user" => Some(Self::AwaitingUser),
 +            "fulfilled" => Some(Self::Fulfilled),
 +            "cancelled" => Some(Self::Cancelled),
 +            "failed" => Some(Self::Failed),
 +            "expired" => Some(Self::Expired),
 +            _ => None,
 +        }
 +    }
+ }
  #[derive(Debug, Clone)]
+     }
+ }
 +#[derive(Debug, Clone, PartialEq, Eq)]
 +pub struct RequestRecord {
 +    pub id: String,
 +    pub owner: RequestOwner,
 +    pub parent_window: Option<ParentWindowContext>,
 +    pub state: RequestState,
 +}
++
  #[derive(Debug)]
  pub struct RequestRegistry {
      entries: HashMap<String, RequestEntry>,
          Ok(())
+     }
 +    pub fn restore_record(
 +        &mut self,
 +        record: RequestRecord,
 +        now: Instant,
 +    ) -> Result<(), RequestError> {
 +        if self.entries.contains_key(&record.id) {
 +            return Err(RequestError::AlreadyExists(record.id));
 +        }
 +        self.entries.insert(
 +            record.id.clone(),
 +            RequestEntry {
 +                id: record.id,
 +                owner: record.owner,
 +                parent_window: record.parent_window,
 +                state: record.state,
 +                started_at: now,
 +                last_updated_at: now,
 +            },
 +        );
 +        Ok(())
 +    }
++
      pub fn transition(
          &mut self,
          id: &str,
      pub fn parent_window(&self, id: &str) -> Option<Option<ParentWindowContext>> {
          self.entries.get(id).map(|entry| entry.parent_window)
+     }
++
 +    #[must_use]
 +    pub fn owner(&self, id: &str) -> Option<RequestOwner> {
 +        self.entries.get(id).map(|entry| entry.owner.clone())
 +    }
++
 +    #[must_use]
 +    pub fn records(&self) -> Vec<RequestRecord> {
 +        let mut records = self
 +            .entries
 +            .values()
 +            .map(|entry| RequestRecord {
 +                id: entry.id.clone(),
 +                owner: entry.owner.clone(),
 +                parent_window: entry.parent_window,
 +                state: entry.state,
 +            })
 +            .collect::<Vec<_>>();
 +        records.sort_by(|left, right| left.id.cmp(&right.id));
 +        records
 +    }
+ }
  fn is_valid_transition(from: RequestState, to: RequestState) -> bool {

garwarp/src/request_store.rsadded

 +#![allow(dead_code)]
++
 +use std::collections::HashMap;
 +use std::fs;
 +use std::io;
 +use std::path::Path;
 +use std::time::{Duration, Instant};
++
 +use crate::request::{RequestOwner, RequestRecord, RequestRegistry, RequestState};
 +use crate::window::parse_optional_parent_window;
++
 +pub fn load_registry(path: &Path, timeout: Duration) -> io::Result<RequestRegistry> {
 +    let mut registry = RequestRegistry::new(timeout);
 +    if !path.exists() {
 +        return Ok(registry);
 +    }
++
 +    let body = fs::read_to_string(path)?;
 +    let now = Instant::now();
++
 +    for (index, line) in body.lines().enumerate() {
 +        if line.trim().is_empty() {
 +            continue;
 +        }
 +        let record = parse_record_line(line).map_err(|error| {
 +            io::Error::new(
 +                io::ErrorKind::InvalidData,
 +                format!("invalid request store line {}: {error}", index + 1),
 +            )
 +        })?;
 +        registry.restore_record(record, now).map_err(|error| {
 +            io::Error::new(
 +                io::ErrorKind::InvalidData,
 +                format!("invalid request store line {}: {error}", index + 1),
 +            )
 +        })?;
 +    }
++
 +    Ok(registry)
 +}
++
 +pub fn persist_registry(path: &Path, registry: &RequestRegistry) -> io::Result<()> {
 +    let mut output = String::new();
 +    for record in registry.records() {
 +        output.push_str(&format_record_line(&record));
 +        output.push('\n');
 +    }
 +    fs::write(path, output)
 +}
++
 +fn format_record_line(record: &RequestRecord) -> String {
 +    let app_id = record.owner.app_id.as_deref().unwrap_or("-");
 +    let parent_window = match record.parent_window {
 +        Some(parent_window) => parent_window.as_str(),
 +        None => "-".to_string(),
 +    };
++
 +    format!(
 +        "id={}\tsender={}\tapp_id={}\tparent={}\tstate={}",
 +        record.id,
 +        record.owner.sender,
 +        app_id,
 +        parent_window,
 +        record.state.as_str()
 +    )
 +}
++
 +fn parse_record_line(line: &str) -> Result<RequestRecord, String> {
 +    let fields = parse_fields(line)?;
++
 +    let id = required_field(&fields, "id")?.to_string();
 +    let sender = required_field(&fields, "sender")?.to_string();
 +    let app_id = optional_field(&fields, "app_id");
 +    let parent_window = parse_optional_parent_window(optional_field_ref(&fields, "parent"))
 +        .map_err(|error| error.to_string())?;
 +    let state = RequestState::parse(required_field(&fields, "state")?)
 +        .ok_or_else(|| "invalid request state".to_string())?;
++
 +    Ok(RequestRecord {
 +        id,
 +        owner: RequestOwner::new(sender, app_id),
 +        parent_window,
 +        state,
 +    })
 +}
++
 +fn parse_fields(line: &str) -> Result<HashMap<&str, &str>, String> {
 +    let mut fields = HashMap::new();
 +    for token in line.split('\t') {
 +        let (key, value) = token
 +            .split_once('=')
 +            .ok_or_else(|| format!("invalid token: {token}"))?;
 +        fields.insert(key, value);
 +    }
 +    Ok(fields)
 +}
++
 +fn required_field<'a>(fields: &'a HashMap<&str, &'a str>, key: &str) -> Result<&'a str, String> {
 +    fields
 +        .get(key)
 +        .copied()
 +        .ok_or_else(|| format!("missing field: {key}"))
 +}
++
 +fn optional_field(fields: &HashMap<&str, &str>, key: &str) -> Option<String> {
 +    optional_field_ref(fields, key).map(ToOwned::to_owned)
 +}
++
 +fn optional_field_ref<'a>(fields: &'a HashMap<&str, &'a str>, key: &str) -> Option<&'a str> {
 +    fields
 +        .get(key)
 +        .copied()
 +        .and_then(|value| if value == "-" { None } else { Some(value) })
 +}
++
 +#[cfg(test)]
 +mod tests {
 +    use super::{load_registry, persist_registry};
 +    use crate::request::{RequestOwner, RequestRegistry, RequestState};
 +    use crate::window::ParentWindowContext;
 +    use std::fs;
 +    use std::path::PathBuf;
 +    use std::time::{Duration, Instant, SystemTime, UNIX_EPOCH};
++
 +    fn unique_temp_file() -> PathBuf {
 +        let nanos = SystemTime::now()
 +            .duration_since(UNIX_EPOCH)
 +            .map_or(0, |duration| duration.as_nanos());
 +        std::env::temp_dir().join(format!("garwarp-request-store-{nanos}.state"))
 +    }
++
 +    #[test]
 +    fn persist_and_load_roundtrip() {
 +        let path = unique_temp_file();
 +        let mut registry = RequestRegistry::new(Duration::from_secs(5));
 +        registry
 +            .begin_at(
 +                "req-1",
 +                RequestOwner::new(":1.2", Some("org.test.App".to_string())),
 +                Some(ParentWindowContext::X11 { window_id: 42 }),
 +                Instant::now(),
 +            )
 +            .expect("request should be created");
 +        registry
 +            .transition(
 +                "req-1",
 +                &RequestOwner::new(":1.2", Some("org.test.App".to_string())),
 +                RequestState::AwaitingUser,
 +            )
 +            .expect("request should transition");
++
 +        persist_registry(&path, &registry).expect("registry should persist");
++
 +        let loaded = load_registry(&path, Duration::from_secs(5)).expect("registry should load");
 +        assert_eq!(loaded.state("req-1"), Some(RequestState::AwaitingUser));
 +        assert_eq!(
 +            loaded.parent_window("req-1"),
 +            Some(Some(ParentWindowContext::X11 { window_id: 42 }))
 +        );
 +        assert_eq!(
 +            loaded.owner("req-1"),
 +            Some(RequestOwner::new(":1.2", Some("org.test.App".to_string())))
 +        );
++
 +        let _ = fs::remove_file(path);
 +    }
++
 +    #[test]
 +    fn invalid_lines_fail_to_load() {
 +        let path = unique_temp_file();
 +        fs::write(&path, "id=req-1\tsender=:1.2\tstate=bogus\n")
 +            .expect("test file should be written");
++
 +        let loaded = load_registry(&path, Duration::from_secs(5));
 +        assert!(loaded.is_err());
++
 +        let _ = fs::remove_file(path);
 +    }
 +}

garwarp/src/runtime.rsmodified

      pub root: PathBuf,
      pub control_socket: PathBuf,
      pub lock_file: PathBuf,
 +    pub request_store: PathBuf,
+ }
  impl RuntimePaths {
          let root = base.join(DEFAULT_RUNTIME_SUBDIR);
          let control_socket = root.join(DEFAULT_CONTROL_SOCKET);
          let lock_file = root.join("garwarp.lock");
 +        let request_store = root.join("requests.state");
          Self {
              root,
              control_socket,
              lock_file,
 +            request_store,
+         }
+     }
              paths.lock_file,
              PathBuf::from("/tmp/runtime/garwarp/garwarp.lock")
          );
 +        assert_eq!(
 +            paths.request_store,
 +            PathBuf::from("/tmp/runtime/garwarp/requests.state")
 +        );
+     }
+ }

garwarp/src/window.rsmodified

      X11 { window_id: u64 },
+ }
 +impl ParentWindowContext {
 +    #[must_use]
 +    pub fn as_str(self) -> String {
 +        match self {
 +            Self::X11 { window_id } => format!("x11:0x{window_id:x}"),
 +        }
 +    }
 +}
++
  #[derive(Debug, Clone, PartialEq, Eq)]
  pub enum ParentWindowError {
      Empty,