refactor: use semaphores for async serialization gates

codex-rs/login/src/agent_identity.rs

@@ -22,7 +22,7 @@ use serde::Deserialize;
 use serde::Serialize;
 use sha2::Digest as _;
 use sha2::Sha512;
-use tokio::sync::Mutex;
+use tokio::sync::Semaphore;
 use tracing::debug;
 use tracing::info;
 use tracing::warn;
@@ -42,7 +42,7 @@ pub(crate) struct BackgroundAgentTaskManager {
     chatgpt_base_url: String,
     auth_mode: BackgroundAgentTaskAuthMode,
     abom: AgentBillOfMaterials,
-    ensure_lock: Arc<Mutex<()>>,
+    ensure_lock: Arc<Semaphore>,
 }
 
 impl std::fmt::Debug for BackgroundAgentTaskManager {
@@ -158,7 +158,7 @@ impl BackgroundAgentTaskManager {
             chatgpt_base_url: normalize_chatgpt_base_url(&chatgpt_base_url),
             auth_mode,
             abom: build_abom(session_source),
-            ensure_lock: Arc::new(Mutex::new(())),
+            ensure_lock: Arc::new(Semaphore::new(1)),
         }
     }
 
@@ -186,7 +186,11 @@ impl BackgroundAgentTaskManager {
             return Ok(None);
         };
 
-        let _guard = self.ensure_lock.lock().await;
+        let _guard = self
+            .ensure_lock
+            .acquire()
+            .await
+            .context("background agent task ensure semaphore closed")?;
         let mut stored_identity = self
             .ensure_registered_identity_for_binding(auth, &binding)
             .await?;

codex-rs/login/src/auth/manager.rs

@@ -15,7 +15,7 @@ use std::sync::Mutex;
 use std::sync::RwLock;
 use std::sync::atomic::AtomicU64;
 use std::sync::atomic::Ordering;
-use tokio::sync::Mutex as AsyncMutex;
+use tokio::sync::Semaphore;
 use tokio::sync::watch;
 
 use codex_app_server_protocol::AuthMode;
@@ -1200,7 +1200,7 @@ pub struct AuthManager {
     forced_chatgpt_workspace_id: RwLock<Option<String>>,
     chatgpt_base_url: RwLock<Option<String>>,
     background_agent_task_auth_mode: RwLock<BackgroundAgentTaskAuthMode>,
-    refresh_lock: AsyncMutex<()>,
+    refresh_lock: Semaphore,
     external_auth: RwLock<Option<Arc<dyn ExternalAuth>>>,
     auth_state_tx: watch::Sender<()>,
 }
@@ -1285,7 +1285,7 @@ impl AuthManager {
             forced_chatgpt_workspace_id: RwLock::new(None),
             chatgpt_base_url: RwLock::new(None),
             background_agent_task_auth_mode: RwLock::new(BackgroundAgentTaskAuthMode::Disabled),
-            refresh_lock: AsyncMutex::new(()),
+            refresh_lock: Semaphore::new(1),
             external_auth: RwLock::new(None),
             auth_state_tx,
         }
@@ -1307,7 +1307,7 @@ impl AuthManager {
             forced_chatgpt_workspace_id: RwLock::new(None),
             chatgpt_base_url: RwLock::new(None),
             background_agent_task_auth_mode: RwLock::new(BackgroundAgentTaskAuthMode::Disabled),
-            refresh_lock: AsyncMutex::new(()),
+            refresh_lock: Semaphore::new(1),
             external_auth: RwLock::new(None),
             auth_state_tx,
         })
@@ -1328,7 +1328,7 @@ impl AuthManager {
             forced_chatgpt_workspace_id: RwLock::new(None),
             chatgpt_base_url: RwLock::new(None),
             background_agent_task_auth_mode: RwLock::new(BackgroundAgentTaskAuthMode::Disabled),
-            refresh_lock: AsyncMutex::new(()),
+            refresh_lock: Semaphore::new(1),
             external_auth: RwLock::new(None),
             auth_state_tx,
         })
@@ -1347,7 +1347,7 @@ impl AuthManager {
             forced_chatgpt_workspace_id: RwLock::new(None),
             chatgpt_base_url: RwLock::new(None),
             background_agent_task_auth_mode: RwLock::new(BackgroundAgentTaskAuthMode::Disabled),
-            refresh_lock: AsyncMutex::new(()),
+            refresh_lock: Semaphore::new(1),
             external_auth: RwLock::new(Some(
                 Arc::new(BearerTokenRefresher::new(config)) as Arc<dyn ExternalAuth>
             )),
@@ -1742,7 +1742,12 @@ impl AuthManager {
     /// can assume that some other instance already refreshed it. If the persisted
     /// token is the same as the cached, then ask the token authority to refresh.
     pub async fn refresh_token(&self) -> Result<(), RefreshTokenError> {
-        let _refresh_guard = self.refresh_lock.lock().await;
+        let _refresh_guard = self.refresh_lock.acquire().await.map_err(|_| {
+            RefreshTokenError::Permanent(RefreshTokenFailedError::new(
+                RefreshTokenFailedReason::Other,
+                REFRESH_TOKEN_UNKNOWN_MESSAGE.to_string(),
+            ))
+        })?;
         let auth_before_reload = self.auth_cached();
         if auth_before_reload
             .as_ref()
@@ -1774,7 +1779,12 @@ impl AuthManager {
     /// observe refreshed token. If the token refresh fails, returns the error to
     /// the caller.
     pub async fn refresh_token_from_authority(&self) -> Result<(), RefreshTokenError> {
-        let _refresh_guard = self.refresh_lock.lock().await;
+        let _refresh_guard = self.refresh_lock.acquire().await.map_err(|_| {
+            RefreshTokenError::Permanent(RefreshTokenFailedError::new(
+                RefreshTokenFailedReason::Other,
+                REFRESH_TOKEN_UNKNOWN_MESSAGE.to_string(),
+            ))
+        })?;
         self.refresh_token_from_authority_impl().await
     }