Revert "Remove timeout for acquiring an isolate permit, fail immediately instead" (#42174)

GitOrigin-RevId: dd33425af61ac7c2f95d407232551e9f3ba31465

Author: emmaling27

crates/common/src/knobs.rs

@@ -1036,6 +1036,11 @@ 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,17 +70,6 @@ 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,6 +9,7 @@ use std::{
 use anyhow::Context as _;
 use common::{
     knobs::{
+        FUNRUN_INITIAL_PERMIT_TIMEOUT,
         ISOLATE_MAX_ARRAY_BUFFER_TOTAL_SIZE,
         ISOLATE_MAX_USER_HEAP_SIZE,
     },
@@ -295,17 +296,18 @@ impl<RT: Runtime> Isolate<RT> {
                 "IsolateNotClean",
                 "Selected isolate was not clean",
             ))?;
-        // 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(
+        // 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(
                     "InitialPermitTimeoutError",
                     "Couldn't acquire a permit on this funrun",
-                ),
-            )?
+                ));
+            }
         };
         let context_handle = self.handle.new_context_created();
         let mut user_timeout = environment.user_timeout();