fix(rust): coalesce CyberArk authentication

Co-Authored-By: Devin AI <158243242+devin-ai-integration[bot]@users.noreply.github.com>

litellm-rust/crates/secrets-cyberark/Cargo.toml

@@ -16,6 +16,7 @@ thiserror.workspace = true
 veil.workspace = true
 tracing = "0.1"
 percent-encoding = "2.3"
+tokio = { workspace = true, features = ["sync"] }
 
 [dev-dependencies]
 rstest.workspace = true

litellm-rust/crates/secrets-cyberark/src/secret_manager.rs

@@ -35,6 +35,7 @@ pub struct CyberArkSecretManager {
     api_key: SecretValue,
     token: Cache<(), SecretValue>,
     secrets: Cache<String, SecretValue>,
+    authentication_lock: Arc<tokio::sync::Mutex<()>>,
 }
 
 #[derive(Clone, Copy, Debug, PartialEq, Eq)]
@@ -65,6 +66,7 @@ impl CyberArkSecretManager {
             api_key,
             token,
             secrets,
+            authentication_lock: Arc::new(tokio::sync::Mutex::new(())),
         }
     }
 
@@ -139,6 +141,10 @@ impl CyberArkSecretManager {
     }
 
     async fn authenticate(&self) -> Result<SecretValue, Error> {
+        if let Some(token) = self.token.get(&()).await {
+            return Ok(token);
+        }
+        let _guard = self.authentication_lock.lock().await;
         if let Some(token) = self.token.get(&()).await {
             return Ok(token);
         }

litellm-rust/crates/secrets-cyberark/tests/secret_manager.rs

@@ -89,6 +89,39 @@ async fn successful_reads_cache_auth_secret_and_redact_values() {
     }
 }
 
+#[tokio::test]
+async fn concurrent_reads_share_authentication_request() {
+    let server = MockServer::start().await;
+    Mock::given(path("/authn/acct/admin/authenticate"))
+        .and(body_string("k3y"))
+        .respond_with(
+            ResponseTemplate::new(200)
+                .set_body_string(TOKEN_JSON)
+                .set_delay(Duration::from_millis(20)),
+        )
+        .expect(1)
+        .mount(&server)
+        .await;
+    Mock::given(path("/secrets/acct/variable/key"))
+        .and(header(
+            "authorization",
+            format!("Token token=\"{}\"", STANDARD.encode(TOKEN_JSON)),
+        ))
+        .respond_with(ResponseTemplate::new(200).set_body_string("value"))
+        .expect(2)
+        .mount(&server)
+        .await;
+    let manager = manager(&server, Duration::from_secs(60));
+
+    let (first, second) = tokio::join!(
+        manager.async_read_secret("key"),
+        manager.async_read_secret("key")
+    );
+
+    assert_eq!(first.unwrap().unwrap().expose(), "value");
+    assert_eq!(second.unwrap().unwrap().expose(), "value");
+}
+
 #[rstest::rstest]
 #[case::not_found(404)]
 #[case::unauthorized(401)]