feat(state): expire stale agent devices with configurable retention

wakey-control-plane/src/config/init.rs

@@ -103,6 +103,8 @@ pub fn write_init_config(args: &InitConfigArgs) -> Result<Option<PathBuf>> {
         .or(base.history_retention)
         .unwrap_or(defaults.history_retention);
     super::resolve::validate_durations(command_timeout, enroll_token_ttl, history_retention)?;
+    let device_retention = base.device_retention.unwrap_or(defaults.device_retention);
+    super::resolve::validate_device_retention(device_retention)?;
 
     let body = FileConfig {
         data_dir: Some(data_dir),
@@ -112,6 +114,7 @@ pub fn write_init_config(args: &InitConfigArgs) -> Result<Option<PathBuf>> {
         command_timeout: Some(command_timeout),
         enroll_token_ttl: Some(enroll_token_ttl),
         history_retention: Some(history_retention),
+        device_retention: Some(device_retention),
         pid_file: Some(pid_file),
         ui_dist_dir: Some(ui_dist_dir),
         // Preserve the previous generated template's omission of an empty list.
@@ -237,6 +240,7 @@ public_url = "https://old.example/"
 state_file = "old.sqlite3"
 command_timeout_ms = 1234
 enroll_token_ttl_seconds = 5678
+device_retention = "0d"
 pid_file = "old.pid"
 ui_dist_dir = "/opt/wakey/ui"
 bootstrap_enroll_tokens = ["old-token"]
@@ -288,6 +292,7 @@ json_logs = true
             Some(std::time::Duration::from_secs(5678))
         );
         assert!(parsed.get("command_timeout_ms").is_none());
+        assert_eq!(converted.device_retention, Some(std::time::Duration::ZERO));
         assert!(parsed.get("enroll_token_ttl_seconds").is_none());
         assert_eq!(
             parsed["state_file"].as_str(),
@@ -349,6 +354,7 @@ json_logs = true
         );
         let converted: super::FileConfig = toml::from_str(&rendered).unwrap();
         assert_eq!(converted.command_timeout, Some(defaults.command_timeout));
+        assert_eq!(converted.device_retention, Some(defaults.device_retention));
         assert_eq!(
             parsed["telemetry"]["service_name"].as_str(),
             Some(defaults.telemetry.service_name.as_str())

wakey-control-plane/src/config/resolve.rs

@@ -67,6 +67,8 @@ impl DaemonConfig {
             .or(file.history_retention)
             .unwrap_or(defaults.history_retention);
         validate_durations(command_timeout, enroll_token_ttl, history_retention)?;
+        let device_retention = file.device_retention.unwrap_or(defaults.device_retention);
+        validate_device_retention(device_retention)?;
 
         let pid_file_raw = args
             .pid_file
@@ -98,6 +100,7 @@ impl DaemonConfig {
             command_timeout,
             enroll_token_ttl,
             history_retention,
+            device_retention,
             pid_file,
             ui_dist_dir,
             bootstrap_enroll_tokens,
@@ -223,6 +226,20 @@ pub(crate) fn load_file_config(path: &Path) -> Result<FileConfig> {
     Ok(config)
 }
 
+/// Device expiry uses whole-second snapshot timestamps, not an active timer.
+/// Unlike history retention, zero is a supported opt-out.
+pub(crate) fn validate_device_retention(retention: Duration) -> Result<()> {
+    anyhow::ensure!(
+        retention.subsec_nanos() == 0,
+        "device_retention must use whole seconds"
+    );
+    anyhow::ensure!(
+        i64::try_from(retention.as_secs()).is_ok(),
+        "device_retention is too large"
+    );
+    Ok(())
+}
+
 pub(crate) fn validate_durations(
     command_timeout: Duration,
     enroll_token_ttl: Duration,
@@ -327,11 +344,12 @@ fn resolve_state_access(
 
 #[cfg(test)]
 mod tests {
-    use super::{DaemonConfig, normalize_public_url, resolve_path};
+    use super::{DaemonConfig, normalize_public_url, resolve_path, validate_device_retention};
     use crate::cli::ServeArgs;
     use crate::config::types::DEFAULT_CONFIG;
     use std::path::Path;
     use std::path::PathBuf;
+    use std::time::Duration;
 
     #[test]
     fn normalize_public_url_trims_trailing_slash() {
@@ -383,12 +401,21 @@ mod tests {
             defaults.data_dir.join(&defaults.state_file)
         );
         assert_eq!(resolved.command_timeout, defaults.command_timeout);
+        assert_eq!(resolved.device_retention, defaults.device_retention);
         assert_eq!(
             resolved.telemetry.service_name,
             defaults.telemetry.service_name
         );
     }
 
+    #[test]
+    fn device_retention_accepts_disabled_and_whole_seconds_only() {
+        assert!(validate_device_retention(Duration::ZERO).is_ok());
+        assert!(validate_device_retention(DEFAULT_CONFIG.device_retention).is_ok());
+        assert!(validate_device_retention(Duration::from_millis(1_500)).is_err());
+        assert!(validate_device_retention(Duration::from_secs(u64::MAX)).is_err());
+    }
+
     #[test]
     fn invalid_duration_policies_are_rejected_without_clamping() {
         use std::time::Duration;

wakey-control-plane/src/config/types.rs

@@ -16,6 +16,8 @@ pub struct DaemonConfig {
     pub command_timeout: Duration,
     pub enroll_token_ttl: Duration,
     pub history_retention: Duration,
+    /// Maximum age since an agent snapshot included a device; zero disables expiry.
+    pub device_retention: Duration,
     pub pid_file: PathBuf,
     pub ui_dist_dir: PathBuf,
     pub bootstrap_enroll_tokens: Vec<String>,
@@ -37,6 +39,8 @@ pub(crate) struct FileConfig {
     pub(crate) enroll_token_ttl: Option<Duration>,
     #[serde(default, with = "humantime_serde::option")]
     pub(crate) history_retention: Option<Duration>,
+    #[serde(default, with = "humantime_serde::option")]
+    pub(crate) device_retention: Option<Duration>,
     pub(crate) pid_file: Option<PathBuf>,
     pub(crate) ui_dist_dir: Option<PathBuf>,
     #[serde(alias = "enroll_tokens")]
@@ -95,6 +99,7 @@ pub(crate) static DEFAULT_CONFIG: LazyLock<DaemonConfig> = LazyLock::new(|| Daem
     command_timeout: Duration::from_millis(30_000),
     enroll_token_ttl: Duration::from_secs(86_400),
     history_retention: Duration::from_secs(2_592_000),
+    device_retention: Duration::from_secs(60 * 86_400),
     pid_file: "wakey-control-plane.pid".into(),
     ui_dist_dir: "ui/dist".into(),
     bootstrap_enroll_tokens: Vec::new(),
@@ -186,4 +191,20 @@ mod tests {
             assert!(toml::from_str::<FileConfig>(&negative).is_err());
         }
     }
+
+    #[test]
+    fn device_retention_round_trips_and_requires_units() {
+        for text in ["60d", "0d"] {
+            let file: FileConfig = toml::from_str(&format!("device_retention = '{text}'")).unwrap();
+            let output = toml::to_string(&file).unwrap();
+            assert_eq!(toml::from_str::<FileConfig>(&output).unwrap(), file);
+        }
+        for input in [
+            "device_retention = 60",
+            "device_retention = 1.5",
+            "device_retention = '-1d'",
+        ] {
+            assert!(toml::from_str::<FileConfig>(input).is_err());
+        }
+    }
 }

wakey-control-plane/src/runtime/mod.rs

@@ -260,6 +260,11 @@ pub async fn serve(daemon: config::DaemonConfig) -> Result<()> {
                         }
                         Err(err) => warn!(error = %err, "periodic retention gc failed"),
                     }
+                    match app_state.store.prune_devices(daemon.device_retention).await {
+                        Ok(removed) if removed > 0 => info!(removed, "periodic gc pruned expired agent devices"),
+                        Ok(_) => {},
+                        Err(err) => warn!(error = %err, "periodic device retention gc failed"),
+                    }
                 }
                 join = &mut server => {
                     let _ = remove_pid_file(&daemon.pid_file);

wakey-control-plane/src/state/store/agent_devices.rs

@@ -6,6 +6,35 @@ use super::helpers::core::*;
 use crate::state::types::*;
 
 impl Store {
+    /// Expire agent-reported devices, preserving all operator-saved identities.
+    /// `last_seen_unix` on the parent is CC snapshot inclusion time, not proof
+    /// of online presence. Cascades remove child addresses, endpoints and facts.
+    #[tracing::instrument(name = "db_prune_devices", skip_all)]
+    pub async fn prune_devices(&self, retention: std::time::Duration) -> Result<u64> {
+        self.prune_devices_at(retention, now_unix()).await
+    }
+
+    async fn prune_devices_at(&self, retention: std::time::Duration, now: u64) -> Result<u64> {
+        if retention.is_zero() {
+            return Ok(0);
+        }
+        let cutoff = i64::try_from(now.saturating_sub(retention.as_secs()))
+            .context("device retention cutoff overflow")?;
+        let mut tx = self.begin_write().await?;
+        let removed = sqlx::query!(
+            "DELETE FROM agent_devices WHERE last_seen_unix < ?1",
+            cutoff
+        )
+        .execute(&mut *tx)
+        .await
+        .context("failed pruning expired agent devices")?
+        .rows_affected();
+        tx.commit()
+            .await
+            .context("failed committing device retention cleanup")?;
+        Ok(removed)
+    }
+
     /// Replace the complete device snapshot for an agent.
     ///
     /// This is the only write path for agent device state. Every sync source
@@ -566,6 +595,151 @@ mod tests {
         }
     }
 
+    #[tokio::test]
+    async fn device_retention_uses_snapshot_age_and_preserves_saved_identity() {
+        use crate::state::types::{DeviceIdentifierInput, KnownDeviceInput};
+
+        let ts = TestStore::new().await;
+        let store = ts.store();
+        let mut device = sample_device("aa:bb:cc:dd:ee:01", "192.168.1.10", "pc");
+        device.neighbors[0].state = NeighborState::Failed;
+        device.observations.push(DeviceObservationFact {
+            kind: "neigh".into(),
+            action: "remove".into(),
+            mac: device.macs.first().copied(),
+            ip: device.ips.first().copied(),
+            hostname: None,
+            first_seen_unix: Some(1),
+            last_seen_unix: Some(1),
+        });
+        let saved = store
+            .create_known_device(KnownDeviceInput {
+                display_name: "saved pc".into(),
+                pinned: true,
+                notes: None,
+                identifiers: vec![DeviceIdentifierInput {
+                    kind: "mac".into(),
+                    value: "aa:bb:cc:dd:ee:01".into(),
+                }],
+            })
+            .await
+            .unwrap();
+
+        // Explicit timestamps exercise the exact boundary and future rows
+        // without sleeping or relying on the test machine's wall clock.
+        for (agent, seen) in [("expired", 99_i64), ("boundary", 100), ("fresh", 101)] {
+            store
+                .replace_agent_device_snapshot(agent, &[device.clone()])
+                .await
+                .unwrap();
+            sqlx::query("UPDATE agent_devices SET last_seen_unix = ?1 WHERE agent_id = ?2")
+                .bind(seen)
+                .bind(agent)
+                .execute(&store.pool)
+                .await
+                .unwrap();
+            sqlx::query("UPDATE agent_device_endpoints SET last_seen_unix = 1 WHERE agent_id = ?1")
+                .bind(agent)
+                .execute(&store.pool)
+                .await
+                .unwrap();
+        }
+        assert_eq!(
+            store.prune_devices_at(Duration::ZERO, 1_000).await.unwrap(),
+            0
+        );
+        assert_eq!(
+            store
+                .prune_devices_at(Duration::from_secs(900), 1_000)
+                .await
+                .unwrap(),
+            1
+        );
+        assert_eq!(
+            store
+                .prune_devices_at(Duration::from_secs(900), 1_000)
+                .await
+                .unwrap(),
+            0
+        );
+
+        for table in [
+            "agent_devices",
+            "agent_device_macs",
+            "agent_device_ips",
+            "agent_device_hostnames",
+            "agent_device_endpoints",
+            "agent_device_facts",
+        ] {
+            let count: i64 = sqlx::query_scalar(&format!(
+                "SELECT COUNT(*) FROM {table} WHERE agent_id = 'expired'"
+            ))
+            .fetch_one(&store.pool)
+            .await
+            .unwrap();
+            assert_eq!(count, 0, "expired rows remain in {table}");
+        }
+        for agent in ["boundary", "fresh"] {
+            let rows = store.list_agent_device_rows_for_agent(agent).await.unwrap();
+            assert_eq!(rows.len(), 1);
+            assert_eq!(rows[0].presence(), Presence::Offline);
+        }
+        let retained = store
+            .get_known_device(&saved.device_id)
+            .await
+            .unwrap()
+            .unwrap();
+        assert_eq!(retained.identifiers.len(), 1);
+        assert!(retained.pinned);
+        // Retention longer than the current epoch must not wrap the cutoff.
+        assert_eq!(
+            store
+                .prune_devices_at(Duration::from_secs(2_000), 1_000)
+                .await
+                .unwrap(),
+            0
+        );
+    }
+
+    #[tokio::test]
+    async fn a_new_snapshot_renews_device_retention_even_when_offline() {
+        let ts = TestStore::new().await;
+        let store = ts.store();
+        let mut device = sample_device("aa:bb:cc:dd:ee:01", "192.168.1.10", "pc");
+        device.neighbors[0].state = NeighborState::Failed;
+        store
+            .replace_agent_device_snapshot("agent-a", &[device.clone()])
+            .await
+            .unwrap();
+        sqlx::query("UPDATE agent_devices SET last_seen_unix = 1")
+            .execute(&store.pool)
+            .await
+            .unwrap();
+        store
+            .replace_agent_device_snapshot("agent-a", &[device])
+            .await
+            .unwrap();
+        let rows = store
+            .list_agent_device_rows_for_agent("agent-a")
+            .await
+            .unwrap();
+        let included_at = u64::try_from(rows[0].device.last_seen_unix).unwrap();
+        assert_eq!(
+            store
+                .prune_devices_at(Duration::from_secs(60), included_at + 60)
+                .await
+                .unwrap(),
+            0
+        );
+        assert_eq!(
+            store
+                .prune_devices_at(Duration::from_secs(60), included_at + 61)
+                .await
+                .unwrap(),
+            1
+        );
+    }
+
     #[tokio::test]
     async fn snapshot_presence_is_derived_from_stored_endpoints() {
         let ts = TestStore::new().await;