fix(auth): serialize token refresh and make config/session writes atomic

src/client/database_session.rs

@@ -12,7 +12,6 @@ use crate::config;
 use crate::util;
 use serde::{Deserialize, Serialize};
 use std::fs;
-use std::io::Write;
 use std::path::PathBuf;
 use std::time::{SystemTime, UNIX_EPOCH};
 
@@ -44,23 +43,13 @@ pub fn load() -> Option<DatabaseSession> {
 
 pub fn save(session: &DatabaseSession) -> Result<(), String> {
     let path = session_path().ok_or_else(|| "no database session path available".to_string())?;
-    if let Some(parent) = path.parent() {
-        fs::create_dir_all(parent).map_err(|e| format!("mkdir failed: {e}"))?;
-    }
     let json =
         serde_json::to_string_pretty(session).map_err(|e| format!("serialize failed: {e}"))?;
 
-    use std::os::unix::fs::OpenOptionsExt;
-    let mut f = fs::OpenOptions::new()
-        .create(true)
-        .write(true)
-        .truncate(true)
-        .mode(0o600)
-        .open(&path)
-        .map_err(|e| format!("open failed: {e}"))?;
-    f.write_all(json.as_bytes())
-        .map_err(|e| format!("write failed: {e}"))?;
-    Ok(())
+    // Atomic replace so a concurrent load never reads a half-written file;
+    // 0600 because the file contains a refresh token — treat it like a
+    // credential on disk.
+    util::atomic_write(&path, json.as_bytes(), 0o600)
 }
 
 #[allow(dead_code)] // Reserved for flows that re-use a cached database session.
@@ -104,7 +93,7 @@ pub fn refresh(api_url: &str, refresh_token: &str) -> Result<DatabaseSession, St
         "refresh_token": redact(refresh_token),
     });
 
-    let client = reqwest::blocking::Client::new();
+    let client = crate::client::raw_http::build_http_client();
     let req = client.post(&url).json(&body);
     let (status, body_text) =
         util::send_debug_with_redaction(&client, req, Some(&body_log), &["token", "refresh_token"])

src/client/jwt.rs

@@ -18,11 +18,12 @@
 //! persisted to the session file — it stays in the main config or the
 //! `HOTDATA_API_KEY` env var and is only used transiently to mint.
 
+use crate::client::raw_http::build_http_client;
 use crate::config;
 use crate::util;
+use crossterm::style::Stylize;
 use serde::{Deserialize, Serialize};
 use std::fs;
-use std::io::Write;
 use std::path::PathBuf;
 use std::time::{SystemTime, UNIX_EPOCH};
 
@@ -61,25 +62,13 @@ pub fn load_session() -> Option<Session> {
 
 pub fn save_session(session: &Session) -> Result<(), String> {
     let path = session_path().ok_or_else(|| "no session path available".to_string())?;
-    if let Some(parent) = path.parent() {
-        fs::create_dir_all(parent).map_err(|e| format!("mkdir failed: {e}"))?;
-    }
     let json =
         serde_json::to_string_pretty(session).map_err(|e| format!("serialize failed: {e}"))?;
 
-    // mode 0600 — session file contains a refresh token, treat it like a
-    // credential on disk.
-    use std::os::unix::fs::OpenOptionsExt;
-    let mut f = fs::OpenOptions::new()
-        .create(true)
-        .write(true)
-        .truncate(true)
-        .mode(0o600)
-        .open(&path)
-        .map_err(|e| format!("open failed: {e}"))?;
-    f.write_all(json.as_bytes())
-        .map_err(|e| format!("write failed: {e}"))?;
-    Ok(())
+    // Atomic replace so a concurrent load_session never reads a half-written
+    // file; 0600 because the session file contains a refresh token — treat it
+    // like a credential on disk.
+    util::atomic_write(&path, json.as_bytes(), 0o600)
 }
 
 pub fn clear_session() {
@@ -180,7 +169,7 @@ pub fn exchange_cli_register_code(
         "code_verifier": util::mask_credential(code_verifier),
     });
 
-    let client = reqwest::blocking::Client::new();
+    let client = build_http_client();
     let req = client.post(&url).json(&body);
     let (status, body_text) =
         util::send_debug_with_redaction(&client, req, Some(&body_log), &["token"])
@@ -217,7 +206,7 @@ pub fn mint_from_pkce_code(
         ("client_id", CLIENT_ID),
     ];
 
-    let client = reqwest::blocking::Client::new();
+    let client = build_http_client();
     let req = client.post(&url).form(&params);
     let body_log = redacted_form_body(&params);
     let (status, body_text) =
@@ -243,7 +232,7 @@ pub fn mint_from_api_token(
         ("client_id", CLIENT_ID),
     ];
 
-    let client = reqwest::blocking::Client::new();
+    let client = build_http_client();
     let req = client.post(&url).form(&params);
     let body_log = redacted_form_body(&params);
     let (status, body_text) =
@@ -259,26 +248,52 @@ pub fn mint_from_api_token(
     Ok(session_from_response(body, None, "api_token"))
 }
 
+/// Why a refresh attempt failed. The distinction decides what happens to the
+/// on-disk session: a rejection means the server saw the grant and refused
+/// it, so the session is dead; a transport failure means the request never
+/// completed and the stored refresh token was not consumed.
+#[derive(Debug)]
+pub enum RefreshError {
+    /// The request never completed (DNS, connect, timeout).
+    Transport(String),
+    /// The server answered and refused the grant (or returned an unusable
+    /// body).
+    Rejected(String),
+}
+
+impl std::fmt::Display for RefreshError {
+    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
+        match self {
+            RefreshError::Transport(msg) | RefreshError::Rejected(msg) => f.write_str(msg),
+        }
+    }
+}
+
 /// Refresh an existing session via the refresh-token grant.
-pub fn refresh(profile: &config::ProfileConfig, session: &Session) -> Result<Session, String> {
+pub fn refresh(
+    profile: &config::ProfileConfig,
+    session: &Session,
+) -> Result<Session, RefreshError> {
     let url = format!("{}/o/token/", oauth_base(profile));
     let params = [
         ("grant_type", "refresh_token"),
         ("refresh_token", session.refresh_token.as_str()),
         ("client_id", CLIENT_ID),
     ];
 
-    let client = reqwest::blocking::Client::new();
+    let client = build_http_client();
     let req = client.post(&url).form(&params);
     let body_log = redacted_form_body(&params);
     let (status, body_text) =
         util::send_debug_with_redaction(&client, req, Some(&body_log), TOKEN_REDACT_KEYS)
-            .map_err(|e| format!("connection error: {e}"))?;
+            .map_err(|e| RefreshError::Transport(format!("connection error: {e}")))?;
     if !status.is_success() {
-        return Err(format!("refresh failed: HTTP {status}: {body_text}"));
+        return Err(RefreshError::Rejected(format!(
+            "refresh failed: HTTP {status}: {body_text}"
+        )));
     }
-    let body: TokenResponse =
-        serde_json::from_str(&body_text).map_err(|e| format!("malformed token response: {e}"))?;
+    let body: TokenResponse = serde_json::from_str(&body_text)
+        .map_err(|e| RefreshError::Rejected(format!("malformed token response: {e}")))?;
     Ok(session_from_response(
         body,
         // Rotation is off server-side, so the same refresh token
@@ -320,28 +335,66 @@ pub fn ensure_access_token(
         return Ok(session.access_token);
     }
 
-    let now = now_unix();
+    fn usable_token(session: &Session, now: u64) -> Option<String> {
+        (!session.access_token.is_empty()
+            && now + REFRESH_LEEWAY_SECONDS < session.access_expires_at)
+            .then(|| session.access_token.clone())
+    }
 
-    // 1) Cached session is still good.
+    // 1) Cached session is still good — read-only fast path, no lock.
+    if let Some(session) = load_session()
+        && let Some(tok) = usable_token(&session, now_unix())
+    {
+        return Ok(tok);
+    }
+
+    // Refreshing/re-minting rewrites session.json, which every concurrent
+    // `hotdata` process shares. Serialize the slow path across processes so
+    // an expiry mid-burst produces one refresh instead of N racing ones —
+    // the losers of that race saw "session expired or revoked" and their
+    // clear_session() deleted the winner's freshly saved session. Advisory
+    // and best-effort: without flock support this degrades to the old
+    // unserialized behavior.
+    let _lock = config::lock_file("session.lock");
+
+    let now = now_unix();
     if let Some(session) = load_session() {
-        if !session.access_token.is_empty()
-            && now + REFRESH_LEEWAY_SECONDS < session.access_expires_at
-        {
-            return Ok(session.access_token);
+        // Re-check after acquiring the lock: another process may have
+        // refreshed while we waited — reuse its token instead of spending
+        // (and potentially invalidating) the shared refresh token again.
+        if let Some(tok) = usable_token(&session, now) {
+            return Ok(tok);
         }
 
         // 2) Access expired but refresh might still work.
         if !session.refresh_token.is_empty() && now < session.refresh_expires_at {
             match refresh(profile, &session) {
                 Ok(new_session) => {
                     let tok = new_session.access_token.clone();
-                    let _ = save_session(&new_session);
+                    if let Err(e) = save_session(&new_session) {
+                        // Not fatal for this invocation — the token is
+                        // valid — but the next process can't see it and
+                        // will spend the refresh token again, so make the
+                        // persistence failure visible.
+                        eprintln!(
+                            "{}",
+                            format!("warning: could not persist refreshed session: {e}").yellow()
+                        );
+                    }
                     return Ok(tok);
                 }
-                Err(_) => {
-                    // Refresh rejected — fall through to re-mint.
+                Err(RefreshError::Rejected(_)) => {
+                    // The server refused the grant: the session is dead.
+                    // Fall through to re-mint.
                     clear_session();
                 }
+                Err(RefreshError::Transport(e)) => {
+                    // The request never completed (network down, DNS,
+                    // timeout): the stored session was not consumed and is
+                    // likely still good, so keep it and surface the real
+                    // failure instead of forcing a re-login.
+                    return Err(e);
+                }
             }
         }
     }
@@ -800,7 +853,10 @@ mod tests {
         };
         let err = refresh(&profile, &session).unwrap_err();
         m.assert();
-        assert!(err.contains("400"), "got: {err}");
+        assert!(
+            matches!(&err, RefreshError::Rejected(msg) if msg.contains("400")),
+            "got: {err}"
+        );
     }
 
     // --- ensure_access_token: each branch of the decision table ---
@@ -1112,6 +1168,97 @@ mod tests {
         assert_eq!(token, "cached-jwt");
     }
 
+    #[test]
+    fn ensure_keeps_session_when_refresh_fails_at_transport_level() {
+        // A transport failure (network down, DNS, timeout) means the refresh
+        // token was never consumed — the session must survive so a working
+        // network can still use it, and the caller gets the real connection
+        // error instead of "session expired or revoked" + forced re-login.
+        let (_tmp, _guard) = with_temp_config_dir();
+        save_session(&cached_session(-10, 86400)).unwrap();
+
+        // Nothing listens on port 1 — connection refused, not an HTTP reply.
+        let profile = mock_profile("http://127.0.0.1:1");
+        let err = ensure_access_token(&profile, None).unwrap_err();
+        assert!(err.contains("connection error"), "got: {err}");
+        assert!(
+            load_session().is_some(),
+            "session must not be cleared on a transport failure"
+        );
+    }
+
+    #[test]
+    fn ensure_returns_token_even_when_session_persist_fails() {
+        // A refreshed token must reach the caller even when session.json
+        // can't be written (read-only config dir): the persistence failure
+        // is warned about, not fatal. The lock acquisition also fails here
+        // (session.lock can't be created), exercising the unlocked degrade.
+        use std::os::unix::fs::PermissionsExt;
+        let (_tmp, _guard) = with_temp_config_dir();
+        let mut server = mockito::Server::new();
+        let m = server
+            .mock("POST", "/o/token/")
+            .match_body(mockito::Matcher::UrlEncoded(
+                "grant_type".into(),
+                "refresh_token".into(),
+            ))
+            .with_status(200)
+            .with_header("content-type", "application/json")
+            .with_body(r#"{"access_token":"refreshed-jwt","expires_in":300,"refresh_token":"r2"}"#)
+            .create();
+
+        save_session(&cached_session(-10, 86400)).unwrap();
+        let profile = mock_profile(&server.url());
+
+        let dir = config::config_dir().unwrap();
+        fs::set_permissions(&dir, fs::Permissions::from_mode(0o555)).unwrap();
+        let result = ensure_access_token(&profile, None);
+        fs::set_permissions(&dir, fs::Permissions::from_mode(0o755)).unwrap();
+
+        m.assert();
+        assert_eq!(result.unwrap(), "refreshed-jwt");
+        // The on-disk session couldn't be updated — still the old one.
+        assert_eq!(load_session().unwrap().access_token, "cached-jwt");
+    }
+
+    #[test]
+    fn ensure_concurrent_expiry_refreshes_once_and_shares_result() {
+        // Regression: when the access token expired mid-burst, N concurrent
+        // processes all spent the same refresh token. The winner saved a new
+        // session; the losers got a rejection, surfaced "session expired or
+        // revoked", and their clear_session() deleted the winner's fresh
+        // session. Under the session lock exactly one refresh runs and the
+        // waiters reuse its result.
+        let (_tmp, _guard) = with_temp_config_dir();
+        let mut server = mockito::Server::new();
+        let refresh_mock = server
+            .mock("POST", "/o/token/")
+            .match_body(mockito::Matcher::UrlEncoded(
+                "grant_type".into(),
+                "refresh_token".into(),
+            ))
+            .with_status(200)
+            .with_header("content-type", "application/json")
+            .with_body(r#"{"access_token":"refreshed-jwt","expires_in":300,"refresh_token":"r2"}"#)
+            .expect(1)
+            .create();
+
+        save_session(&cached_session(-10, 86400)).unwrap();
+        let profile = mock_profile(&server.url());
+
+        let handles: Vec<_> = (0..8)
+            .map(|_| {
+                let profile = profile.clone();
+                std::thread::spawn(move || ensure_access_token(&profile, None).unwrap())
+            })
+            .collect();
+        for h in handles {
+            assert_eq!(h.join().unwrap(), "refreshed-jwt");
+        }
+        refresh_mock.assert();
+        assert_eq!(load_session().unwrap().access_token, "refreshed-jwt");
+    }
+
     #[test]
     fn ensure_clears_session_when_refresh_dies_with_no_fallback() {
         let (_tmp, _guard) = with_temp_config_dir();