hive-github-notify: pick up a rotated PAT without a restart

The poller read `<state>/github-token` once at start, so after a PAT
rotation it kept polling with the revoked token until the unit
restarted. It now re-reads the file on every tick, as
`hive-forge-notify` does: a different non-empty value rebuilds the
GitHub source, clears the own-login cache, and logs one info line. A
missing or empty file keeps the token in use.

A 401 from the notifications endpoint is now logged at warn, once per
token value; other non-2xx statuses stay at debug.

Part of #5017 (slice G).

hive-forge-notify/src/bin/hive-github-notify/main.rs

@@ -9,7 +9,9 @@
 //! and `HIVE_AGENT_SOCKET` is the harness's todo socket.
 //! Having a PAT *is* the opt-in: with no token the poller logs why and
 //! exits 0, so deploying this unit to an agent that never gets one costs a
-//! settled process rather than a restart loop.
+//! settled process rather than a restart loop. Once polling, it re-reads
+//! `github-token` on every tick, so a rotated PAT takes effect on the next
+//! poll without a restart.
 //!
 //! ⚠️ Reading the notification stream needs the **`notifications` scope**
 //! on the PAT — a token minted for `gh` + `git push` usually carries `repo`
@@ -75,8 +77,8 @@ async fn github_loop(state_dir: String, socket: std::path::PathBuf) {
         }
     };
 
-    let source = GithubSource::new(&token);
-    let mut own_login = resolve_own_login(&client, &source).await;
+    let mut pat = Pat::new(source::API_BASE, token_path, token);
+    let mut own_login = resolve_own_login(&client, &pat.source).await;
     let mut delivered: HashMap<String, String> = HashMap::new();
 
     // GitHub tells callers how often it is willing to be polled
@@ -93,10 +95,7 @@ async fn github_loop(state_dir: String, socket: std::path::PathBuf) {
 
     loop {
         interval.tick().await;
-        if own_login.is_empty() {
-            own_login = resolve_own_login(&client, &source).await;
-        }
-        let hint = poll_once(&source, &client, &socket, &mut delivered, &own_login).await;
+        let hint = poll_tick(&mut pat, &mut own_login, &client, &socket, &mut delivered).await;
         if let Some(secs) = hint.filter(|s| *s > cadence) {
             debug!(
                 secs,
@@ -109,3 +108,194 @@ async fn github_loop(state_dir: String, socket: std::path::PathBuf) {
         }
     }
 }
+
+/// One poll: pick up a rotated PAT, resolve the own login if it is not
+/// known for the PAT in use, then poll. Returns the server's poll-interval
+/// hint.
+async fn poll_tick(
+    pat: &mut Pat,
+    own_login: &mut String,
+    client: &reqwest::Client,
+    socket: &std::path::Path,
+    delivered: &mut HashMap<String, String>,
+) -> Option<u64> {
+    if pat.reread() {
+        info!("forge_notify: github token changed on disk; using the new one");
+        own_login.clear();
+    }
+    if own_login.is_empty() {
+        *own_login = resolve_own_login(client, &pat.source).await;
+    }
+    poll_once(&pat.source, client, socket, delivered, own_login).await
+}
+
+/// The PAT in use and the source built from it. The agent's store fetch
+/// replaces `github-token` by atomic rename when the PAT rotates and never
+/// deletes it, so a missing or empty file means "not readable right now"
+/// and keeps the current token.
+struct Pat {
+    api_base: String,
+    path: std::path::PathBuf,
+    token: String,
+    source: GithubSource,
+}
+
+impl Pat {
+    fn new(api_base: &str, path: impl Into<std::path::PathBuf>, token: String) -> Self {
+        let source = GithubSource::new(api_base, &token);
+        Self {
+            api_base: api_base.to_owned(),
+            path: path.into(),
+            token,
+            source,
+        }
+    }
+
+    /// Re-read the file and swap to its token if it differs. `true` when
+    /// the token changed.
+    fn reread(&mut self) -> bool {
+        match hive_forge_notify::read_first_token(std::slice::from_ref(&self.path)) {
+            Some(t) if t != self.token => {
+                self.source = GithubSource::new(&self.api_base, &t);
+                self.token = t;
+                true
+            }
+            _ => false,
+        }
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use std::collections::HashMap;
+
+    use tokio::io::{AsyncReadExt as _, AsyncWriteExt as _};
+    use tokio::sync::mpsc;
+
+    use super::{Pat, poll_tick};
+
+    fn scratch(tag: &str) -> std::path::PathBuf {
+        let ts = std::time::SystemTime::now()
+            .duration_since(std::time::UNIX_EPOCH)
+            .map_or(0, |d| d.as_nanos());
+        let dir = std::env::temp_dir().join(format!("hive-github-notify-{tag}-{ts}"));
+        std::fs::create_dir_all(&dir).expect("create scratch dir");
+        dir
+    }
+
+    /// A stand-in for api.github.com on loopback. Answers `/user` with a
+    /// login and `/notifications` with an empty page, and reports each
+    /// request as `(path, Authorization header)`.
+    async fn stub() -> (String, mpsc::UnboundedReceiver<(String, String)>) {
+        let listener = tokio::net::TcpListener::bind("127.0.0.1:0")
+            .await
+            .expect("bind");
+        let base = format!("http://{}", listener.local_addr().expect("addr"));
+        let (tx, rx) = mpsc::unbounded_channel();
+        tokio::spawn(async move {
+            loop {
+                let (mut sock, _) = listener.accept().await.expect("accept");
+                let mut buf = Vec::new();
+                let mut chunk = [0u8; 1024];
+                while !buf.windows(4).any(|w| w == b"\r\n\r\n") {
+                    let n = sock.read(&mut chunk).await.expect("read");
+                    if n == 0 {
+                        break;
+                    }
+                    buf.extend_from_slice(&chunk[..n]);
+                }
+                let head = String::from_utf8_lossy(&buf).into_owned();
+                let path = head
+                    .split_whitespace()
+                    .nth(1)
+                    .unwrap_or_default()
+                    .to_owned();
+                let auth = head
+                    .lines()
+                    .find_map(|l| {
+                        let (k, v) = l.split_once(':')?;
+                        k.eq_ignore_ascii_case("authorization")
+                            .then(|| v.trim().to_owned())
+                    })
+                    .unwrap_or_default();
+                let body = if path.starts_with("/user") {
+                    r#"{"login":"me"}"#
+                } else {
+                    "[]"
+                };
+                let reply = format!(
+                    "HTTP/1.1 200 OK\r\ncontent-type: application/json\r\n\
+                     content-length: {}\r\nconnection: close\r\n\r\n{body}",
+                    body.len()
+                );
+                let _ = sock.write_all(reply.as_bytes()).await;
+                let _ = tx.send((path, auth));
+            }
+        });
+        (base, rx)
+    }
+
+    /// Every request one tick made, in order.
+    fn drain(rx: &mut mpsc::UnboundedReceiver<(String, String)>) -> Vec<(String, String)> {
+        std::iter::from_fn(|| rx.try_recv().ok()).collect()
+    }
+
+    fn poll(path: &str) -> bool {
+        path.starts_with("/notifications")
+    }
+
+    #[tokio::test]
+    async fn a_rotated_token_is_sent_on_the_next_poll() {
+        let (base, mut rx) = stub().await;
+        let dir = scratch("rotate");
+        let path = dir.join("github-token");
+        std::fs::write(&path, "old\n").expect("write");
+        let mut pat = Pat::new(&base, &path, "old".to_owned());
+        let mut own_login = String::new();
+        let mut delivered = HashMap::new();
+        let client = reqwest::Client::new();
+        let socket = dir.join("no-socket");
+
+        poll_tick(&mut pat, &mut own_login, &client, &socket, &mut delivered).await;
+        let first = drain(&mut rx);
+        assert!(first.iter().any(|(p, a)| poll(p) && a == "Bearer old"));
+        assert_eq!(own_login, "me");
+
+        std::fs::write(&path, "new\n").expect("write");
+        poll_tick(&mut pat, &mut own_login, &client, &socket, &mut delivered).await;
+        let second = drain(&mut rx);
+        assert!(second.iter().all(|(_, a)| a == "Bearer new"), "{second:?}");
+        assert!(second.iter().any(|(p, _)| poll(p)), "{second:?}");
+        // The own login is resolved again, under the new token.
+        assert!(second.iter().any(|(p, _)| p == "/user"), "{second:?}");
+        let _ = std::fs::remove_dir_all(&dir);
+    }
+
+    #[tokio::test]
+    async fn an_empty_or_missing_file_keeps_the_token_in_use() {
+        let (base, mut rx) = stub().await;
+        let dir = scratch("keep");
+        let path = dir.join("github-token");
+        let mut pat = Pat::new(&base, &path, "old".to_owned());
+        let mut own_login = String::new();
+        let mut delivered = HashMap::new();
+        let client = reqwest::Client::new();
+        let socket = dir.join("no-socket");
+
+        std::fs::write(&path, "\n").expect("write");
+        poll_tick(&mut pat, &mut own_login, &client, &socket, &mut delivered).await;
+        let empty = drain(&mut rx);
+        assert!(empty.iter().any(|(p, _)| poll(p)), "{empty:?}");
+        assert!(empty.iter().all(|(_, a)| a == "Bearer old"), "{empty:?}");
+
+        std::fs::remove_file(&path).expect("remove");
+        poll_tick(&mut pat, &mut own_login, &client, &socket, &mut delivered).await;
+        let missing = drain(&mut rx);
+        assert!(missing.iter().any(|(p, _)| poll(p)), "{missing:?}");
+        assert!(
+            missing.iter().all(|(_, a)| a == "Bearer old"),
+            "{missing:?}"
+        );
+        let _ = std::fs::remove_dir_all(&dir);
+    }
+}

hive-forge-notify/src/bin/hive-github-notify/source.rs

@@ -7,6 +7,8 @@
 //! tolerant parse in the shared library is the right common layer, not
 //! the client.
 
+use std::sync::atomic::{AtomicBool, Ordering};
+
 use hive_forge_notify::notify::UNREAD_FETCH_LIMIT;
 use hive_forge_notify::source::Source;
 use tracing::{debug, warn};
@@ -15,7 +17,7 @@ use tracing::{debug, warn};
 /// github.com only — the same constraint the `gh` wrapper and the git
 /// credential helper already carry — so this is a constant rather than
 /// another operator-entered URL.
-const API_BASE: &str = "https://api.github.com";
+pub const API_BASE: &str = "https://api.github.com";
 
 /// Value of the `X-GitHub-Api-Version` header. GitHub dates its REST
 /// versions; pinning one means a future default bump cannot silently
@@ -27,13 +29,20 @@ const USER_AGENT: &str = "hyperhive-forge-notify";
 
 /// github.com, via a personal access token.
 pub struct GithubSource {
+    /// [`API_BASE`] everywhere but the tests, which point it at a local stub.
+    api_base: String,
     token: String,
+    /// Set once a 401 has been logged at `warn`. A source holds one token,
+    /// so this caps the warning at one per token value.
+    warned_unauthorized: AtomicBool,
 }
 
 impl GithubSource {
-    pub fn new(token: &str) -> Self {
+    pub fn new(api_base: &str, token: &str) -> Self {
         Self {
+            api_base: api_base.to_owned(),
             token: token.to_owned(),
+            warned_unauthorized: AtomicBool::new(false),
         }
     }
 }
@@ -62,7 +71,7 @@ impl Source for GithubSource {
     }
 
     async fn own_login(&self, client: &reqwest::Client) -> String {
-        let url = format!("{API_BASE}/user");
+        let url = format!("{}/user", self.api_base);
         let Ok(resp) = self.authorize(client.get(&url)).send().await else {
             return String::new();
         };
@@ -80,7 +89,10 @@ impl Source for GithubSource {
         &self,
         client: &reqwest::Client,
     ) -> Option<(Vec<serde_json::Value>, Option<u64>)> {
-        let url = format!("{API_BASE}/notifications?all=false&per_page={UNREAD_FETCH_LIMIT}");
+        let url = format!(
+            "{}/notifications?all=false&per_page={UNREAD_FETCH_LIMIT}",
+            self.api_base
+        );
         let resp = match self.authorize(client.get(&url)).send().await {
             Ok(resp) => resp,
             Err(e) => {
@@ -101,7 +113,17 @@ impl Source for GithubSource {
             // 403 with a rate-limit body and 401 on a revoked PAT are
             // both "stay quiet and retry", not fatal: the unit must not
             // restart-loop on a credential the operator fixes out of band.
-            debug!("forge_notify: github poll status {status}");
+            // A 401 is the one an operator must act on, so it is a `warn`.
+            if status == reqwest::StatusCode::UNAUTHORIZED
+                && !self.warned_unauthorized.swap(true, Ordering::Relaxed)
+            {
+                warn!(
+                    "forge_notify: github refused the PAT (401) — no notifications \
+                     until github-token holds a working one"
+                );
+            } else {
+                debug!("forge_notify: github poll status {status}");
+            }
             return None;
         }
         let raw = resp.text().await.ok()?;
@@ -116,7 +138,7 @@ impl Source for GithubSource {
     }
 
     async fn mark_read(&self, client: &reqwest::Client, id: &str) {
-        let url = format!("{API_BASE}/notifications/threads/{id}");
+        let url = format!("{}/notifications/threads/{id}", self.api_base);
         match self.authorize(client.patch(&url)).send().await {
             Ok(resp) if resp.status().is_success() => {
                 debug!(%id, "forge_notify: marked read (github)");