perf(nvlink-manager): process partition monitor groups concurrently

crates/api-core/src/cfg/README.md

@@ -350,6 +350,7 @@ extracted identifier contains characters such as `/` or `.`.
 | `allow_insecure` | `bool` | `false` | Skip TLS verification for NMX-C. |
 | `nmx_c_endpoint_port` | `Option<u16>` | — | TCP port for NMX-C endpoints derived from switch NVOS IP. Unset uses the production NMX-C port. |
 | `nmx_c_certificate_rotation` | `NmxCCertificateRotationConfig` | *(default)* | Optional expiry-driven rotation for NMX-C server certificates. |
+| `partition_monitor_max_concurrent_groups` | `NonZeroUsize` | `16` | Maximum number of NMX-C machine groups (chassis or rack) processed concurrently per monitor iteration. Bounds DB pool usage and gRPC fan-out. Must be ≥ 1. |
 
 ### `NmxCCertificateRotationConfig`
 

crates/api-db/src/nvlink_nmxc_endpoints.rs

@@ -34,6 +34,19 @@ pub async fn find_by_chassis_serial(
         .map_err(|e| DatabaseError::new(Q, e))
 }
 
+pub async fn find_by_chassis_serials(
+    txn: impl DbReader<'_>,
+    chassis_serials: &[&str],
+) -> DatabaseResult<Vec<NvlinkNmxcEndpoint>> {
+    const Q: &str =
+        "SELECT chassis_serial, endpoint FROM nvlink_nmxc_endpoints WHERE chassis_serial = ANY($1)";
+    sqlx::query_as(Q)
+        .bind(chassis_serials)
+        .fetch_all(txn)
+        .await
+        .map_err(|e| DatabaseError::new(Q, e))
+}
+
 pub async fn find_all(txn: impl DbReader<'_>) -> DatabaseResult<Vec<NvlinkNmxcEndpoint>> {
     const Q: &str =
         "SELECT chassis_serial, endpoint FROM nvlink_nmxc_endpoints ORDER BY chassis_serial";
@@ -93,3 +106,63 @@ pub async fn update(
         .await
         .map_err(|e| DatabaseError::new(Q, e))
 }
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    #[crate::sqlx_test]
+    async fn find_by_chassis_serials_returns_matching_rows(pool: sqlx::PgPool) {
+        let mut txn = pool.begin().await.unwrap();
+        create(txn.as_mut(), "SN-A", "https://a.example:9370")
+            .await
+            .unwrap();
+        create(txn.as_mut(), "SN-B", "https://b.example:9370")
+            .await
+            .unwrap();
+        create(txn.as_mut(), "SN-C", "https://c.example:9370")
+            .await
+            .unwrap();
+
+        let mut rows = find_by_chassis_serials(txn.as_mut(), &["SN-A", "SN-C"])
+            .await
+            .unwrap();
+        rows.sort_by(|a, b| a.chassis_serial.cmp(&b.chassis_serial));
+        assert_eq!(rows.len(), 2);
+        assert_eq!(rows[0].chassis_serial, "SN-A");
+        assert_eq!(rows[0].endpoint, "https://a.example:9370");
+        assert_eq!(rows[1].chassis_serial, "SN-C");
+        assert_eq!(rows[1].endpoint, "https://c.example:9370");
+
+        txn.rollback().await.unwrap();
+    }
+
+    #[crate::sqlx_test]
+    async fn find_by_chassis_serials_unknown_serials_are_excluded(pool: sqlx::PgPool) {
+        let mut txn = pool.begin().await.unwrap();
+        create(txn.as_mut(), "SN-KNOWN", "https://known.example:9370")
+            .await
+            .unwrap();
+
+        let rows = find_by_chassis_serials(txn.as_mut(), &["SN-KNOWN", "SN-MISSING"])
+            .await
+            .unwrap();
+        assert_eq!(rows.len(), 1);
+        assert_eq!(rows[0].chassis_serial, "SN-KNOWN");
+
+        txn.rollback().await.unwrap();
+    }
+
+    #[crate::sqlx_test]
+    async fn find_by_chassis_serials_empty_slice_returns_empty(pool: sqlx::PgPool) {
+        let mut txn = pool.begin().await.unwrap();
+        create(txn.as_mut(), "SN-X", "https://x.example:9370")
+            .await
+            .unwrap();
+
+        let rows = find_by_chassis_serials(txn.as_mut(), &[]).await.unwrap();
+        assert!(rows.is_empty());
+
+        txn.rollback().await.unwrap();
+    }
+}

crates/nvlink-manager/Cargo.toml

@@ -32,6 +32,7 @@ async-trait = { workspace = true }
 chrono = { workspace = true }
 duration-str = { workspace = true }
 eyre = { workspace = true }
+futures = { workspace = true, features = ["std"] }
 hex = { workspace = true }
 http = { workspace = true }
 librms = { workspace = true }

crates/nvlink-manager/src/config.rs

@@ -54,12 +54,23 @@ pub struct NvLinkConfig {
     /// Optional expiry-driven rotation for NMX-C server certificates.
     #[serde(default)]
     pub nmx_c_certificate_rotation: NmxCCertificateRotationConfig,
+
+    /// Maximum number of NMX-C machine groups (chassis or rack) processed concurrently
+    /// during a partition monitor iteration. Bounds DB pool usage and gRPC fan-out.
+    /// Defaults to 16. Must be non-zero; deserialization rejects 0.
+    #[serde(default = "NvLinkConfig::default_partition_monitor_max_concurrent_groups")]
+    pub partition_monitor_max_concurrent_groups: std::num::NonZeroUsize,
 }
 
 impl NvLinkConfig {
     pub const fn default_monitor_run_interval() -> std::time::Duration {
         std::time::Duration::from_secs(60)
     }
+
+    pub const fn default_partition_monitor_max_concurrent_groups() -> std::num::NonZeroUsize {
+        // SAFETY: 16 is non-zero.
+        unsafe { std::num::NonZeroUsize::new_unchecked(16) }
+    }
 }
 
 #[derive(Clone, Debug, Deserialize, Serialize, PartialEq)]
@@ -131,6 +142,8 @@ impl Default for NvLinkConfig {
             nmx_c_endpoint_port: None,
             allow_insecure: false,
             nmx_c_certificate_rotation: NmxCCertificateRotationConfig::default(),
+            partition_monitor_max_concurrent_groups:
+                Self::default_partition_monitor_max_concurrent_groups(),
         }
     }
 }
@@ -157,6 +170,8 @@ mod test {
                 nmx_c_endpoint_port: None,
                 allow_insecure: true,
                 nmx_c_certificate_rotation: NmxCCertificateRotationConfig::default(),
+                partition_monitor_max_concurrent_groups:
+                    NvLinkConfig::default_partition_monitor_max_concurrent_groups(),
             }
         );
     }
@@ -172,6 +187,14 @@ mod test {
         );
     }
 
+    #[test]
+    fn deserialize_zero_concurrent_groups_is_rejected() {
+        let err = serde_json::from_str::<NvLinkConfig>(
+            r#"{"allow_insecure":false,"partition_monitor_max_concurrent_groups":0}"#,
+        );
+        assert!(err.is_err(), "zero must be rejected by NonZeroUsize");
+    }
+
     #[test]
     fn deserialize_legacy_expiry_warning_window_as_rotation_window() {
         let config: NmxCCertificateRotationConfig =