feat(dpf): gate the outdated-DPU scan on DPFOperatorConfig readiness

crates/dpf/src/repository/kube.rs

@@ -705,6 +705,17 @@ impl K8sConfigRepository for KubeRepository {
 
 #[async_trait]
 impl DpfOperatorConfigRepository for KubeRepository {
+    async fn get(
+        &self,
+        name: &str,
+        namespace: &str,
+    ) -> Result<Option<crate::crds::dpfoperatorconfigs_generated::DPFOperatorConfig>, DpfError>
+    {
+        use crate::crds::dpfoperatorconfigs_generated::DPFOperatorConfig;
+        let api: Api<DPFOperatorConfig> = Api::namespaced(self.client.clone(), namespace);
+        Ok(api.get_opt(name).await?)
+    }
+
     async fn patch(
         &self,
         name: &str,

crates/dpf/src/repository/traits.rs

@@ -25,6 +25,7 @@ use async_trait::async_trait;
 
 use crate::crds::bfbs_generated::BFB;
 use crate::crds::bluefieldsoftwares_generated::BlueFieldSoftware;
+use crate::crds::dpfoperatorconfigs_generated::DPFOperatorConfig;
 use crate::crds::dpuclusters_generated::DPUCluster;
 use crate::crds::dpudeployments_generated::DPUDeployment;
 use crate::crds::dpudevices_generated::DPUDevice;
@@ -296,6 +297,9 @@ pub trait K8sConfigRepository: Send + Sync {
 /// Repository for DPFOperatorConfig resources.
 #[async_trait]
 pub trait DpfOperatorConfigRepository: Send + Sync {
+    async fn get(&self, name: &str, namespace: &str)
+    -> Result<Option<DPFOperatorConfig>, DpfError>;
+
     async fn patch(
         &self,
         name: &str,

crates/dpf/src/sdk.rs

@@ -2248,7 +2248,55 @@ impl<R: DpuRepository, L> DpfSdk<R, L> {
     }
 }
 
-impl<R: DpuDeploymentRepository + DpuRepository, L> DpfSdk<R, L> {
+/// Name of the singleton DPFOperatorConfig, as created by helm-prereqs and by the
+/// manual install in `docs/manuals/dpf.md`.
+const DPF_OPERATOR_CONFIG_NAME: &str = "dpfoperatorconfig";
+
+impl<R: DpuDeploymentRepository + DpuRepository + DpfOperatorConfigRepository, L> DpfSdk<R, L> {
+    /// Whether the DPF operator reports `Ready=True` at its current generation.
+    ///
+    /// Fails closed: absent, unreconciled, or condition-less all read as not
+    /// ready, so callers that gate disruptive work skip rather than guess.
+    async fn dpf_operator_config_is_ready(&self) -> Result<bool, DpfError> {
+        let config = DpfOperatorConfigRepository::get(
+            &*self.repo,
+            DPF_OPERATOR_CONFIG_NAME,
+            &self.namespace,
+        )
+        .await?;
+
+        let Some(config) = config else {
+            tracing::info!(
+                name = DPF_OPERATOR_CONFIG_NAME,
+                namespace = %self.namespace,
+                "DPFOperatorConfig not found; treating DPF as not ready"
+            );
+            return Ok(false);
+        };
+
+        let ready = config
+            .status
+            .as_ref()
+            .and_then(|status| status.conditions.as_ref())
+            .and_then(|conditions| conditions.iter().find(|c| c.type_ == "Ready"))
+            .is_some_and(|condition| {
+                condition.status == "True"
+                    && observed_generation_is_current(
+                        condition.observed_generation,
+                        config.metadata.generation,
+                    )
+            });
+
+        if !ready {
+            tracing::info!(
+                name = DPF_OPERATOR_CONFIG_NAME,
+                namespace = %self.namespace,
+                "DPFOperatorConfig is not Ready; treating DPF as not ready"
+            );
+        }
+        Ok(ready)
+    }
+
     /// Find DPUs whose installed BFB, BlueFieldSoftware, or `spec.dpuFlavor` no
     /// longer matches the values declared on the DPUDeployment that owns them.
     ///
@@ -2286,6 +2334,14 @@ impl<R: DpuDeploymentRepository + DpuRepository, L> DpfSdk<R, L> {
         &self,
         dpu_label_selector: Option<&str>,
     ) -> Result<Vec<DpuMismatch>, DpfError> {
+        // A DPF upgrade republishes the CRs this scan reads, so mid-upgrade a DPU
+        // can look outdated against a deployment that is still settling. Report
+        // nothing until the operator says it is Ready, so an upgrade never
+        // triggers reprovisioning on its own.
+        if !self.dpf_operator_config_is_ready().await? {
+            return Ok(vec![]);
+        }
+
         let deployments = DpuDeploymentRepository::list(&*self.repo, &self.namespace).await?;
         let ready_deployments: HashMap<String, &DPUDeployment> = deployments
             .iter()
@@ -2451,6 +2507,16 @@ fn dpu_deployment_is_ready(d: &DPUDeployment) -> bool {
     cond.status == "True" && cond.observed_generation == Some(generation)
 }
 
+/// True when a condition's `observedGeneration` matches the object's, or when
+/// either is absent. Stamping it is optional, and demanding it would leave the
+/// object permanently unready against an operator that omits it.
+fn observed_generation_is_current(observed: Option<i64>, generation: Option<i64>) -> bool {
+    match (observed, generation) {
+        (Some(observed), Some(generation)) => observed == generation,
+        _ => true,
+    }
+}
+
 impl<R: DpuNodeMaintenanceRepository, L> DpfSdk<R, L> {
     /// Release the hold on a DPU node maintenance.
     /// If the DpuNodeMaintenance CR doesn't exist, this is a no-op
@@ -3815,6 +3881,15 @@ mod tests {
 
     #[async_trait]
     impl crate::repository::DpfOperatorConfigRepository for SdkMock {
+        async fn get(
+            &self,
+            _name: &str,
+            _ns: &str,
+        ) -> Result<Option<crate::crds::dpfoperatorconfigs_generated::DPFOperatorConfig>, DpfError>
+        {
+            Ok(None)
+        }
+
         async fn patch(&self, _: &str, _: &str, _: serde_json::Value) -> Result<(), DpfError> {
             Ok(())
         }
@@ -4514,6 +4589,15 @@ mod tests {
 
     #[async_trait]
     impl crate::repository::DpfOperatorConfigRepository for SecretTrackingMock {
+        async fn get(
+            &self,
+            _name: &str,
+            _ns: &str,
+        ) -> Result<Option<crate::crds::dpfoperatorconfigs_generated::DPFOperatorConfig>, DpfError>
+        {
+            Ok(None)
+        }
+
         async fn patch(&self, _: &str, _: &str, _: serde_json::Value) -> Result<(), DpfError> {
             Ok(())
         }

crates/dpf/src/test/helpers.rs

@@ -328,6 +328,15 @@ impl K8sConfigRepository for ConfigMock {
 
 #[async_trait]
 impl DpfOperatorConfigRepository for ConfigMock {
+    async fn get(
+        &self,
+        _name: &str,
+        _ns: &str,
+    ) -> Result<Option<crate::crds::dpfoperatorconfigs_generated::DPFOperatorConfig>, DpfError>
+    {
+        Ok(None)
+    }
+
     async fn patch(&self, _: &str, _: &str, _: serde_json::Value) -> Result<(), DpfError> {
         Ok(())
     }

crates/dpf/src/test/maintenance_flow.rs

@@ -255,6 +255,15 @@ impl K8sConfigRepository for MaintenanceFlowMock {
 
 #[async_trait]
 impl DpfOperatorConfigRepository for MaintenanceFlowMock {
+    async fn get(
+        &self,
+        _name: &str,
+        _ns: &str,
+    ) -> Result<Option<crate::crds::dpfoperatorconfigs_generated::DPFOperatorConfig>, DpfError>
+    {
+        Ok(None)
+    }
+
     async fn patch(&self, _: &str, _: &str, _: serde_json::Value) -> Result<(), DpfError> {
         Ok(())
     }

crates/dpf/src/test/sdk_device_registration.rs

@@ -216,6 +216,15 @@ impl K8sConfigRepository for DeviceRegistrationMock {
 
 #[async_trait]
 impl DpfOperatorConfigRepository for DeviceRegistrationMock {
+    async fn get(
+        &self,
+        _name: &str,
+        _ns: &str,
+    ) -> Result<Option<crate::crds::dpfoperatorconfigs_generated::DPFOperatorConfig>, DpfError>
+    {
+        Ok(None)
+    }
+
     async fn patch(&self, _: &str, _: &str, _: serde_json::Value) -> Result<(), DpfError> {
         Ok(())
     }

crates/dpf/src/test/sdk_initialization.rs

@@ -452,6 +452,15 @@ impl K8sConfigRepository for InitializationMock {
 
 #[async_trait]
 impl DpfOperatorConfigRepository for InitializationMock {
+    async fn get(
+        &self,
+        _name: &str,
+        _ns: &str,
+    ) -> Result<Option<crate::crds::dpfoperatorconfigs_generated::DPFOperatorConfig>, DpfError>
+    {
+        Ok(None)
+    }
+
     async fn patch(&self, _: &str, _: &str, _: serde_json::Value) -> Result<(), DpfError> {
         Ok(())
     }

crates/dpf/src/test/sdk_maintenance_hold.rs

@@ -123,6 +123,15 @@ impl K8sConfigRepository for MaintenanceHoldMock {
 
 #[async_trait]
 impl DpfOperatorConfigRepository for MaintenanceHoldMock {
+    async fn get(
+        &self,
+        _name: &str,
+        _ns: &str,
+    ) -> Result<Option<crate::crds::dpfoperatorconfigs_generated::DPFOperatorConfig>, DpfError>
+    {
+        Ok(None)
+    }
+
     async fn patch(&self, _: &str, _: &str, _: serde_json::Value) -> Result<(), DpfError> {
         Ok(())
     }

crates/dpf/src/test/sdk_outdated_dpu.rs

@@ -29,12 +29,14 @@ use async_trait::async_trait;
 use dashmap::DashMap;
 use kube::core::ObjectMeta;
 
+use crate::crds::dpfoperatorconfigs_generated::DPFOperatorConfig;
 use crate::crds::dpudeployments_generated::DPUDeployment;
 use crate::crds::dpus_generated::DPU;
 use crate::crds::dpuservicetemplates_generated::DPUServiceTemplate;
 use crate::error::DpfError;
 use crate::repository::{
-    DpuDeploymentRepository, DpuRepository, DpuServiceTemplateRepository, K8sConfigRepository,
+    DpfOperatorConfigRepository, DpuDeploymentRepository, DpuRepository,
+    DpuServiceTemplateRepository, K8sConfigRepository,
 };
 use crate::sdk::DpfSdkBuilder;
 
@@ -48,6 +50,7 @@ const OWNED_BY_LABEL: &str = "svc.dpu.nvidia.com/owned-by-dpudeployment";
 struct OutdatedDpuMock {
     dpus: Arc<DashMap<String, DPU>>,
     deployments: Arc<DashMap<String, DPUDeployment>>,
+    operator_config: Arc<DashMap<String, DPFOperatorConfig>>,
 }
 
 impl OutdatedDpuMock {
@@ -92,6 +95,16 @@ impl DpuRepository for OutdatedDpuMock {
     }
 }
 
+#[async_trait]
+impl DpfOperatorConfigRepository for OutdatedDpuMock {
+    async fn get(&self, name: &str, _ns: &str) -> Result<Option<DPFOperatorConfig>, DpfError> {
+        Ok(self.operator_config.get(name).map(|c| c.clone()))
+    }
+    async fn patch(&self, _: &str, _: &str, _: serde_json::Value) -> Result<(), DpfError> {
+        Ok(())
+    }
+}
+
 #[async_trait]
 impl DpuDeploymentRepository for OutdatedDpuMock {
     async fn get(&self, name: &str, _ns: &str) -> Result<Option<DPUDeployment>, DpfError> {

crates/dpf/src/test/sdk_provisioning_flow.rs

@@ -216,6 +216,15 @@ impl K8sConfigRepository for ProvisioningFlowMock {
 
 #[async_trait]
 impl DpfOperatorConfigRepository for ProvisioningFlowMock {
+    async fn get(
+        &self,
+        _name: &str,
+        _ns: &str,
+    ) -> Result<Option<crate::crds::dpfoperatorconfigs_generated::DPFOperatorConfig>, DpfError>
+    {
+        Ok(None)
+    }
+
     async fn patch(&self, _: &str, _: &str, _: serde_json::Value) -> Result<(), DpfError> {
         Ok(())
     }

crates/dpf/src/test/sdk_reboot_annotation.rs

@@ -138,6 +138,15 @@ impl K8sConfigRepository for RebootAnnotationMock {
 
 #[async_trait]
 impl DpfOperatorConfigRepository for RebootAnnotationMock {
+    async fn get(
+        &self,
+        _name: &str,
+        _ns: &str,
+    ) -> Result<Option<crate::crds::dpfoperatorconfigs_generated::DPFOperatorConfig>, DpfError>
+    {
+        Ok(None)
+    }
+
     async fn patch(&self, _: &str, _: &str, _: serde_json::Value) -> Result<(), DpfError> {
         Ok(())
     }