From ce3ef13b73e356638c2db709356ee0d027c97a22 Mon Sep 17 00:00:00 2001 From: lda Date: Sat, 10 Oct 2026 23:59:12 +0700 Subject: [PATCH] feat(state): expire stale agent devices with configurable retention --- ...3f6c8d68898654fc0873682c8212339088306.json | 12 ++ docs/config-durations.md | 15 +- docs/roadmap.md | 3 +- ...3f6c8d68898654fc0873682c8212339088306.json | 12 ++ wakey-control-plane/src/config/init.rs | 6 + wakey-control-plane/src/config/resolve.rs | 29 ++- wakey-control-plane/src/config/types.rs | 21 +++ wakey-control-plane/src/runtime/mod.rs | 5 + .../src/state/store/agent_devices.rs | 174 ++++++++++++++++++ 9 files changed, 274 insertions(+), 3 deletions(-) create mode 100644 .sqlx/query-cc4b885c31f965013ccaa6f8ae33f6c8d68898654fc0873682c8212339088306.json create mode 100644 wakey-control-plane/.sqlx/query-cc4b885c31f965013ccaa6f8ae33f6c8d68898654fc0873682c8212339088306.json diff --git a/.sqlx/query-cc4b885c31f965013ccaa6f8ae33f6c8d68898654fc0873682c8212339088306.json b/.sqlx/query-cc4b885c31f965013ccaa6f8ae33f6c8d68898654fc0873682c8212339088306.json new file mode 100644 index 0000000..438dda0 --- /dev/null +++ b/.sqlx/query-cc4b885c31f965013ccaa6f8ae33f6c8d68898654fc0873682c8212339088306.json @@ -0,0 +1,12 @@ +{ + "db_name": "SQLite", + "query": "DELETE FROM agent_devices WHERE last_seen_unix < ?1", + "describe": { + "columns": [], + "parameters": { + "Right": 1 + }, + "nullable": [] + }, + "hash": "cc4b885c31f965013ccaa6f8ae33f6c8d68898654fc0873682c8212339088306" +} diff --git a/docs/config-durations.md b/docs/config-durations.md index 8daa721..97a1d94 100644 --- a/docs/config-durations.md +++ b/docs/config-durations.md @@ -31,12 +31,25 @@ Reconnect scheduling retains its existing 100ms minimum backoff. command_timeout = "30s" enroll_token_ttl = "1d" history_retention = "30d" +device_retention = "60d" ``` All three settings must be positive. Enrollment TTL and history retention must resolve to whole seconds. Command timeout can use subsecond values. `history_retention` deletes old audit events and alert transitions, not device -snapshots or saved identities. Device retention is a separate follow-up. +snapshots or saved identities. + +`device_retention` expires agent-collected device rows and their child endpoints, +addresses, names and facts. The default is `"60d"`; `"0d"` disables expiry. +It must resolve to whole seconds. Age is measured from the last CC-received +snapshot that included the device, not endpoint timestamps or online presence. +An offline device included in new snapshots therefore stays retained. Saved +devices and their manually attached identifiers are never deleted by this cleanup. +Complete snapshots still remove devices omitted by that agent immediately; +retention handles old rows left behind when an agent stops syncing. +Expired rows are removed by the Unix daemon's existing maintenance loop every +five minutes, including an initial tick at startup. Restart CC to apply changes +to this setting; the state-only reload does not reload configuration. ## Legacy Files diff --git a/docs/roadmap.md b/docs/roadmap.md index e9f121f..9e71b39 100644 --- a/docs/roadmap.md +++ b/docs/roadmap.md @@ -36,7 +36,8 @@ of remembered identifiers based on observation expiry. ## Telemetry Follow-Ups -- [x] Human-readable agent/CC config durations with unit-aware legacy loading, stderr deprecation warnings, and canonical init-config output. See [configuration durations](config-durations.md). History retention remains separate from future device expiry. +- [x] Human-readable agent/CC config durations with unit-aware legacy loading, stderr deprecation warnings, and canonical init-config output. See [configuration durations](config-durations.md). +- [x] CC device retention by last snapshot inclusion (`device_retention = "60d"`, zero disables), independent of audit history and preserving saved identities. - [x] Separate console, OTLP log, and trace filters using per-layer filtering. Explicit signal config overrides `RUST_LOG`, then CLI verbosity supplies defaults. Unconfigured defaults remain info-level; exporter-failure logs stay local. Live validation pending. - [ ] Record hostname-resolution success/failure without exporting hostnames or resolved addresses. Current spans show timing but do not explicitly classify the result. diff --git a/wakey-control-plane/.sqlx/query-cc4b885c31f965013ccaa6f8ae33f6c8d68898654fc0873682c8212339088306.json b/wakey-control-plane/.sqlx/query-cc4b885c31f965013ccaa6f8ae33f6c8d68898654fc0873682c8212339088306.json new file mode 100644 index 0000000..438dda0 --- /dev/null +++ b/wakey-control-plane/.sqlx/query-cc4b885c31f965013ccaa6f8ae33f6c8d68898654fc0873682c8212339088306.json @@ -0,0 +1,12 @@ +{ + "db_name": "SQLite", + "query": "DELETE FROM agent_devices WHERE last_seen_unix < ?1", + "describe": { + "columns": [], + "parameters": { + "Right": 1 + }, + "nullable": [] + }, + "hash": "cc4b885c31f965013ccaa6f8ae33f6c8d68898654fc0873682c8212339088306" +} diff --git a/wakey-control-plane/src/config/init.rs b/wakey-control-plane/src/config/init.rs index 85c470b..4eb46b6 100644 --- a/wakey-control-plane/src/config/init.rs +++ b/wakey-control-plane/src/config/init.rs @@ -103,6 +103,8 @@ pub fn write_init_config(args: &InitConfigArgs) -> Result> { .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> { 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()) diff --git a/wakey-control-plane/src/config/resolve.rs b/wakey-control-plane/src/config/resolve.rs index 9f42d5b..3eb6722 100644 --- a/wakey-control-plane/src/config/resolve.rs +++ b/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 { 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; diff --git a/wakey-control-plane/src/config/types.rs b/wakey-control-plane/src/config/types.rs index bb31704..ec3dc7e 100644 --- a/wakey-control-plane/src/config/types.rs +++ b/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, @@ -37,6 +39,8 @@ pub(crate) struct FileConfig { pub(crate) enroll_token_ttl: Option, #[serde(default, with = "humantime_serde::option")] pub(crate) history_retention: Option, + #[serde(default, with = "humantime_serde::option")] + pub(crate) device_retention: Option, pub(crate) pid_file: Option, pub(crate) ui_dist_dir: Option, #[serde(alias = "enroll_tokens")] @@ -95,6 +99,7 @@ pub(crate) static DEFAULT_CONFIG: LazyLock = 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::(&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::(&output).unwrap(), file); + } + for input in [ + "device_retention = 60", + "device_retention = 1.5", + "device_retention = '-1d'", + ] { + assert!(toml::from_str::(input).is_err()); + } + } } diff --git a/wakey-control-plane/src/runtime/mod.rs b/wakey-control-plane/src/runtime/mod.rs index e1fd905..44e1dbc 100644 --- a/wakey-control-plane/src/runtime/mod.rs +++ b/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); diff --git a/wakey-control-plane/src/state/store/agent_devices.rs b/wakey-control-plane/src/state/store/agent_devices.rs index db2fbd2..a8ebcf2 100644 --- a/wakey-control-plane/src/state/store/agent_devices.rs +++ b/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 { + self.prune_devices_at(retention, now_unix()).await + } + + async fn prune_devices_at(&self, retention: std::time::Duration, now: u64) -> Result { + 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;