fix(authenticator): audit drop-metric + wider refresher store-retry (post-merge #1851 review)

src/backend/services/authenticator/src/audit.rs

@@ -166,7 +166,9 @@ impl AuditEmitter {
         let meter = opentelemetry::global::meter("authenticator.audit");
         let dropped = meter
             .u64_counter("auth_audit_dropped_total")
-            .with_description("Audit events dropped (queue full or delivery failure)")
+            .with_description(
+                "Audit events dropped (queue full, serialization, or delivery failure)",
+            )
             .build();
 
         if brokers.trim().is_empty() {
@@ -213,16 +215,23 @@ impl AuditEmitter {
                     uuid::Uuid::now_v7().to_string(),
                     chrono::Utc::now().to_rfc3339_opts(chrono::SecondsFormat::Millis, true),
                 );
-                let Ok(payload) = serde_json::to_vec(&env) else {
-                    continue;
+                let payload = match serde_json::to_vec(&env) {
+                    Ok(p) => p,
+                    Err(e) => {
+                        // Count + log like every other drop path, so a
+                        // malformed event still moves auth_audit_dropped_total.
+                        dropped_in_task.add(1, &[]);
+                        tracing::warn!(target: "audit", error = %e, action = env.action, "audit event serialization failed (dropped)");
+                        continue;
+                    }
                 };
                 // Key by tenant: per-tenant ordering, balanced partitions.
                 let record = FutureRecord::to(&topic)
                     .key(&env.tenant_id)
                     .payload(&payload);
                 if let Err((e, _)) = producer.send(record, SEND_TIMEOUT).await {
                     dropped_in_task.add(1, &[]);
-                    tracing::warn!(error = %e, action = env.action, "audit event delivery failed (dropped)");
+                    tracing::warn!(target: "audit", error = %e, action = env.action, "audit event delivery failed (dropped)");
                 }
             }
         });
@@ -249,7 +258,7 @@ impl AuditEmitter {
         let Some(tx) = &self.tx else { return };
         if let Err(e) = tx.try_send(event) {
             self.dropped.add(1, &[]);
-            tracing::warn!(error = %e, "audit queue full: event dropped");
+            tracing::warn!(target: "audit", error = %e, "audit queue full: event dropped");
         }
     }
 }

src/backend/services/authenticator/src/refresher.rs

@@ -37,6 +37,15 @@ use crate::session::SessionManager;
 const LEADER_KEY: &str = "asm:leader:idp_refresher";
 /// Per-session lock TTL — covers one grant round-trip with generous margin.
 const SESSION_LOCK_TTL_MS: u64 = 30_000;
+/// Wall-clock cap on the whole lock-holding critical section (grant + store
+/// retries). Kept safely below `SESSION_LOCK_TTL_MS` so the flow can never run
+/// past the lock it holds: if it did, the lock would expire mid-rotation and a
+/// second worker could pick up the same session and burn the one-time grant.
+const LOCK_HOLD_BUDGET_MS: u64 = SESSION_LOCK_TTL_MS - 5_000;
+/// Post-grant store retries (exponential backoff 200ms→3.2s, ~6s total) — the
+/// grant is already rotated, so we must persist the new token or the next
+/// attempt burns it; wide enough to ride a Redis blip, under the lock TTL.
+const STORE_RETRY_ATTEMPTS: u32 = 6;
 /// Backoff for transient failures: `min(base << failures, max)` seconds.
 const BACKOFF_BASE_SECONDS: u64 = 15;
 const BACKOFF_MAX_SECONDS: u64 = 300;
@@ -165,23 +174,39 @@ async fn run(state: Arc<AppState>, cancel: CancellationToken) {
 /// Refresh a single due session under its rotation lock.
 async fn refresh_one(state: &Arc<AppState>, metrics: &Metrics, session_id: &str) {
     let sessions = &state.sessions;
-    match sessions
+    let owner_token = match sessions
         .lock_session_refresh(session_id, SESSION_LOCK_TTL_MS)
         .await
     {
-        Ok(true) => {}
-        Ok(false) => return, // another worker is mid-rotation
+        Ok(Some(token)) => token,
+        Ok(None) => return, // another worker is mid-rotation
         Err(e) => {
             tracing::warn!(error = %e, session_id, "refresh lock failed");
             return;
         }
-    }
+    };
 
-    let result = do_refresh(state, metrics, session_id).await;
-    if let Err(e) = result {
-        tracing::warn!(error = %e, session_id, "idp refresh: store error");
+    // Bound the critical section below the lock TTL so we can never operate on
+    // an expired lock (which a second worker could have re-acquired).
+    match tokio::time::timeout(
+        Duration::from_millis(LOCK_HOLD_BUDGET_MS),
+        do_refresh(state, metrics, session_id),
+    )
+    .await
+    {
+        Ok(Ok(())) => {}
+        Ok(Err(e)) => tracing::warn!(error = %e, session_id, "idp refresh: store error"),
+        Err(_) => tracing::error!(
+            session_id,
+            budget_ms = LOCK_HOLD_BUDGET_MS,
+            "idp refresh exceeded the lock budget; aborting so the lock is not outlived"
+        ),
     }
-    if let Err(e) = sessions.unlock_session_refresh(session_id).await {
+    // Owner-safe release: only drops the lock if we still hold this token.
+    if let Err(e) = sessions
+        .unlock_session_refresh(session_id, &owner_token)
+        .await
+    {
         tracing::debug!(error = %e, session_id, "refresh unlock failed (lock TTL covers it)");
     }
 }
@@ -231,8 +256,12 @@ async fn do_refresh(
             // token → invalid_grant → false logout (review M3). So retry the
             // store a few times before giving up; the guard returns false only
             // when the session was concurrently revoked (then just unschedule).
+            // Retry generously: exponential backoff 200ms→3.2s over STORE_RETRY
+            // attempts (~6s total), to ride out a realistic Redis failover/blip
+            // rather than only a sub-second hiccup. Still far under the 30s
+            // per-session lock TTL, so the lock/permit is never held past it.
             let mut stored = false;
-            for attempt in 0..3u32 {
+            for attempt in 0..STORE_RETRY_ATTEMPTS {
                 match sessions
                     .store_idp_refresh(
                         session_id,
@@ -252,12 +281,13 @@ async fn do_refresh(
                         stored = true;
                         break;
                     }
-                    Err(e) if attempt == 2 => {
+                    Err(e) if attempt == STORE_RETRY_ATTEMPTS - 1 => {
                         tracing::error!(error = %e, session_id, "idp refresh: store failed after retries — the rotated token is lost, session will be logged out on the next attempt");
                     }
                     Err(e) => {
-                        tracing::warn!(error = %e, session_id, attempt, "idp refresh store failed, retrying");
-                        tokio::time::sleep(std::time::Duration::from_millis(200)).await;
+                        let backoff_ms = 200u64 << attempt;
+                        tracing::warn!(error = %e, session_id, attempt, backoff_ms, "idp refresh store failed, retrying");
+                        tokio::time::sleep(std::time::Duration::from_millis(backoff_ms)).await;
                     }
                 }
             }

src/backend/services/authenticator/src/session.rs

@@ -578,37 +578,57 @@ impl SessionManager {
 
     /// Per-session refresh lock (`SET NX PX`): refresh-token rotation is
     /// one-time-use at most IdPs; two workers racing the same rotation would
-    /// burn the grant and falsely kill the session. Returns `true` when this
-    /// caller holds the lock.
+    /// burn the grant and falsely kill the session. Returns `Some(owner_token)`
+    /// when this caller acquired the lock (a fresh, unique value stored under
+    /// the key) or `None` when another worker already holds it. Hand the token
+    /// back to [`Self::unlock_session_refresh`] so the release only removes our
+    /// own lock and can never clobber a lock a later worker took after ours
+    /// expired.
     ///
     /// # Errors
     /// Fails on a Redis error.
     pub async fn lock_session_refresh(
         &self,
         session_id: &str,
         ttl_ms: u64,
-    ) -> anyhow::Result<bool> {
+    ) -> anyhow::Result<Option<String>> {
         let mut conn = self.conn.clone();
+        let token = uuid::Uuid::now_v7().to_string();
         let set: Option<String> = redis::cmd("SET")
             .arg(format!("asm:refresh_lock:{session_id}"))
-            .arg("1")
+            .arg(&token)
             .arg("NX")
             .arg("PX")
             .arg(ttl_ms.max(1))
             .query_async(&mut conn)
             .await
             .context("acquire per-session refresh lock")?;
-        Ok(set.is_some())
+        Ok(set.is_some().then_some(token))
     }
 
-    /// Release the per-session refresh lock.
+    /// Release the per-session refresh lock, but only if `owner_token` still
+    /// matches the stored value (compare-and-del via Lua). If our lock already
+    /// expired and a later worker re-acquired it, this is a no-op — we never
+    /// delete someone else's lock.
     ///
     /// # Errors
     /// Fails on a Redis error.
-    pub async fn unlock_session_refresh(&self, session_id: &str) -> anyhow::Result<()> {
+    pub async fn unlock_session_refresh(
+        &self,
+        session_id: &str,
+        owner_token: &str,
+    ) -> anyhow::Result<()> {
+        const UNLOCK: &str = r"
+            if redis.call('GET', KEYS[1]) == ARGV[1] then
+                return redis.call('DEL', KEYS[1])
+            end
+            return 0
+        ";
         let mut conn = self.conn.clone();
-        let _: i64 = conn
-            .del(format!("asm:refresh_lock:{session_id}"))
+        let _: i64 = redis::Script::new(UNLOCK)
+            .key(format!("asm:refresh_lock:{session_id}"))
+            .arg(owner_token)
+            .invoke_async(&mut conn)
             .await
             .context("release per-session refresh lock")?;
         Ok(())