Remove the claim instance lock (#43993)

GitOrigin-RevId: 58ecc71e53e512ce5d35cecf4f943125a8496f1f

Author: goffrie

crates/common/src/knobs.rs

@@ -1386,11 +1386,6 @@ pub static MAX_IMPORT_AGE: LazyLock<Duration> =
 pub static PARTITION_LOADER_MAX_STALE_SECS: LazyLock<Duration> =
     LazyLock::new(|| Duration::from_secs(env_config("PARTITION_LOADER_MAX_STALE_SECS", 1)));
 
-/// Seconds to wait before timing out when trying to acquire the lock for
-/// claiming an instance in big-brain.
-pub static CLAIM_INSTANCE_TIMEOUT_SECS: LazyLock<Duration> =
-    LazyLock::new(|| Duration::from_secs(env_config("CLAIM_INSTANCE_TIMEOUT_SECS", 10)));
-
 /// The maximum number of bytes to buffer in an multipart upload.
 /// There may be stricter limits imposed by the storage provider, but this is
 /// the target max size for the buffer to protect against memory exhaustion.

crates/common/src/runtime/mod.rs

@@ -584,32 +584,6 @@ pub struct TimeoutError {
     duration: Duration,
 }
 
-pub struct MutexWithTimeout<T: Send> {
-    timeout: Duration,
-    mutex: tokio::sync::Mutex<T>,
-}
-
-impl<T: Send> MutexWithTimeout<T> {
-    pub fn new(timeout: Duration, value: T) -> Self {
-        Self {
-            timeout,
-            mutex: tokio::sync::Mutex::new(value),
-        }
-    }
-
-    pub async fn acquire_lock_with_timeout(
-        &self,
-    ) -> anyhow::Result<tokio::sync::MutexGuard<'_, T>> {
-        let acquire_lock = async { Ok(self.mutex.lock().await) };
-        select_biased! {
-            result = acquire_lock.fuse() => result,
-            _ = tokio::time::sleep(self.timeout).fuse() => {
-                anyhow::bail!(TimeoutError{description: "acquire_lock", duration: self.timeout});
-            },
-        }
-    }
-}
-
 /// Binds the current tracing & sentry contexts to the provided future.
 pub fn propagate_tracing<F: Future>(
     f: F,

crates/common/src/runtime/testing/mod.rs

@@ -208,10 +208,7 @@ impl RngCore for TestRng {
 #[cfg(test)]
 mod tests {
     use super::*;
-    use crate::runtime::{
-        JoinError,
-        MutexWithTimeout,
-    };
+    use crate::runtime::JoinError;
 
     #[test]
     fn test_runtime2() -> anyhow::Result<()> {
@@ -232,13 +229,4 @@ mod tests {
         });
         Ok(())
     }
-
-    #[tokio::test]
-    async fn test_mutex_with_timeout() -> anyhow::Result<()> {
-        let mutex = MutexWithTimeout::new(Duration::from_secs(1), ());
-        let _lock = mutex.acquire_lock_with_timeout().await?;
-        // Trying to acquire lock while the lock is already held should timeout
-        assert!(mutex.acquire_lock_with_timeout().await.is_err());
-        Ok(())
-    }
 }