fix(driver-kubernetes): wait for operator namespace listing before workspace checks

crates/openshell-core/src/dynamic_string_allowlist.rs

@@ -3,11 +3,20 @@
 
 use std::collections::BTreeSet;
 use std::sync::{Arc, RwLock};
+use std::time::Duration;
+
+use tokio::sync::watch;
 
 /// Thread-safe dynamic allowlist of strings shared across component boundaries.
+///
+/// An allowlist created with [`Self::new`] starts unsynced: it is empty only
+/// because its source has not been read yet. The first [`Self::replace`] (or
+/// an explicit [`Self::mark_synced`]) marks it synced, so callers can tell a
+/// missing entry from a list that has not loaded.
 #[derive(Debug, Clone)]
 pub struct DynamicStringAllowlist {
     inner: Arc<RwLock<BTreeSet<String>>>,
+    synced: Arc<watch::Sender<bool>>,
 }
 
 impl DynamicStringAllowlist {
@@ -23,22 +32,50 @@ impl DynamicStringAllowlist {
             .unwrap_or_else(std::sync::PoisonError::into_inner)
     }
 
+    /// Create an empty, unsynced allowlist.
     #[must_use]
     pub fn new() -> Self {
         Self {
             inner: Arc::new(RwLock::new(BTreeSet::new())),
+            synced: Arc::new(watch::Sender::new(false)),
         }
     }
 
+    /// Create a synced allowlist from a known set.
     #[must_use]
     pub fn from_set(set: BTreeSet<String>) -> Self {
         Self {
             inner: Arc::new(RwLock::new(set)),
+            synced: Arc::new(watch::Sender::new(true)),
         }
     }
 
+    /// Replace the whole set from an authoritative listing and mark the
+    /// allowlist synced.
     pub fn replace(&self, new_set: BTreeSet<String>) {
         *self.write_guard() = new_set;
+        self.mark_synced();
+    }
+
+    /// Mark the allowlist synced without changing its contents.
+    pub fn mark_synced(&self) {
+        self.synced
+            .send_if_modified(|synced| !std::mem::replace(synced, true));
+    }
+
+    /// Whether the allowlist has received its first authoritative listing.
+    #[must_use]
+    pub fn is_synced(&self) -> bool {
+        *self.synced.borrow()
+    }
+
+    /// Wait up to `timeout` for the allowlist to sync. Returns whether it is
+    /// synced.
+    pub async fn wait_synced(&self, timeout: Duration) -> bool {
+        let mut synced = self.synced.subscribe();
+        tokio::time::timeout(timeout, synced.wait_for(|synced| *synced))
+            .await
+            .is_ok_and(|result| result.is_ok())
     }
 
     pub fn merge(&self, additional: &BTreeSet<String>) {
@@ -73,3 +110,47 @@ impl Default for DynamicStringAllowlist {
         Self::new()
     }
 }
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    #[test]
+    fn new_allowlist_is_unsynced_until_replaced() {
+        let allowlist = DynamicStringAllowlist::new();
+        assert!(!allowlist.is_synced());
+
+        allowlist.insert("ns1".to_string());
+        assert!(!allowlist.is_synced());
+
+        allowlist.replace(BTreeSet::from(["ns1".to_string()]));
+        assert!(allowlist.is_synced());
+    }
+
+    #[test]
+    fn from_set_and_mark_synced_report_synced() {
+        assert!(DynamicStringAllowlist::from_set(BTreeSet::new()).is_synced());
+
+        let allowlist = DynamicStringAllowlist::new();
+        allowlist.mark_synced();
+        assert!(allowlist.is_synced());
+        assert!(allowlist.read().is_empty());
+    }
+
+    #[tokio::test]
+    async fn wait_synced_times_out_while_unsynced() {
+        let allowlist = DynamicStringAllowlist::new();
+        assert!(!allowlist.wait_synced(Duration::from_millis(20)).await);
+    }
+
+    #[tokio::test]
+    async fn wait_synced_returns_when_a_clone_is_replaced() {
+        let allowlist = DynamicStringAllowlist::new();
+        let writer = allowlist.clone();
+        let waiter =
+            tokio::spawn(async move { allowlist.wait_synced(Duration::from_secs(30)).await });
+        tokio::task::yield_now().await;
+        writer.replace(BTreeSet::from(["ns1".to_string()]));
+        assert!(waiter.await.unwrap());
+    }
+}

crates/openshell-driver-kubernetes/README.md

@@ -27,7 +27,10 @@ workspace namespace modes via `workspace_mode`:
   through exactly one source: either a label selector
   (`operator_namespace_label`) or a drop-in allowlist file
   (`operator_namespace_file`). Sandbox creation fails closed if the workspace
-  namespace is not in the current allowlist. Workspace deletion only removes
+  namespace is not in the current allowlist. With a label selector, workspace
+  checks that run before the watcher's first namespace listing wait up to 10
+  seconds for it and then return `Unavailable`, so a gateway restart does not
+  reject labeled workspaces as unknown. Workspace deletion only removes
   gateway state; it never deletes or otherwise accesses the operator-managed
   Kubernetes namespace.
 

crates/openshell-driver-kubernetes/src/driver.rs

@@ -267,6 +267,9 @@ impl From<KubernetesDriverError> for openshell_core::ComputeDriverError {
 /// This prevents gRPC handlers from blocking indefinitely when the k8s
 /// API server is unreachable or slow.
 const KUBE_API_TIMEOUT: Duration = Duration::from_secs(30);
+/// How long an operator-mode workspace check waits for the namespace label
+/// watcher's first full listing before reporting the driver unavailable.
+const OPERATOR_ALLOWLIST_SYNC_TIMEOUT: Duration = Duration::from_secs(10);
 const OPENSHIFT_SCC_ALLOCATOR_POLL_INTERVAL: Duration = Duration::from_millis(100);
 fn admission_error(error: tonic::Status) -> KubernetesDriverError {
     match error.code() {
@@ -907,6 +910,7 @@ impl KubernetesComputeDriver {
             .ok_or_else(|| tonic::Status::internal("TokenReview response missing status"))?;
         let identity = token_review_identity(&status, &self.config.service_account_name)?
             .ok_or_else(|| tonic::Status::unauthenticated("sandbox credential was not accepted"))?;
+        self.wait_for_operator_allowlist().await?;
         if !self.accepts_auth_namespace(&identity.namespace) {
             return Err(tonic::Status::permission_denied(
                 "sandbox credential namespace is not accepted by the driver",
@@ -982,6 +986,25 @@ impl KubernetesComputeDriver {
         self.operator_allowlist.as_ref()
     }
 
+    /// Wait for the operator namespace allowlist's first full listing.
+    ///
+    /// The label watcher fills the allowlist in the background after the
+    /// driver starts. Checking a workspace before that listing arrives would
+    /// reject a labeled namespace as unknown, so callers wait here first and
+    /// report `Unavailable` if the listing is still missing.
+    pub async fn wait_for_operator_allowlist(&self) -> Result<(), tonic::Status> {
+        let Some(allowlist) = self.operator_allowlist.as_ref() else {
+            return Ok(());
+        };
+        if allowlist.wait_synced(OPERATOR_ALLOWLIST_SYNC_TIMEOUT).await {
+            Ok(())
+        } else {
+            Err(tonic::Status::unavailable(
+                "operator namespace allowlist has not finished its initial namespace listing",
+            ))
+        }
+    }
+
     pub fn default_image(&self) -> &str {
         &self.config.default_image
     }
@@ -1675,6 +1698,7 @@ impl KubernetesComputeDriver {
         )?;
         self.validate_workspace_namespace(&sandbox.workspace)
             .map_err(|error| tonic::Status::failed_precondition(error.to_string()))?;
+        self.wait_for_operator_allowlist().await?;
         let namespace = self
             .config
             .namespace_for_workspace(&sandbox.workspace, self.operator_allowlist.as_ref())
@@ -1916,6 +1940,9 @@ impl KubernetesComputeDriver {
             WorkspaceMode::Shared => self.config.namespace.clone(),
             WorkspaceMode::Managed => self.ensure_namespace(workspace).await?,
             WorkspaceMode::Operator => {
+                self.wait_for_operator_allowlist()
+                    .await
+                    .map_err(admission_error)?;
                 if let Some(ref allowlist) = self.operator_allowlist
                     && !allowlist.contains(workspace)
                 {
@@ -7451,6 +7478,9 @@ fn spawn_namespace_file_watcher(
                 error = %err,
                 "failed to load initial operator namespace file, allowlist empty"
             );
+            // The file is the only source, so an unreadable file is an empty
+            // allowlist rather than one still loading.
+            allowlist.mark_synced();
         }
     }
 
@@ -11746,6 +11776,72 @@ mod tests {
         }
     }
 
+    fn operator_driver_with_allowlist(
+        allowlist: OperatorNamespaceAllowlist,
+    ) -> crate::ComputeDriverService {
+        let mut driver = KubernetesComputeDriver::new_for_test(KubernetesComputeConfig {
+            workspace_mode: WorkspaceMode::Operator,
+            operator_namespace_label: Some("openshell.ai/workspace=true".into()),
+            ..Default::default()
+        });
+        driver.operator_allowlist = Some(allowlist);
+        crate::ComputeDriverService::new(driver)
+    }
+
+    #[tokio::test]
+    async fn ensure_workspace_waits_for_operator_allowlist_initial_listing() {
+        let allowlist = OperatorNamespaceAllowlist::new();
+        let service = operator_driver_with_allowlist(allowlist.clone());
+
+        let request = tokio::spawn(async move { call_ensure_workspace(&service).await });
+        tokio::task::yield_now().await;
+        allowlist.replace(BTreeSet::from(["team-a".to_string()]));
+
+        request
+            .await
+            .unwrap()
+            .expect("a labeled workspace should be accepted once the listing arrives");
+    }
+
+    #[tokio::test(start_paused = true)]
+    async fn ensure_workspace_reports_unlisted_operator_allowlist_unavailable() {
+        let service = operator_driver_with_allowlist(OperatorNamespaceAllowlist::new());
+
+        let error = call_ensure_workspace(&service)
+            .await
+            .expect_err("an allowlist without its first listing cannot admit the workspace");
+
+        assert_eq!(error.code(), tonic::Code::Unavailable);
+    }
+
+    #[tokio::test]
+    async fn ensure_workspace_denies_workspace_missing_from_listed_operator_allowlist() {
+        let allowlist = OperatorNamespaceAllowlist::new();
+        allowlist.replace(BTreeSet::from(["team-b".to_string()]));
+        let service = operator_driver_with_allowlist(allowlist);
+
+        let error = call_ensure_workspace(&service)
+            .await
+            .expect_err("an unlabeled workspace should stay denied");
+
+        assert_eq!(error.code(), tonic::Code::PermissionDenied);
+    }
+
+    #[tokio::test]
+    async fn unreadable_operator_namespace_file_leaves_a_listed_empty_allowlist() {
+        let path = std::env::temp_dir().join(format!(
+            "openshell-missing-namespaces-{}.json",
+            std::process::id()
+        ));
+        let allowlist = OperatorNamespaceAllowlist::new();
+        let (_shutdown_tx, shutdown_rx) = tokio::sync::watch::channel(false);
+
+        spawn_namespace_file_watcher(path, allowlist.clone(), shutdown_rx);
+
+        assert!(allowlist.is_synced());
+        assert!(allowlist.read().is_empty());
+    }
+
     #[tokio::test]
     async fn managed_pods_reference_generation_image_pull_secrets() {
         let names = SandboxRuntimeNames::for_generation("sandbox-1", "gen7");

crates/openshell-driver-kubernetes/src/grpc.rs

@@ -285,6 +285,7 @@ impl ComputeDriver for ComputeDriverService {
                         )?;
                     }
                     WorkspaceMode::Operator => {
+                        self.driver.wait_for_operator_allowlist().await?;
                         if let Some(allowlist) = self.driver.operator_allowlist()
                             && !allowlist.contains(&workspace)
                         {