From fb9e2d198ea0907a2111011aff0f602365fbd98a Mon Sep 17 00:00:00 2001 From: lda Date: Mon, 12 Oct 2026 00:41:20 +0700 Subject: [PATCH] refactor(state): trace snapshot write phases and strengthen refresh coverage --- docs/roadmap.md | 38 ++- docs/telemetry.md | 14 + .../src/api/control/fleet/refresh.rs | 273 ++++++++++++------ wakey-control-plane/src/state/store.rs | 14 +- .../src/state/store/agent_devices.rs | 40 +-- .../src/state/store/agent_devices/snapshot.rs | 48 ++- 6 files changed, 296 insertions(+), 131 deletions(-) diff --git a/docs/roadmap.md b/docs/roadmap.md index 6da9c21..0082a80 100644 --- a/docs/roadmap.md +++ b/docs/roadmap.md @@ -1,6 +1,6 @@ # Wakey Roadmap -Last updated: 2026-10-10. This records agreed work and follow-ups, not a release +Last updated: 2026-10-12. This records agreed work and follow-ups, not a release promise. Checked items are implemented; live validation is called out separately. ## Completed Observability @@ -44,16 +44,50 @@ of remembered identifiers based on observation expiry. - [x] Upgrade the matched OpenTelemetry stack to 0.33 with exporter-managed backoff/jitter retries. Controlled collector tests cover all-signal recovery, bounded retries, and no retry on authentication rejection; live outage validation remains pending. - [ ] Record hostname-resolution success/failure without exporting hostnames or resolved addresses. Current spans show timing but do not explicitly classify the result. - [ ] Investigate the slow resolver with a reproducible device-side check. Timing isolated the phase, not the cause; do not treat instrumentation as a fix. -- [x] Propagate optional W3C trace context for periodic/requested WS snapshots so agent collection and CC persistence can be inspected together. Wire compatibility and context linkage tested; live validation pending. +- [x] Propagate optional W3C trace context for periodic/requested WS snapshots so agent collection and CC persistence can be inspected together. Wire compatibility and context linkage tested; live Grafana inspection confirmed CC storage is a child of agent collection in the same trace. Storage can start after collection ends: this is an asynchronous handoff, not a persistence acknowledgement. - [x] Add current-state gauges: retained device/endpoint rows per agent and unfiltered grouped fleet identities split saved/unsaved. Sample independently every 30 seconds with bounded revision checks; replace labels and suppress failed samples. Collector and state lifecycle tests included; dashboard recipes ready. - [ ] Decide whether snapshots need persistence acknowledgements. The existing sent-device counter does not prove CC storage; any confirmed-storage metric needs an authoritative measurement point. +- [x] Break down snapshot persistence timing into pool acquisition, transaction acquisition, row replacement, and commit child spans. Shared write acquisition retains SQLx ownership and `BEGIN IMMEDIATE`; snapshot row work stays in the caller's transaction. Existing rollback and cache-refresh tests cover behavior. Live trace screenshot confirmed all phases: about 83 microseconds acquiring a connection, 154 microseconds beginning the transaction, 12.8 ms replacing rows, and 18.94 ms committing. This sample does not justify changing SQLite durability or prove a persistent bottleneck. See [telemetry span semantics](telemetry.md#signals). The existing snapshot counter is cumulative collection volume, not unique devices or current fleet size. Metric labels should remain bounded; no per-device labels. +## Process And Host Visibility + +- [ ] Add shared process measurements for agent and CC: start time/uptime, resident memory, CPU time, and open file descriptors where supported. Keep version as resource metadata rather than repeating it on every sample. +- [ ] Add a small host summary: hostname, OS/kernel, CPU count, host uptime, load, and available/total memory. Distinguish host uptime from process uptime so router reboots and agent restarts are visible separately. Sources must tolerate missing Linux files; not every agent host is a router. +- [ ] Expose changing measurements through OTLP on a modest sampling interval; show selected process/host details on Fleet Health and agent details. Keep host information behind the protected operator interface, not public health endpoints. Avoid expensive reads on every request and unbounded labels. +- [ ] Keep `/healthz` as cheap liveness; consider a separate bounded `/readyz` database-access check for CC. Collector outages and disconnected fleet agents must not make a functioning CC unready. +- [ ] Add application resource gauges where useful: DB pool usage, connected agents, pending commands, and active terminal sessions. Check existing instruments first to avoid duplicates; distinguish pool pressure from host/process pressure. + +## Device Diagnostics + +- [ ] Add agent-side ICMP ping through the existing command relay, with an explicit IP and selected agent/network route. Bound probe count, total duration, output and concurrent work; no shell interpolation. Account for missing tools and ICMP permissions, and support IPv4/IPv6 where available. +- [ ] Return probe replies, loss and round-trip times; offer Ping alongside Wake in device details. MAC-only devices cannot be pinged without an IP. A reply means that IP responded from that agent's network, not proof of saved identity; no reply must not automatically mark a device offline or remove an identifier. + +## Optional Agent Sources And Capabilities + +These are exploration items, not a commitment to a router-only model or a new +message bus. Sparse WSL inventory is legitimate; missing evidence is not failure. + +- [ ] Discover and advertise usable observation sources separately from supported commands and configured permissions. Extend the existing hello/capability mechanism where appropriate; CC/UI must tolerate missing sources and older agents. Do not infer router status from Linux alone. +- [ ] Investigate read-only Wi-Fi station data on the actual router: signal in **dBm**, association/interface, rates and traffic counters where available. Evaluate driver-supported `iw`/netlink or local `ubus`/hostapd interfaces before choosing a dependency. Keep measurements scoped to agent/interface/station with source and capture time; they are not permanent identity or globally valid presence. UI may show a signal badge only when supported and sufficiently fresh. +- [ ] Investigate interface lifecycle hooks for WAN/link changes, keeping scripts minimal and logic in the binary. Interface up/down does not prove Internet reachability; any active connectivity probe needs an explicit target and bounded policy. Existing DHCP/neighbor hooks remain independent evidence, not a way to enrich or overwrite other sources. +- [ ] Keep source support, operator authorization and local agent policy distinct. Capability advertisement is not permission: CC checks authorization and the agent enforces configured limits. New diagnostic APIs should be typed, protected operations, not arbitrary shell, UCI or bus calls exposed to the browser. + +OpenWrt [hotplug](https://openwrt.org/docs/guide-user/base-system/hotplug) provides +event hooks; [ubus](https://openwrt.org/docs/techref/ubus) is local IPC/RPC, and +[UCI](https://openwrt.org/docs/guide-developer/ubus/uci) manages configuration. +None is a Kafka-style retained observation history. Verify what the vendor +firmware actually exposes before making any of them an agent requirement. + ## Validation And Reliability - [x] Bound fleet-cache rebuilds to three attempts under sustained mutations; return retryable HTTP 503 `fleet_busy` rather than silently serving stale wake routes. Deterministic limit/recovery tests included. +- [x] Separate fleet wake selection from dispatch; wake lookup uses the complete cached view rather than the presentation list's 1,000-row cap. Automated coverage checks explicit-route rejection and lookup beyond the cap; live checks confirm missing-device and invalid-route errors without dispatching wakes. +- [x] Share device read assembly within a single SQLite read transaction. Query-plan inspection on the test database with Python SQLite confirmed all six per-agent reads use primary-key searches; full-fleet reads scan intentionally, with covering indexes for MAC/IP projections and no temporary sort. No additional indexes justified by this check. +- [x] Separate fleet refresh selection, relay, decoding/storage, and aggregation. Preserve concurrency of eight and per-agent failures; tests cover disconnected agents, sorted partial results, invalid payload preservation, and valid empty snapshot replacement. +- [x] Add a refresh integration regression: a simulated agent answers through the production command relay; its inventory reaches SQLite, invalidates a previously populated fleet cache, and appears in the next fleet read. A disconnected agent does not cancel success; duplicate selection sends only one command. This covers the CC application path, not a real agent process or WebSocket transport. - [ ] Reproduce/check Ctrl-C cleanup on the test device: process groups, PTYs, relay tasks, and telemetry shutdown. Leftovers were suspected, not confirmed. - [ ] Exercise collector unavailability and recovery with running agent/CC instances. Verify normal operations continue and telemetry queues remain bounded; outage losses are expected under the current best-effort policy. - [ ] Validate terminal recovery instrumentation live: generation timing/size, transport attach/resume, and backpressure episodes ending in drain or disconnect. diff --git a/docs/telemetry.md b/docs/telemetry.md index 8424068..a976548 100644 --- a/docs/telemetry.md +++ b/docs/telemetry.md @@ -106,6 +106,20 @@ existing debug events too; inspect them before enabling it broadly. - Inventory source spans separate neighbor, DHCP lease, and hook-store reads from merging. `db_*` spans cover snapshot replacement, fleet reads, and manual identity changes, including transaction waits and result assembly where applicable. + Snapshot replacement retains `db_replace_agent_snapshot` as its total duration, + with four sequential child spans: + - `db_pool_acquire`: obtaining a pooled connection, including any connection + setup/checks; it is not exclusively queue wait. + - `db_write_transaction_begin`: executing `BEGIN IMMEDIATE`, including SQLite + writer-lock wait and worker scheduling, not a direct lock-wait measurement. + - `db_snapshot_write_rows`: existing-key lookup, normalization, parent/child + writes, and deletion of devices missing from the complete snapshot. + - `db_snapshot_commit`: transaction commit, including its SQLite/worker work; + this does not isolate filesystem flush time. + Pool and transaction-begin spans come from the shared write helper, so they + also appear under other write operations. No per-row spans or SQL values are + recorded. A failed write rolls back through SQLx and has no commit span; + fleet cache invalidation still happens only after a successful commit. - Fleet spans cover list/load/grouping, refresh fan-out with per-agent outcomes, and wake dispatch. Counts distinguish grouped, filtered, and returned rows; wake events identify the chosen agent and whether the route was explicit, diff --git a/wakey-control-plane/src/api/control/fleet/refresh.rs b/wakey-control-plane/src/api/control/fleet/refresh.rs index 7ae69de..8a1faaf 100644 --- a/wakey-control-plane/src/api/control/fleet/refresh.rs +++ b/wakey-control-plane/src/api/control/fleet/refresh.rs @@ -163,75 +163,12 @@ mod tests { } } - #[test] - fn decode_keeps_relay_and_payload_failures_distinct() { - assert_eq!( - decode_inventory(Ok(response(None))).unwrap_err(), - "inventory command returned no result" - ); - assert!(decode_inventory(Ok(response(Some(json!({"devices": "invalid"}))))).is_err()); - let mut failed = response(Some(json!({"devices": []}))); - failed.status = "error".into(); - assert_eq!( - decode_inventory(Ok(failed)).unwrap_err(), - "inventory command failed" - ); - assert_eq!( - decode_inventory(Err(ApiError::new( - StatusCode::NOT_FOUND, - "agent_not_connected", - "agent is not connected" - ))) - .unwrap_err(), - "404 Not Found: agent is not connected" - ); - assert!( - decode_inventory(Ok(response(Some(json!({"devices": []}))))) - .unwrap() - .devices - .is_empty() - ); - } - - #[test] - fn summary_keeps_partial_failures_and_sorts_by_agent() { - let result = summarize(vec![ - RefreshFleetAgentResult { - agent_id: "z".into(), - status: "ok".into(), - accepted: 2, - error: None, - }, - RefreshFleetAgentResult { - agent_id: "b".into(), - status: "error".into(), - accepted: 0, - error: Some("disconnected".into()), - }, - RefreshFleetAgentResult { - agent_id: "a".into(), - status: "ok".into(), - accepted: 3, - error: None, - }, - ]); - assert_eq!(result.total_accepted, 5); - assert_eq!( - result - .agents - .iter() - .map(|row| row.agent_id.as_str()) - .collect::>(), - ["a", "b", "z"] - ); - assert_eq!(result.agents[1].error.as_deref(), Some("disconnected")); - assert!(summarize(vec![]).agents.is_empty()); - } - #[tokio::test] - async fn handler_reports_explicit_disconnected_agents_without_omitting_them() { + async fn refresh_updates_cached_fleet_despite_a_disconnected_agent() { + use crate::runtime::{AgentReply, AgentSession, SessionEvent}; use std::{collections::HashMap, sync::Arc, time::Duration}; - use tokio::sync::{Mutex, RwLock}; + use tokio::sync::{Mutex, RwLock, mpsc}; + use wakey_agent::protocol::{CommandResult, ServerMessage}; let dir = std::env::temp_dir().join(format!("wakey-refresh-handler-{}", uuid::Uuid::new_v4())); @@ -250,35 +187,164 @@ mod tests { alerts: crate::api::AlertsHub::new(), fleet: Arc::new(super::super::FleetCache::default()), }; - assert!(select_agents(&state, vec![]).await.is_empty()); - let response = refresh_fleet_devices( + let token = state + .store + .issue_enroll_token(Duration::from_secs(60)) + .await + .unwrap(); + let agent_id = state + .store + .enroll(&token.enroll_token) + .await + .unwrap() + .agent_id; + let device = |ip: &str| { + wakey_core::Device::from_parts( + vec![wakey_core::NeighborEntry { + ip: ip.parse().unwrap(), + mac: Some("aa:bb:cc:dd:ee:01".parse().unwrap()), + dev: Some("br-lan".into()), + state: wakey_core::NeighborState::Reachable, + }], + vec![], + ) + }; + state + .store + .replace_agent_device_snapshot(&agent_id, &[device("192.168.1.10")]) + .await + .unwrap(); + let (tx, mut rx) = mpsc::unbounded_channel(); + state.sessions.write().await.insert( + agent_id.clone(), + AgentSession { + connection_id: "test".into(), + tx, + capabilities: vec![], + capability_options: Default::default(), + }, + ); + assert_eq!( + select_agents(&state, vec![]).await.as_slice(), + std::slice::from_ref(&agent_id) + ); + + // Prime the real cache with old data before refresh touches the store. + let query = super::super::types::ListFleetDevicesQuery { + visibility: Some("all".into()), + ..Default::default() + }; + let before = super::super::load_fleet_devices(&state, &query) + .await + .unwrap(); + assert_eq!(before.len(), 1); + assert_eq!( + before[0].ips, + ["192.168.1.10".parse::().unwrap()] + ); + + let refresh = refresh_fleet_devices( State(state.clone()), Json(RefreshFleetDevicesRequest { - agent_ids: vec!["missing-b".into(), "missing-a".into(), "missing-b".into()], + agent_ids: vec!["disconnected".into(), agent_id.clone(), agent_id.clone()], timeout_ms: None, }), - ) - .await - .unwrap() - .into_response(); + ); + // Answer through the production relay's pending-request channel. No + // sleep or polling: receiving the command proves its waiter exists. + let reply = async { + let SessionEvent::Message(ServerMessage::Command { + request_id, + command, + .. + }) = rx.recv().await.unwrap() + else { + panic!("expected inventory command"); + }; + let AgentCommand::Inventory(request) = command else { + panic!("expected inventory command"); + }; + assert!(request.query.is_none() && request.name.is_none()); + assert!( + request.ips.is_empty() + && request.macs.is_empty() + && request.devs.is_empty() + && request.nuds.is_empty() + ); + let result = + serde_json::to_value(CommandResult::Inventory(wakey_core::DeviceInventory { + devices: vec![device("192.168.1.20")], + })) + .unwrap(); + state + .pending + .lock() + .await + .remove(request_id.as_str()) + .unwrap() + .send(AgentReply::Result(result)) + .unwrap_or_else(|_| panic!("relay waiter dropped")); + }; + let (response, ()) = tokio::join!(refresh, reply); + let response = response.unwrap().into_response(); assert_eq!(response.status(), StatusCode::OK); let body = axum::body::to_bytes(response.into_body(), 4096) .await .unwrap(); let result: RefreshFleetDevicesResponse = serde_json::from_slice(&body).unwrap(); - assert_eq!(result.total_accepted, 0); + assert_eq!(result.total_accepted, 1); assert_eq!(result.agents.len(), 2); - assert_eq!(result.agents[0].agent_id, "missing-a"); - assert_eq!(result.agents[1].agent_id, "missing-b"); - for agent in result.agents { - assert_eq!(agent.status, "error"); - assert_eq!(agent.accepted, 0); - assert_eq!( - agent.error.as_deref(), - Some("404 Not Found: agent is not connected") - ); - } + assert!( + result + .agents + .windows(2) + .all(|pair| pair[0].agent_id < pair[1].agent_id) + ); + let success = result + .agents + .iter() + .find(|row| row.agent_id == agent_id) + .unwrap(); + assert_eq!(success.status, "ok"); + assert_eq!(success.accepted, 1); + assert!(success.error.is_none()); + let failed = result + .agents + .iter() + .find(|row| row.agent_id == "disconnected") + .unwrap(); + assert_eq!(failed.status, "error"); + assert_eq!(failed.accepted, 0); + assert_eq!( + failed.error.as_deref(), + Some("404 Not Found: agent is not connected") + ); assert!(state.pending.lock().await.is_empty()); + assert!(matches!( + rx.try_recv(), + Err(mpsc::error::TryRecvError::Empty) + )); + + let after = super::super::load_fleet_devices(&state, &query) + .await + .unwrap(); + assert_eq!(after.len(), 1); + assert_eq!(after[0].device_key, before[0].device_key); + assert_eq!( + after[0].ips, + ["192.168.1.20".parse::().unwrap()] + ); + let stored = state + .store + .list_agent_device_rows_for_agent(&agent_id) + .await + .unwrap(); + assert_eq!(stored.len(), 1); + assert_eq!(stored[0].endpoints.len(), 1); + assert_eq!( + stored[0].endpoints[0].key.ip, + Some("192.168.1.20".parse().unwrap()) + ); drop(state); std::fs::remove_dir_all(dir).unwrap(); } @@ -310,10 +376,39 @@ mod tests { assert_eq!(stored.status, "ok"); assert_eq!(stored.accepted, 1); let revision = store.fleet_revision(); - let invalid = store_inventory(&store, "agent", Ok(response(None))).await; - assert_eq!(invalid.status, "error"); - assert_eq!(invalid.accepted, 0); - assert_eq!(store.fleet_revision(), revision); + let mut command_failed = response(None); + command_failed.status = "error".into(); + command_failed.error = Some(wakey_agent::protocol::ErrorPayload { + code: "inventory_failed".into(), + message: "source unavailable".into(), + retryable: None, + }); + for failed in [ + Ok(response(None)), + Ok(response(Some(json!({"devices": "invalid"})))), + Ok(command_failed), + Err(ApiError::new( + StatusCode::BAD_GATEWAY, + "agent_send_failed", + "send failed", + )), + ] { + let invalid = store_inventory(&store, "agent", failed).await; + assert_eq!(invalid.status, "error"); + assert_eq!(invalid.accepted, 0); + assert!(invalid.error.is_some()); + assert_eq!(store.fleet_revision(), revision); + let rows = store + .list_agent_device_rows_for_agent("agent") + .await + .unwrap(); + assert_eq!(rows.len(), 1); + assert_eq!(rows[0].endpoints.len(), 1); + assert_eq!( + rows[0].endpoints[0].key.ip, + Some("192.168.1.10".parse().unwrap()) + ); + } let rows = store .list_agent_device_rows_for_agent("agent") .await diff --git a/wakey-control-plane/src/state/store.rs b/wakey-control-plane/src/state/store.rs index 135df85..24d288c 100644 --- a/wakey-control-plane/src/state/store.rs +++ b/wakey-control-plane/src/state/store.rs @@ -4,7 +4,7 @@ use std::time::{Duration, SystemTime, UNIX_EPOCH}; use anyhow::{Context, Result}; use sqlx::sqlite::{SqliteConnectOptions, SqliteJournalMode, SqlitePoolOptions, SqliteSynchronous}; use sqlx::{Row, Sqlite, SqlitePool, Transaction}; -use tracing::{info, warn}; +use tracing::{Instrument, info, warn}; use uuid::Uuid; use crate::state::types::{ @@ -33,9 +33,17 @@ impl Store { /// Begins a new transaction with an IMMEDIATE write lock. /// /// Use this for all transactions that perform writes to avoid SQLite upgrade deadlocks. + /// Separate spans distinguish pool wait/setup from acquiring SQLite's writer + /// transaction. Keep SQLx ownership and rollback-on-drop behavior intact. pub async fn begin_write(&self) -> Result> { - self.pool - .begin_with("BEGIN IMMEDIATE") + let connection = self + .pool + .acquire() + .instrument(tracing::info_span!("db_pool_acquire")) + .await + .context("failed to begin write transaction")?; + Transaction::begin(connection, Some("BEGIN IMMEDIATE".into())) + .instrument(tracing::info_span!("db_write_transaction_begin")) .await .context("failed to begin write transaction") } diff --git a/wakey-control-plane/src/state/store/agent_devices.rs b/wakey-control-plane/src/state/store/agent_devices.rs index 3cf4672..7c4d5fe 100644 --- a/wakey-control-plane/src/state/store/agent_devices.rs +++ b/wakey-control-plane/src/state/store/agent_devices.rs @@ -1,4 +1,5 @@ use anyhow::{Context, Result}; +use tracing::Instrument; use wakey_core::{Device, DeviceEndpoint, DeviceId, EndpointSource}; use super::Store; @@ -53,47 +54,16 @@ impl Store { ) -> Result { let mut tx = self.begin_write().await?; - let mut incoming_keys = std::collections::HashSet::with_capacity(devices.len()); - let snapshot_time = now_unix(); - - let existing_keys: Vec = sqlx::query_scalar!( - "SELECT device_key FROM agent_devices WHERE agent_id = ?1", - agent_id - ) - .fetch_all(&mut *tx) - .await - .context("failed fetching existing keys")?; - - for device in devices { - let Some(device_id) = &device.id else { - continue; - }; - let device_key = device_key_from_id(device_id); - incoming_keys.insert(device_key.clone()); - - snapshot::write_device(&mut tx, agent_id, &device_key, device, snapshot_time).await?; - } - - for old_key in existing_keys { - if !incoming_keys.contains(&old_key) { - sqlx::query!( - "DELETE FROM agent_devices WHERE agent_id = ?1 AND device_key = ?2", - agent_id, - old_key - ) - .execute(&mut *tx) - .await - .context("failed pruning old agent device")?; - } - } + let accepted = snapshot::replace(&mut tx, agent_id, devices).await?; tx.commit() + .instrument(tracing::info_span!("db_snapshot_commit")) .await .context("failed committing device snapshot")?; self.touch_fleet(); - tracing::Span::current().record("accepted", incoming_keys.len()); - Ok(incoming_keys.len()) + tracing::Span::current().record("accepted", accepted); + Ok(accepted) } /// List all agent devices and their child rows from one consistent read view. diff --git a/wakey-control-plane/src/state/store/agent_devices/snapshot.rs b/wakey-control-plane/src/state/store/agent_devices/snapshot.rs index a581c8e..ac098b2 100644 --- a/wakey-control-plane/src/state/store/agent_devices/snapshot.rs +++ b/wakey-control-plane/src/state/store/agent_devices/snapshot.rs @@ -1,4 +1,4 @@ -//! Writes one snapshot device using the caller's transaction. No helper commits. +//! Replaces snapshot rows using the caller's transaction. No helper commits. use anyhow::{Context, Result}; use sqlx::SqliteConnection; @@ -9,7 +9,51 @@ use super::{ presence_to_str, }; -pub(super) async fn write_device( +/// Includes key lookup, normalization, parent/child writes and missing-device +/// deletion. This measures replacement work, not pure SQL execution time. +#[tracing::instrument(name = "db_snapshot_write_rows", skip_all, fields(devices = devices.len()))] +pub(super) async fn replace( + connection: &mut SqliteConnection, + agent_id: &str, + devices: &[Device], +) -> Result { + let mut incoming_keys = std::collections::HashSet::with_capacity(devices.len()); + let snapshot_time = super::now_unix(); + + let existing_keys: Vec = sqlx::query_scalar!( + "SELECT device_key FROM agent_devices WHERE agent_id = ?1", + agent_id + ) + .fetch_all(&mut *connection) + .await + .context("failed fetching existing keys")?; + + for device in devices { + let Some(device_id) = &device.id else { + continue; + }; + let device_key = super::device_key_from_id(device_id); + incoming_keys.insert(device_key.clone()); + + write_device(connection, agent_id, &device_key, device, snapshot_time).await?; + } + + for old_key in existing_keys { + if !incoming_keys.contains(&old_key) { + sqlx::query!( + "DELETE FROM agent_devices WHERE agent_id = ?1 AND device_key = ?2", + agent_id, + old_key + ) + .execute(&mut *connection) + .await + .context("failed pruning old agent device")?; + } + } + Ok(incoming_keys.len()) +} + +async fn write_device( connection: &mut SqliteConnection, agent_id: &str, device_key: &str,