fix(ci): stop stale CI reds, keep unit tests off the host env, retry CyberArk policy conflicts

.circleci/scripts/run_integration.sh

@@ -26,6 +26,7 @@ guard_created=false
 guard_installed=false
 guard6_created=false
 guard6_installed=false
+egress_cgroup=litellm-integration
 cleanup() {
   original_status=$?
   trap - EXIT INT TERM
@@ -47,14 +48,14 @@ cleanup() {
     fi
   done
   if [ "$guard_installed" = true ]; then
-    sudo iptables -D OUTPUT -m owner --uid-owner "$(id -u)" -j integration_only || original_status=1
+    sudo iptables -D OUTPUT -m cgroup --path "$egress_cgroup" -j integration_only || original_status=1
   fi
   if [ "$guard_created" = true ]; then
     sudo iptables -F integration_only || original_status=1
     sudo iptables -X integration_only || original_status=1
   fi
   if [ "$guard6_installed" = true ]; then
-    sudo ip6tables -D OUTPUT -m owner --uid-owner "$(id -u)" -j integration_only || original_status=1
+    sudo ip6tables -D OUTPUT -m cgroup --path "$egress_cgroup" -j integration_only || original_status=1
   fi
   if [ "$guard6_created" = true ]; then
     sudo ip6tables -F integration_only || original_status=1
@@ -100,6 +101,8 @@ if [ "$mode" = parity ]; then
   export INTEGRATION_ROUTING=capture
 fi
 
+sudo mkdir -p "/sys/fs/cgroup/$egress_cgroup"
+echo "$$" | sudo tee "/sys/fs/cgroup/$egress_cgroup/cgroup.procs" > /dev/null
 sudo iptables -N integration_only
 guard_created=true
 sudo iptables -A integration_only -o lo -j ACCEPT
@@ -109,13 +112,13 @@ for service in postgres-db redis-cache; do
   sudo iptables -A integration_only -d "$address" -j ACCEPT
 done
 sudo iptables -A integration_only -j REJECT
-sudo iptables -I OUTPUT 1 -m owner --uid-owner "$(id -u)" -j integration_only
+sudo iptables -I OUTPUT 1 -m cgroup --path "$egress_cgroup" -j integration_only
 guard_installed=true
 sudo ip6tables -N integration_only
 guard6_created=true
 sudo ip6tables -A integration_only -o lo -j ACCEPT
 sudo ip6tables -A integration_only -j REJECT
-sudo ip6tables -I OUTPUT 1 -m owner --uid-owner "$(id -u)" -j integration_only
+sudo ip6tables -I OUTPUT 1 -m cgroup --path "$egress_cgroup" -j integration_only
 guard6_installed=true
 
 if curl --noproxy '*' --connect-timeout 2 -s http://198.51.100.1 >/dev/null 2>&1; then

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

@@ -48,6 +48,7 @@ pub struct CyberArkSecretManager {
     token: Cache<(), SecretValue>,
     secrets: SecretCache<String, SecretValue>,
     authentication_lock: Arc<tokio::sync::Mutex<()>>,
+    policy_load_lock: Arc<tokio::sync::Mutex<()>>,
 }
 
 #[derive(Clone, Copy, Debug, PartialEq, Eq)]

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

@@ -26,6 +26,7 @@ impl CyberArkSecretManager {
             token,
             secrets,
             authentication_lock: Arc::new(tokio::sync::Mutex::new(())),
+            policy_load_lock: Arc::new(tokio::sync::Mutex::new(())),
         }
     }
 

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

@@ -1,5 +1,8 @@
 use super::*;
 
+const POLICY_LOAD_ATTEMPTS: u32 = 5;
+const POLICY_LOAD_RETRY_DELAY: std::time::Duration = std::time::Duration::from_millis(200);
+
 impl CyberArkSecretManager {
     pub async fn async_write_secret(
         &self,
@@ -105,37 +108,43 @@ impl CyberArkSecretManager {
             "- !variable {}\n",
             serde_json::to_string(name).expect("serializing a string cannot fail")
         );
-        let response = with_timeout(
-            self.client
-                .post(policy_url)
-                .header("Authorization", authorization)
-                .header("Content-Type", "application/x-yaml")
-                .body(body),
-            context,
-        )
-        .send()
-        .await;
-        match response {
-            Ok(response) if response.status().is_success() => {}
-            Ok(response)
-                if matches!(
-                    response.status(),
-                    reqwest::StatusCode::CONFLICT | reqwest::StatusCode::UNPROCESSABLE_ENTITY
-                ) =>
-            {
-                litellm_tracing::debug!(
-                    "CyberArk variable policy already exists or conflicts: {}",
-                    response.status()
-                );
-            }
-            Ok(response) => {
-                litellm_tracing::warn!(
-                    "Could not ensure CyberArk variable exists: {}",
-                    response.status()
-                );
-            }
-            Err(error) => {
-                litellm_tracing::warn!("Error ensuring CyberArk variable exists: {error}");
+        let _policy_load = self.policy_load_lock.lock().await;
+        for attempt in 0..POLICY_LOAD_ATTEMPTS {
+            let response = with_timeout(
+                self.client
+                    .post(policy_url.clone())
+                    .header("Authorization", authorization.clone())
+                    .header("Content-Type", "application/x-yaml")
+                    .body(body.clone()),
+                context,
+            )
+            .send()
+            .await;
+            match response {
+                Ok(response)
+                    if response.status() == reqwest::StatusCode::CONFLICT
+                        && attempt + 1 < POLICY_LOAD_ATTEMPTS =>
+                {
+                    tokio::time::sleep(POLICY_LOAD_RETRY_DELAY * 2_u32.pow(attempt)).await;
+                }
+                Ok(response) if response.status().is_success() => return,
+                Ok(response) if response.status() == reqwest::StatusCode::UNPROCESSABLE_ENTITY => {
+                    litellm_tracing::debug!(
+                        "CyberArk variable policy was rejected as unprocessable"
+                    );
+                    return;
+                }
+                Ok(response) => {
+                    litellm_tracing::warn!(
+                        "Could not ensure CyberArk variable exists: {}",
+                        response.status()
+                    );
+                    return;
+                }
+                Err(error) => {
+                    litellm_tracing::warn!("Error ensuring CyberArk variable exists: {error}");
+                    return;
+                }
             }
         }
     }

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

@@ -6,7 +6,7 @@ async fn rejected_write_token_is_reauthenticated_once() {
     let server = MockServer::start().await;
     mount_auth(&server, 2).await;
     Mock::given(path("/policies/acct/policy/root"))
-        .respond_with(ResponseTemplate::new(409))
+        .respond_with(ResponseTemplate::new(201))
         .expect(1)
         .mount(&server)
         .await;
@@ -45,7 +45,6 @@ async fn rejected_write_token_is_reauthenticated_once() {
 
 #[rstest]
 #[case::created(201)]
-#[case::already_exists(409)]
 #[case::unprocessable(422)]
 #[case::server_error(500)]
 #[tokio::test]
@@ -81,13 +80,81 @@ async fn writes_tolerate_policy_status_and_cache_value(#[case] policy_status: u1
     );
 }
 
+#[rstest]
+#[tokio::test]
+async fn policy_load_conflict_is_retried_before_the_value_write() {
+    let server = MockServer::start().await;
+    mount_auth(&server, 1).await;
+    let policy_loads = Arc::new(AtomicUsize::new(0));
+    let policy_loads_for_response = Arc::clone(&policy_loads);
+    Mock::given(path("/policies/acct/policy/root"))
+        .respond_with(move |_: &Request| {
+            if policy_loads_for_response.fetch_add(1, Ordering::SeqCst) < 2 {
+                ResponseTemplate::new(409)
+            } else {
+                ResponseTemplate::new(201)
+            }
+        })
+        .expect(3)
+        .mount(&server)
+        .await;
+    let policy_loads_at_value_write = Arc::clone(&policy_loads);
+    Mock::given(method("POST"))
+        .and(path("/secrets/acct/variable/key"))
+        .respond_with(move |_: &Request| {
+            if policy_loads_at_value_write.load(Ordering::SeqCst) == 3 {
+                ResponseTemplate::new(201)
+            } else {
+                ResponseTemplate::new(404)
+            }
+        })
+        .expect(1)
+        .mount(&server)
+        .await;
+    let manager = manager(&server, Duration::from_secs(60));
+
+    manager
+        .async_write_secret("key", &SecretValue::new("v"), None)
+        .await
+        .unwrap();
+}
+
+#[rstest]
+#[tokio::test]
+async fn concurrent_writes_load_policy_one_at_a_time() {
+    let server = MockServer::start().await;
+    mount_auth(&server, 1).await;
+    Mock::given(path("/policies/acct/policy/root"))
+        .respond_with(ResponseTemplate::new(201).set_delay(Duration::from_millis(100)))
+        .expect(4)
+        .mount(&server)
+        .await;
+    Mock::given(method("POST"))
+        .respond_with(ResponseTemplate::new(201))
+        .mount(&server)
+        .await;
+    let manager = manager(&server, Duration::from_secs(60));
+    let started = std::time::Instant::now();
+
+    let value = SecretValue::new("v");
+    let results = tokio::join!(
+        manager.async_write_secret("key-0", &value, None),
+        manager.async_write_secret("key-1", &value, None),
+        manager.async_write_secret("key-2", &value, None),
+        manager.async_write_secret("key-3", &value, None),
+    );
+
+    assert!(results.0.is_ok() && results.1.is_ok() && results.2.is_ok() && results.3.is_ok());
+    assert!(started.elapsed() >= Duration::from_millis(400));
+}
+
 #[rstest]
 #[tokio::test]
 async fn failed_value_write_is_not_cached() {
     let server = MockServer::start().await;
     mount_auth(&server, 1).await;
     Mock::given(path("/policies/acct/policy/root"))
-        .respond_with(ResponseTemplate::new(409))
+        .respond_with(ResponseTemplate::new(201))
         .mount(&server)
         .await;
     Mock::given(path("/secrets/acct/variable/key"))

litellm/litellm_core_utils/logging_worker.py

@@ -165,7 +165,9 @@ def start(self) -> None:
         if self._worker_task is None or self._worker_task.done():
             self._worker_task = asyncio.create_task(self._worker_loop())
 
-    async def _process_log_task(self, task: LoggingTask, sem: asyncio.Semaphore):
+    async def _process_log_task(
+        self, task: LoggingTask, sem: asyncio.Semaphore, queue: "asyncio.Queue[LoggingTask]"
+    ) -> None:
         """Runs the logging task and handles cleanup. Releases semaphore when done."""
         try:
             if self._queue is not None:
@@ -182,7 +184,7 @@ async def _process_log_task(self, task: LoggingTask, sem: asyncio.Semaphore):
                     verbose_logger.exception("LoggingWorker error: %s", e)
                 finally:
                     self._untrack_dequeued(task)
-                    self._queue.task_done()
+                    queue.task_done()
         finally:
             # Always release semaphore, even if queue is None
             sem.release()
@@ -219,18 +221,19 @@ def _emit_timeout_summary(self) -> None:
     async def _worker_loop(self) -> None:
         """Main worker loop that gets tasks and schedules them to run concurrently."""
         try:
-            if self._queue is None or self._sem is None:
+            queue: Final = self._queue
+            if queue is None or self._sem is None:
                 return
 
             while True:
                 # Acquire semaphore before removing task from queue to prevent
                 # unbounded growth of waiting tasks
                 await self._sem.acquire()
                 try:
-                    task = await self._queue.get()
+                    task = await queue.get()
                     self._track_dequeued(task)
                     # Track each spawned coroutine so we can cancel on shutdown.
-                    processing_task = asyncio.create_task(self._process_log_task(task, self._sem))
+                    processing_task = asyncio.create_task(self._process_log_task(task, self._sem, queue))
                     self._running_tasks.add(processing_task)
                     processing_task.add_done_callback(self._running_tasks.discard)
                 except Exception:
@@ -497,7 +500,8 @@ async def clear_queue(self):
         """
         Clear the queue with a maximum time limit.
         """
-        if self._queue is None:
+        queue: Final = self._queue
+        if queue is None:
             return
 
         start_time: Final = asyncio.get_event_loop().time()
@@ -509,7 +513,7 @@ async def clear_queue(self):
                 break
 
             try:
-                task = self._queue.get_nowait()
+                task = queue.get_nowait()
                 # Await the coroutine to properly execute and avoid "never awaited" warnings
                 try:
                     await asyncio.wait_for(
@@ -522,7 +526,7 @@ async def clear_queue(self):
                 finally:
                     # Clear reference to prevent memory leaks
                     task = None
-                self._queue.task_done()  # If you're using join() elsewhere
+                queue.task_done()
             except asyncio.QueueEmpty:
                 break