Remove timeout for acquiring an isolate permit, fail immediately instead (#41988)

Instead of waiting for 100ms timeout to acquire a permit if the isolate concurrency limit is reached, fail immediately.

GitOrigin-RevId: 287cdd4620a6bdc152634bbb530ebc8e72a103ce

Author: emmaling27

crates/common/src/knobs.rs

@@ -1036,11 +1036,6 @@ pub static SEARCHLIGHT_CLUSTER_NAME: LazyLock<String> = LazyLock::new(|| {
 pub static FUNRUN_ISOLATE_ACTIVE_THREADS: LazyLock<usize> =
     LazyLock::new(|| env_config("FUNRUN_ISOLATE_ACTIVE_THREADS", 0));
 
-/// The maximum length of time to wait to start running a function (when the
-/// FUNRUN_ISOLATE_ACTIVE_THREADS limit is reached).
-pub static FUNRUN_INITIAL_PERMIT_TIMEOUT: LazyLock<Duration> =
-    LazyLock::new(|| Duration::from_millis(env_config("FUNRUN_INITIAL_PERMIT_TIMEOUT_MS", 100)));
-
 /// How long to splay deploying AWS Lambdas due to changes in the backend. This
 /// knob doesn't delay deploys that are required due to the user pushing new
 /// node actions. Only affects deploys on startup triggered by changes to

crates/isolate/src/concurrency_limiter.rs

@@ -70,6 +70,17 @@ impl ConcurrencyLimiter {
         }
     }
 
+    pub fn try_acquire(&self, client_id: Arc<String>) -> anyhow::Result<ConcurrencyPermit> {
+        self.tx.try_send(())?;
+        let permit_id = self.tracker.lock().register(client_id.clone());
+        Ok(ConcurrencyPermit {
+            permit_id,
+            rx: self.rx.clone(),
+            limiter: self.clone(),
+            client_id,
+        })
+    }
+
     pub async fn acquire(&self, client_id: Arc<String>) -> ConcurrencyPermit {
         let timer = concurrency_permit_acquire_timer();
         self.tx

crates/isolate/src/isolate.rs

@@ -9,7 +9,6 @@ use std::{
 use anyhow::Context as _;
 use common::{
     knobs::{
-        FUNRUN_INITIAL_PERMIT_TIMEOUT,
         ISOLATE_MAX_ARRAY_BUFFER_TOTAL_SIZE,
         ISOLATE_MAX_USER_HEAP_SIZE,
     },
@@ -296,18 +295,17 @@ impl<RT: Runtime> Isolate<RT> {
                 "IsolateNotClean",
                 "Selected isolate was not clean",
             ))?;
-        // Acquire a concurrency permit without counting it against the timeout.
-        let permit = tokio::select! {
-            biased;
-            permit = self.limiter.acquire(client_id) => permit,
-            // Do not apply a timeout for subfunctions that can't be retried
-            () = self.rt.wait(*FUNRUN_INITIAL_PERMIT_TIMEOUT),
-                    if !environment.is_nested_function() => {
-                anyhow::bail!(ErrorMetadata::rejected_before_execution(
+        // Wait for a permit for subfunctions that cannot be retried, otherwise try to
+        // acquire and fail immediately if there are no permits available.
+        let permit = if environment.is_nested_function() {
+            self.limiter.acquire(client_id).await
+        } else {
+            self.limiter.try_acquire(client_id).context(
+                ErrorMetadata::rejected_before_execution(
                     "InitialPermitTimeoutError",
                     "Couldn't acquire a permit on this funrun",
-                ));
-            }
+                ),
+            )?
         };
         let context_handle = self.handle.new_context_created();
         let mut user_timeout = environment.user_timeout();