feat(runtime): configure worker inhibition with direct-routing bypass

docs/fault-tolerance/README.md

@@ -71,6 +71,12 @@ Dynamo provides multiple health check mechanisms:
 
 See [Health Checks](../observability/health-checks.md) for details.
 
+### Local Worker Inhibition
+
+After a request-path failure, a runtime client temporarily removes that worker from its local routing set while service discovery propagates the worker's state. `DYN_RUNTIME_INHIBITED_DURATION_SECS` sets the maximum local inhibition window and is read once during client initialization. The default is `5` seconds; set it to `0` to disable local inhibition. Discovery updates remain authoritative and can restore or remove workers sooner. Direct dispatch bypasses local inhibition and honors an upstream-selected worker as long as it remains present in service discovery.
+
+Changes to the environment variable take effect the next time the process starts.
+
 ### Shadow Engine Failover
 
 For Kubernetes deployments, [Shadow Engine Failover](../kubernetes/shadow-engine-failover.md) can help with same-node recovery from unknown backend engine or software-process failures. It uses GPU Memory Service to keep model weights resident while a standby or replacement engine attaches. It does not preserve in-flight requests or KV cache state, and it does not cover GPU or node loss.
@@ -86,6 +92,7 @@ For Kubernetes deployments, [Shadow Engine Failover](../kubernetes/shadow-engine
 | Decode blocks threshold | `DYN_ACTIVE_DECODE_BLOCKS_THRESHOLD` | unset |
 | Prefill tokens threshold | `DYN_ACTIVE_PREFILL_TOKENS_THRESHOLD` | unset |
 | Prefill tokens fraction threshold | `DYN_ACTIVE_PREFILL_TOKENS_THRESHOLD_FRAC` | unset |
+| Local worker inhibition | `DYN_RUNTIME_INHIBITED_DURATION_SECS` | `5` seconds (`0` disables) |
 
 ## Failure Scenarios and Recovery
 

lib/runtime/src/component/client.rs

@@ -4,7 +4,7 @@
 use std::sync::atomic::{AtomicU64, Ordering};
 use std::{
     collections::{HashMap, HashSet},
-    sync::{Arc, Mutex as StdMutex},
+    sync::{Arc, LazyLock, Mutex as StdMutex},
     time::Duration,
 };
 
@@ -15,6 +15,7 @@ use futures::StreamExt;
 use rand::Rng;
 
 use crate::component::{Endpoint, Instance};
+use crate::config::environment_names::runtime as env_runtime;
 use crate::discovery::{DiscoveryEvent, DiscoveryInstance, DiscoveryInstanceId};
 use crate::traits::DistributedRuntimeProvider;
 
@@ -142,7 +143,31 @@ pub(crate) async fn get_or_create_routing_occupancy_state(
 }
 
 /// Default interval for periodic reconciliation of instance_avail with instance_source
-const DEFAULT_RECONCILE_INTERVAL: Duration = Duration::from_secs(5);
+const DEFAULT_INHIBITED_DURATION_SECS: u64 = 5;
+
+/// Process-wide inhibited duration, resolved from the environment on first client construction.
+static INHIBITED_DURATION: LazyLock<Duration> =
+    LazyLock::new(|| inhibited_duration_from_env(|name| std::env::var(name).ok()));
+
+fn inhibited_duration_from_env(mut lookup: impl FnMut(&str) -> Option<String>) -> Duration {
+    let seconds = match lookup(env_runtime::DYN_RUNTIME_INHIBITED_DURATION_SECS) {
+        None => DEFAULT_INHIBITED_DURATION_SECS,
+        Some(raw) => match raw.parse::<u64>() {
+            Ok(seconds) => seconds,
+            Err(err) => {
+                tracing::warn!(
+                    value = raw,
+                    %err,
+                    "invalid {}; using the default of {} seconds",
+                    env_runtime::DYN_RUNTIME_INHIBITED_DURATION_SECS,
+                    DEFAULT_INHIBITED_DURATION_SECS,
+                );
+                DEFAULT_INHIBITED_DURATION_SECS
+            }
+        },
+    };
+    Duration::from_secs(seconds)
+}
 
 /// Shared endpoint discovery state for a single endpoint query.
 ///
@@ -462,13 +487,14 @@ pub struct Client {
     routing_instances: Arc<RoutingInstancesState>,
     /// Interval for periodic reconciliation of instance_avail with instance_source.
     /// This ensures instances removed via `report_instance_down` are eventually restored.
+    /// A zero value disables local worker inhibition.
     reconcile_interval: Duration,
 }
 
 impl Client {
     // Client with auto-discover instances using key-value store
     pub(crate) async fn new(endpoint: Endpoint) -> Result<Self> {
-        Self::with_reconcile_interval(endpoint, DEFAULT_RECONCILE_INTERVAL).await
+        Self::with_reconcile_interval(endpoint, *INHIBITED_DURATION).await
     }
 
     /// Create a client with a custom reconcile interval.
@@ -575,6 +601,14 @@ impl Client {
 
     /// Mark an instance as down/unavailable
     pub fn report_instance_down(&self, instance_id: u64) {
+        if self.reconcile_interval.is_zero() {
+            tracing::debug!(
+                instance_id,
+                "local worker inhibition is disabled; leaving instance routable"
+            );
+            return;
+        }
+
         self.routing_instances.report_instance_down(instance_id);
         tracing::debug!("inhibiting instance {instance_id}");
     }
@@ -640,6 +674,7 @@ impl Client {
                 }
 
                 tokio::select! {
+                    _ = cancel_token.cancelled() => break,
                     result = rx.changed() => {
                         if let Err(err) = result {
                             tracing::error!(
@@ -648,7 +683,7 @@ impl Client {
                             cancel_token.cancel();
                         }
                     }
-                    _ = tokio::time::sleep(reconcile_interval) => {
+                    _ = tokio::time::sleep(reconcile_interval), if !reconcile_interval.is_zero() => {
                         tracing::trace!(
                             "monitor_instance_source: periodic reconciliation for endpoint={endpoint_id}",
                         );
@@ -757,6 +792,26 @@ mod tests {
     use super::*;
     use crate::{DistributedRuntime, Runtime, distributed::DistributedConfig};
 
+    #[test]
+    fn test_inhibited_duration_from_env() {
+        assert_eq!(
+            inhibited_duration_from_env(|_| None),
+            Duration::from_secs(DEFAULT_INHIBITED_DURATION_SECS)
+        );
+        assert_eq!(
+            inhibited_duration_from_env(|_| Some("17".to_string())),
+            Duration::from_secs(17)
+        );
+        assert_eq!(
+            inhibited_duration_from_env(|_| Some("0".to_string())),
+            Duration::ZERO
+        );
+        assert_eq!(
+            inhibited_duration_from_env(|_| Some("invalid".to_string())),
+            Duration::from_secs(DEFAULT_INHIBITED_DURATION_SECS)
+        );
+    }
+
     /// Test that instances removed via report_instance_down are restored after
     /// the reconciliation interval elapses.
     #[tokio::test]
@@ -804,6 +859,35 @@ mod tests {
         rt.shutdown();
     }
 
+    /// A zero inhibited duration disables local worker inhibition.
+    #[tokio::test]
+    async fn test_zero_inhibited_duration_leaves_instance_routable() {
+        let rt = Runtime::from_current().unwrap();
+        let drt = DistributedRuntime::new(rt.clone(), DistributedConfig::process_local())
+            .await
+            .unwrap();
+        let ns = drt
+            .namespace("test_disabled_inhibition".to_string())
+            .unwrap();
+        let component = ns.component("test_component".to_string()).unwrap();
+        let endpoint = component.endpoint("test_endpoint".to_string());
+
+        let client = Client::with_reconcile_interval(endpoint, Duration::ZERO)
+            .await
+            .unwrap();
+
+        client.override_instance_avail(vec![1, 2, 3]);
+        client.report_instance_down(2);
+
+        assert_eq!(
+            client.instance_ids_avail(),
+            vec![1, 2, 3],
+            "a zero inhibited duration should leave the reported instance routable"
+        );
+
+        rt.shutdown();
+    }
+
     /// Test that report_instance_down correctly removes an instance from instance_avail.
     #[tokio::test]
     async fn test_report_instance_down() {

lib/runtime/src/config/environment_names.rs

@@ -89,6 +89,9 @@ pub mod runtime {
     pub const DYN_RUNTIME_GRACEFUL_SHUTDOWN_TIMEOUT_SECS: &str =
         "DYN_RUNTIME_GRACEFUL_SHUTDOWN_TIMEOUT_SECS";
 
+    /// Maximum duration for local worker inhibition after a request failure. Zero disables it.
+    pub const DYN_RUNTIME_INHIBITED_DURATION_SECS: &str = "DYN_RUNTIME_INHIBITED_DURATION_SECS";
+
     /// Enable Tokio task poll-time histogram (calls enable_metrics_poll_time_histogram on builder).
     /// Set to "1", "true", or "yes" to enable. Adds ~2× overhead of Instant::now() per task poll.
     pub const DYN_ENABLE_POLL_HISTOGRAM: &str = "DYN_ENABLE_POLL_HISTOGRAM";
@@ -761,6 +764,7 @@ mod tests {
             runtime::DYN_RUNTIME_NUM_WORKER_THREADS,
             runtime::DYN_RUNTIME_MAX_BLOCKING_THREADS,
             runtime::DYN_RUNTIME_GRACEFUL_SHUTDOWN_TIMEOUT_SECS,
+            runtime::DYN_RUNTIME_INHIBITED_DURATION_SECS,
             runtime::system::DYN_SYSTEM_ENABLED,
             runtime::system::DYN_SYSTEM_HOST,
             runtime::system::DYN_SYSTEM_PORT,

lib/runtime/src/pipeline/network/egress/push_router.rs

@@ -812,19 +812,9 @@ where
     where
         F: FnOnce(&mut T, u64) -> anyhow::Result<M>,
     {
-        // When fault detection is disabled, check the raw discovery list
-        // (not filtered by report_instance_down) so transient failures
-        // don't poison the instance for subsequent retries.
-        let found = {
-            if self.fault_detection_enabled {
-                let routing_instances = self.client.routing_instances();
-                routing_instances.routable_ids().contains(&instance_id)
-            } else {
-                self.client.instance_ids().contains(&instance_id)
-            }
-        };
-
-        if !found {
+        // Direct dispatch honors the caller-selected worker while it remains in discovery.
+        // Local inhibition only filters worker selection owned by this router.
+        if !self.client.instance_ids().contains(&instance_id) {
             return Err(DynamoError::builder()
                 .error_type(ErrorType::CannotConnect)
                 .message(format!(
@@ -2491,6 +2481,70 @@ mod tests {
         rt.shutdown();
     }
 
+    /// Direct dispatch honors an upstream-selected worker even after local inhibition.
+    #[tokio::test]
+    async fn direct_dispatch_ignores_local_inhibition() {
+        let rt = Runtime::from_current().unwrap();
+        let drt = DistributedRuntime::new(rt.clone(), DistributedConfig::process_local())
+            .await
+            .unwrap();
+        let ns = drt
+            .namespace("test_direct_bypasses_inhibition".to_string())
+            .unwrap();
+        let component = ns.component("test_component".to_string()).unwrap();
+        let endpoint = component.endpoint("test_endpoint".to_string());
+        let client = endpoint.client().await.unwrap();
+        endpoint.register_endpoint_instance().await.unwrap();
+        let instance_id = client.wait_for_instances().await.unwrap()[0].id();
+
+        // KV routing selects upstream and dispatches through PushRouter::direct.
+        let router = PushRouter::<u64, TestResponse>::from_client(client.clone(), RouterMode::KV)
+            .await
+            .unwrap();
+
+        client.report_instance_down(instance_id);
+        assert!(
+            !client.instance_ids_avail().contains(&instance_id),
+            "precondition: worker should be locally inhibited"
+        );
+
+        let result = router
+            .direct_within_prepared(
+                SingleIn::new(42),
+                instance_id,
+                None,
+                |_, selected_instance_id| {
+                    assert_eq!(selected_instance_id, instance_id);
+                    Err::<(), _>(anyhow::anyhow!("direct prepare sentinel"))
+                },
+            )
+            .await;
+        let error = match result {
+            Ok(_) => panic!("direct dispatch should reach request preparation"),
+            Err(error) => error,
+        };
+        assert_eq!(error.to_string(), "direct prepare sentinel");
+
+        let missing_instance_id = instance_id.wrapping_add(1);
+        let result = router
+            .direct_within_prepared(SingleIn::new(42), missing_instance_id, None, |_, _| {
+                Ok::<(), anyhow::Error>(())
+            })
+            .await;
+        let error = match result {
+            Ok(_) => panic!("direct dispatch should reject a worker absent from discovery"),
+            Err(error) => error,
+        };
+        assert!(
+            error
+                .to_string()
+                .contains(&format!("instance_id={missing_instance_id} not found")),
+            "unexpected missing-worker error: {error}"
+        );
+
+        rt.shutdown();
+    }
+
     /// When the router selects an instance that has deregistered between selection
     /// and transport resolution, it should fall back to another available instance
     /// rather than returning a 500 error.