diff --git a/crates/rhema-claw/src/forge.rs b/crates/rhema-claw/src/forge.rs index 5119ec5..43def46 100644 --- a/crates/rhema-claw/src/forge.rs +++ b/crates/rhema-claw/src/forge.rs @@ -31,7 +31,7 @@ use rhema_forge_integration::{ use rhema_identity::{IdentityStore, SecretKind}; use rhema_rsd::{Clock, Dispatcher, Event, KeyFn, SystemClock}; -use crate::watch_store::{AccountWatchRecord, ClawStore, WatchRecord}; +use crate::watch_store::{AccountWatchRecord, ClawStore, SCOPE_ACCOUNT, SCOPE_REPO, WatchRecord}; use rhema_forge_integration::{ForgeKind, resolve_token}; /// The credential a watch polls with: the value the forge takes as a @@ -315,7 +315,16 @@ impl Source for ForgeSource { } }; let client = self.client_for(&record); - match self.watcher.poll(client.as_ref(), &repo).await { + let polled = self.watcher.poll(client.as_ref(), &repo).await; + // Health first, whatever the outcome: a watch that has been + // failing since it was created must be able to say so. + self.record_poll( + SCOPE_REPO, + &record.forge, + &record.repo, + polled.as_ref().err().map(|err| err.to_string()), + ); + match polled { Ok(count) => fed += count, // One dead watch (a lapsed token, a forge outage) must // not starve the kind's other watches either. @@ -335,7 +344,14 @@ impl Source for ForgeSource { if watch.forge != self.name { continue; } - match self.poll_account(&watch).await { + let polled = self.poll_account(&watch).await; + self.record_poll( + SCOPE_ACCOUNT, + &watch.forge, + &watch.account_id, + polled.as_ref().err().map(|err| err.to_string()), + ); + match polled { Ok(count) => fed += count, // One dead account (a revoked token, a forge outage) // must not starve the kind's other watches either. @@ -350,6 +366,47 @@ impl Source for ForgeSource { } impl ForgeSource { + /// Records one poll's outcome, and says something the first time a + /// streak begins: a watch refused on every tick looks exactly like a + /// quiet repository from the outside, and only the failing tick knows + /// the difference. One notification per streak — the health row + /// carries the count, so the log line does not have to. + fn record_poll(&self, scope: &str, forge: &str, subject: &str, failure: Option) { + // The table stamps in i64 millis like every other column here; + // the clock speaks u64. + let at = SystemClock.now_millis() as i64; + let recorded = match &failure { + Some(error) => self + .store + .record_poll_failure(scope, forge, subject, error, at), + None => self.store.record_poll_ok(scope, forge, subject, at), + }; + let health = match recorded { + Ok(health) => health, + Err(err) => { + // Losing the health row must not stop the poll loop. + eprintln!("[claw] warning: watch health for {subject}: {err}"); + return; + } + }; + if !health.opens_a_failure_streak() { + return; + } + let Some(error) = failure else { + return; + }; + let title = format!("{forge} watch {subject} is failing"); + let body = format!( + "{error}\n\nFirst failure of the streak: GET /v1/watches has the count and the error." + ); + if let Err(err) = + self.notifications + .push(crate::notifications::DEFAULT_DEVICE, &title, &body) + { + eprintln!("[claw] warning: notifying {title} failed: {err}"); + } + } + /// One account-watch poll: fetch the account's notifications as the /// account, triage each, and land the verdicts — agent dispatches /// ingest as role-tagged thread events, surfaces hit the @@ -654,6 +711,60 @@ mod tests { ); } + /// A watch that cannot poll is not a quiet watch: the pass records + /// the failure in the health table and tells the device once per + /// streak, so a dead credential looks dead instead of quiet. + #[tokio::test] + async fn a_failing_poll_records_health_and_notifies_once() { + let store = Arc::new(ClawStore::in_memory().expect("store opens")); + // Nothing listens on port 1: every poll fails on the connect. + store + .add("gitea", "acme/barn", Some("http://127.0.0.1:1"), "") + .expect("add"); + let (dispatcher, accounts, notifications) = test_parts(store.clone()); + let source = ForgeSource::new( + &dispatcher, + "gitea", + store.clone(), + accounts, + notifications.clone(), + ); + + source + .detect() + .await + .expect("a failed poll is not a pass error"); + let health = store + .health(SCOPE_REPO, "gitea", "acme/barn") + .expect("health reads") + .expect("the failure is recorded"); + assert_eq!(health.consecutive_failures, 1); + assert_eq!(health.last_ok_at, None, "it has never polled once"); + assert!(health.last_error.is_some(), "the reason is kept"); + + let queued = notifications + .pending(crate::notifications::DEFAULT_DEVICE, 0) + .expect("the queue reads"); + assert_eq!(queued.len(), 1, "one notification for the streak"); + assert!( + queued[0].title.contains("acme/barn"), + "the title names the watch: {}", + queued[0].title + ); + + // The second failing pass extends the streak and stays quiet. + source.detect().await.expect("the second pass runs"); + let health = store + .health(SCOPE_REPO, "gitea", "acme/barn") + .expect("health reads") + .expect("the failure is still recorded"); + assert_eq!(health.consecutive_failures, 2); + let queued = notifications + .pending(crate::notifications::DEFAULT_DEVICE, 0) + .expect("the queue reads"); + assert_eq!(queued.len(), 1, "the streak is not re-announced"); + } + /// The client cache pays off when two watches share their /// coordinates: one client, not two. #[tokio::test] diff --git a/crates/rhema-claw/src/http.rs b/crates/rhema-claw/src/http.rs index 3d79365..e8c97d0 100644 --- a/crates/rhema-claw/src/http.rs +++ b/crates/rhema-claw/src/http.rs @@ -834,9 +834,19 @@ async fn watches_list(State(state): State) -> Result, (St .watches .list_account_watches() .map_err(|err| (StatusCode::INTERNAL_SERVER_ERROR, err.to_string()))?; - Ok(Json( - json!({ "watches": watches, "account_watches": account_watches }), - )) + // The poll health rides along for the same reason the account + // watches do: "has this ever polled?" is the first question a silent + // watch raises, and a tick that failed with a 401 answers it, while + // an empty table would say nothing either way. + let health = state + .watches + .list_health() + .map_err(|err| (StatusCode::INTERNAL_SERVER_ERROR, err.to_string()))?; + Ok(Json(json!({ + "watches": watches, + "account_watches": account_watches, + "health": health, + }))) } /// `{"forge", "repo", "host"?, "token"?}` — the shape every watch entry diff --git a/crates/rhema-claw/src/watch_store.rs b/crates/rhema-claw/src/watch_store.rs index 928fc1f..0999be5 100644 --- a/crates/rhema-claw/src/watch_store.rs +++ b/crates/rhema-claw/src/watch_store.rs @@ -8,6 +8,12 @@ //! config no longer carries watches at all — the table is the one //! truth, managed over `/v1/watches` and (from chat) the watch tools. //! +//! Beside it, the poll health of every watch ([`WatchHealth`]): when it +//! last polled successfully, why it last failed and how many failures +//! that makes. A watch that has been refused by its forge on every tick +//! since it was created looks exactly like a quiet repository without +//! this — the difference has to be readable somewhere. +//! //! Two connections, one file: this store opens its own rusqlite //! connection to the same database as the dispatcher's store — WAL and a //! busy timeout, the pattern rhema-rsd's SQLite tests use — and the two @@ -98,6 +104,16 @@ CREATE TABLE IF NOT EXISTS account_watches ( created_at INTEGER NOT NULL, UNIQUE(forge, account_id) ); +CREATE TABLE IF NOT EXISTS watch_health ( + scope TEXT NOT NULL, + forge TEXT NOT NULL, + subject TEXT NOT NULL, + last_ok_at INTEGER, + last_error TEXT, + last_error_at INTEGER, + consecutive_failures INTEGER NOT NULL DEFAULT 0, + PRIMARY KEY (scope, forge, subject) +); "; /// Why a watch-table operation failed. @@ -192,6 +208,45 @@ pub struct AccountWatchRecord { pub created_at: i64, } +/// The scope of a health row: a repository watch's thread feed. +pub const SCOPE_REPO: &str = "repo"; + +/// The scope of a health row: an account's notification feed. +pub const SCOPE_ACCOUNT: &str = "account"; + +/// One watch's poll health, as stored: the last success, the last +/// failure and its reason, and the streak length. `last_ok_at: None` with +/// `consecutive_failures: 0` is a watch that has not polled yet; a growing +/// `consecutive_failures` with no `last_ok_at` is a watch that has never +/// once succeeded — the shape a wrong or mislabelled credential leaves. +#[derive(Debug, Clone, PartialEq, Eq, Serialize)] +pub struct WatchHealth { + /// [`SCOPE_REPO`] or [`SCOPE_ACCOUNT`]. + pub scope: String, + /// The forge's source name: `github`, `gitea`, or `gitlab`. + pub forge: String, + /// The repo (`owner/name`) for a repo watch, the account id for an + /// account watch. + pub subject: String, + /// When a poll last succeeded, millis since the Unix epoch. + pub last_ok_at: Option, + /// The last failure's reason, as the poll reported it. + pub last_error: Option, + /// When that failure happened, millis since the Unix epoch. + pub last_error_at: Option, + /// Failures since the last success — what tells a caller to say + /// something once, not on every tick of a long outage. + pub consecutive_failures: i64, +} + +impl WatchHealth { + /// Whether this is the first failure of a streak: the moment worth a + /// notification, before the tick's log line becomes wallpaper. + pub fn opens_a_failure_streak(&self) -> bool { + self.consecutive_failures == 1 + } +} + impl ClawStore { /// Opens (creating if needed) the watch table in the database at /// `path` — the same file the dispatcher's store holds. WAL keeps the @@ -364,6 +419,142 @@ impl ClawStore { .map_err(WatchStoreError::from)?; Ok(rows) } + /// Records a successful poll: the streak resets, and the row stops + /// reading as broken. + pub fn record_poll_ok( + &self, + scope: &str, + forge: &str, + subject: &str, + at: i64, + ) -> Result { + let conn = self.conn.lock().unwrap_or_else(PoisonError::into_inner); + conn.execute( + "INSERT INTO watch_health (scope, forge, subject, last_ok_at, consecutive_failures) + VALUES (?1, ?2, ?3, ?4, 0) + ON CONFLICT(scope, forge, subject) DO UPDATE SET + last_ok_at = excluded.last_ok_at, + consecutive_failures = 0", + params![scope, forge, subject, at], + ) + .map_err(WatchStoreError::from)?; + read_health(&conn, scope, forge, subject) + } + + /// Records a failed poll and the reason: the streak grows, and the + /// last success (if any) stays visible beside it. + pub fn record_poll_failure( + &self, + scope: &str, + forge: &str, + subject: &str, + error: &str, + at: i64, + ) -> Result { + let conn = self.conn.lock().unwrap_or_else(PoisonError::into_inner); + conn.execute( + "INSERT INTO watch_health + (scope, forge, subject, last_error, last_error_at, consecutive_failures) + VALUES (?1, ?2, ?3, ?4, ?5, 1) + ON CONFLICT(scope, forge, subject) DO UPDATE SET + last_error = excluded.last_error, + last_error_at = excluded.last_error_at, + consecutive_failures = watch_health.consecutive_failures + 1", + params![scope, forge, subject, error, at], + ) + .map_err(WatchStoreError::from)?; + read_health(&conn, scope, forge, subject) + } + + /// Every health row, ordered for a stable answer. + pub fn list_health(&self) -> Result, WatchStoreError> { + let conn = self.conn.lock().unwrap_or_else(PoisonError::into_inner); + let mut statement = conn + .prepare( + "SELECT scope, forge, subject, last_ok_at, last_error, last_error_at, + consecutive_failures + FROM watch_health ORDER BY scope ASC, forge ASC, subject ASC", + ) + .map_err(WatchStoreError::from)?; + let rows = statement + .query_map([], |row| { + Ok(WatchHealth { + scope: row.get(0)?, + forge: row.get(1)?, + subject: row.get(2)?, + last_ok_at: row.get(3)?, + last_error: row.get(4)?, + last_error_at: row.get(5)?, + consecutive_failures: row.get(6)?, + }) + }) + .map_err(WatchStoreError::from)? + .collect::, rusqlite::Error>>() + .map_err(WatchStoreError::from)?; + Ok(rows) + } + + /// One watch's health, when it has polled at all. + pub fn health( + &self, + scope: &str, + forge: &str, + subject: &str, + ) -> Result, WatchStoreError> { + let conn = self.conn.lock().unwrap_or_else(PoisonError::into_inner); + let mut statement = conn + .prepare( + "SELECT scope, forge, subject, last_ok_at, last_error, last_error_at, + consecutive_failures + FROM watch_health WHERE scope = ?1 AND forge = ?2 AND subject = ?3", + ) + .map_err(WatchStoreError::from)?; + let mut rows = statement + .query_map(params![scope, forge, subject], |row| { + Ok(WatchHealth { + scope: row.get(0)?, + forge: row.get(1)?, + subject: row.get(2)?, + last_ok_at: row.get(3)?, + last_error: row.get(4)?, + last_error_at: row.get(5)?, + consecutive_failures: row.get(6)?, + }) + }) + .map_err(WatchStoreError::from)?; + match rows.next() { + Some(row) => Ok(Some(row.map_err(WatchStoreError::from)?)), + None => Ok(None), + } + } +} + +/// One health row by key: the upserts' return value, read back under the +/// connection they wrote with. +fn read_health( + conn: &rusqlite::Connection, + scope: &str, + forge: &str, + subject: &str, +) -> Result { + conn.query_row( + "SELECT scope, forge, subject, last_ok_at, last_error, last_error_at, + consecutive_failures + FROM watch_health WHERE scope = ?1 AND forge = ?2 AND subject = ?3", + params![scope, forge, subject], + |row| { + Ok(WatchHealth { + scope: row.get(0)?, + forge: row.get(1)?, + subject: row.get(2)?, + last_ok_at: row.get(3)?, + last_error: row.get(4)?, + last_error_at: row.get(5)?, + consecutive_failures: row.get(6)?, + }) + }, + ) + .map_err(WatchStoreError::from) } /// Milliseconds since the Unix epoch; stamps `created_at`. @@ -401,6 +592,51 @@ pub fn validate_watch( mod tests { use super::*; + /// A watch's health is a streak, not a flag: the first failure opens + /// it, the next extends it, and a success closes it without losing + /// the last success's stamp. + #[test] + fn a_health_row_tracks_the_streak_and_survives_a_recovery() { + let store = ClawStore::in_memory().expect("store opens"); + assert_eq!( + store + .health(SCOPE_REPO, "gitea", "acme/barn") + .expect("read"), + None, + "a watch that never polled has no row" + ); + + let first = store + .record_poll_failure(SCOPE_REPO, "gitea", "acme/barn", "401 denied", 1_000) + .expect("failure lands"); + assert_eq!(first.consecutive_failures, 1); + assert!(first.opens_a_failure_streak()); + assert_eq!(first.last_ok_at, None); + assert_eq!(first.last_error.as_deref(), Some("401 denied")); + + let second = store + .record_poll_failure(SCOPE_REPO, "gitea", "acme/barn", "401 denied", 2_000) + .expect("failure lands"); + assert_eq!(second.consecutive_failures, 2); + assert!( + !second.opens_a_failure_streak(), + "only the first failure of a streak is news" + ); + assert_eq!(second.last_error_at, Some(2_000)); + + let recovered = store + .record_poll_ok(SCOPE_REPO, "gitea", "acme/barn", 3_000) + .expect("success lands"); + assert_eq!(recovered.consecutive_failures, 0); + assert_eq!(recovered.last_ok_at, Some(3_000)); + assert_eq!( + recovered.last_error.as_deref(), + Some("401 denied"), + "the last failure stays on the record after a recovery" + ); + assert_eq!(store.list_health().expect("list reads").len(), 1); + } + #[test] fn validation_fails_a_watch_loudly_and_specificly() { assert_eq!( diff --git a/crates/rhema-claw/src/watch_tools.rs b/crates/rhema-claw/src/watch_tools.rs index 778062b..d9584ec 100644 --- a/crates/rhema-claw/src/watch_tools.rs +++ b/crates/rhema-claw/src/watch_tools.rs @@ -18,7 +18,7 @@ use rhema_autonomous_agent::{ToolError, ToolOp, ToolRegistry, async_trait}; use serde::Deserialize; use serde_json::{Value, json}; -use crate::watch_store::{ClawStore, validate_watch}; +use crate::watch_store::{ClawStore, SCOPE_REPO, validate_watch}; use rhema_forge_integration::ForgeKind; /// The compiled operations behind the manifests, by `run.builtin` id. @@ -161,13 +161,25 @@ struct ListWatches { impl ToolOp for ListWatches { async fn invoke(&self, _args: Value) -> Result { let watches = self.store.list().map_err(|err| err.to_string())?; + let health = self.store.list_health().map_err(|err| err.to_string())?; let rows: Vec = watches .iter() .map(|watch| { + // A watch that has never polled has no health row at all: + // it answers with every health field empty and a zero + // streak, which is exactly what it is. + let health = health.iter().find(|row| { + row.scope == SCOPE_REPO && row.forge == watch.forge && row.subject == watch.repo + }); json!({ "forge": watch.forge, "repo": watch.repo, "host": watch.host, + "last_ok_at": health.and_then(|row| row.last_ok_at), + "last_error": health.and_then(|row| row.last_error.clone()), + "consecutive_failures": health + .map(|row| row.consecutive_failures) + .unwrap_or(0), }) }) .collect(); @@ -363,8 +375,22 @@ mod tests { assert_eq!( serde_json::from_str::(&listed.content).unwrap(), json!({"watches": [ - {"forge": "github", "repo": "acme/widgets", "host": null}, - {"forge": "gitlab", "repo": "group/prop", "host": null}, + { + "forge": "github", + "repo": "acme/widgets", + "host": null, + "last_ok_at": null, + "last_error": null, + "consecutive_failures": 0, + }, + { + "forge": "gitlab", + "repo": "group/prop", + "host": null, + "last_ok_at": null, + "last_error": null, + "consecutive_failures": 0, + }, ]}), "{}", listed.content diff --git a/crates/rhema-claw/tools/list_watches/TOOL.yaml b/crates/rhema-claw/tools/list_watches/TOOL.yaml index 3fd1fd2..0252706 100644 --- a/crates/rhema-claw/tools/list_watches/TOOL.yaml +++ b/crates/rhema-claw/tools/list_watches/TOOL.yaml @@ -2,7 +2,11 @@ # compiled builtin named under `run`. name: list_watches description: >- - List the watched repositories: forge, repo, and host for each. + List the watched repositories: forge, repo, and host for each, with its + poll health — last_ok_at is when it last polled successfully, + consecutive_failures is the streak it is on now, and last_error is why + the last attempt failed. A watch with failures and no last_ok_at has + never polled once. replay_safety: safe parameters: type: object