scheduler: feat: transfers need only network

backend/gradient-db/src/task_board/dep_counts.rs

@@ -49,7 +49,8 @@ crate::sql! {
                                         WHERE derivation = ANY($1::uuid[]))) \
                    ORDER BY id FOR NO KEY UPDATE) locked \
              WHERE e.id = locked.id",
-        params = [DerivationIds(64)];
+        params = [DerivationIds(64)],
+        tier = Bulk;
 }
 
 pub async fn bump_graph_version<C: ConnectionTrait>(

backend/gradient-pool/src/worker_caps.rs

@@ -43,10 +43,13 @@ impl WorkerCaps {
             .builds
             .iter()
             .any(|b| build_needs_network(b.kind, b.is_fixed_output));
-        self.can_build(
-            &job.requirement.architecture,
-            &job.requirement.required_features,
-        ) && (!needs_network || self.capabilities.network)
+        let only_transfers = job.builds.iter().all(|b| b.kind != BuildSpecKind::Build);
+        let runs_here = only_transfers
+            || self.can_build(
+                &job.requirement.architecture,
+                &job.requirement.required_features,
+            );
+        runs_here && (!needs_network || self.capabilities.network)
     }
 
     pub fn can_eval(&self, job: &FlakeJob) -> bool {
@@ -97,6 +100,70 @@ mod tests {
         assert!(builder.can_build("x86_64-linux", &[]));
     }
 
+    fn fixed_output_job() -> BuildJob {
+        job_of(BuildSpecKind::Build)
+    }
+
+    fn job_of(kind: BuildSpecKind) -> BuildJob {
+        BuildJob {
+            builds: vec![gradient_wire::types::BuildSpec {
+                build_id: "b1".into(),
+                drv_path: String::new(),
+                kind,
+                is_fixed_output: true,
+                outputs: Vec::new(),
+                timeout_secs: None,
+                max_silent_secs: None,
+                build_cores: None,
+            }],
+            requirement: gradient_wire::types::BuildRequirement {
+                architecture: "x86_64-linux".into(),
+                required_features: Vec::new(),
+            },
+        }
+    }
+
+    #[test]
+    fn a_fixed_output_build_needs_network_and_build() {
+        let with = |network: bool, build: bool| WorkerCaps {
+            architectures: vec!["x86_64-linux".into()],
+            capabilities: GradientCapabilities {
+                network,
+                build,
+                ..Default::default()
+            },
+            ..Default::default()
+        };
+        let job = fixed_output_job();
+        assert!(with(true, true).can_run_build(&job));
+        assert!(
+            !with(true, false).can_run_build(&job),
+            "network without build"
+        );
+        assert!(
+            !with(false, true).can_run_build(&job),
+            "build without network"
+        );
+    }
+
+    #[test]
+    fn a_substitute_or_download_needs_only_network() {
+        let network_only = WorkerCaps {
+            capabilities: GradientCapabilities {
+                network: true,
+                ..Default::default()
+            },
+            ..Default::default()
+        };
+        for kind in [BuildSpecKind::Substitute, BuildSpecKind::Download] {
+            assert!(network_only.can_run_build(&job_of(kind)), "{kind:?}");
+            assert!(
+                !WorkerCaps::default().can_run_build(&job_of(kind)),
+                "{kind:?} without network"
+            );
+        }
+    }
+
     fn repository(url: &str) -> FlakeSource {
         FlakeSource::Repository {
             url: url.into(),

backend/gradient-scheduler/src/assessment_memo.rs

@@ -75,14 +75,14 @@ impl AssessmentMemo {
 }
 
 pub(crate) fn caps_fingerprint(caps: &Caps) -> u64 {
-    let mut workers: Vec<(Vec<&str>, Vec<&str>, bool)> = caps
+    let mut workers: Vec<(Vec<&str>, Vec<&str>, bool, bool)> = caps
         .iter()
-        .map(|(archs, feats, network)| {
-            let mut archs: Vec<&str> = archs.iter().map(String::as_str).collect();
-            let mut feats: Vec<&str> = feats.iter().map(String::as_str).collect();
+        .map(|w| {
+            let mut archs: Vec<&str> = w.architectures.iter().map(String::as_str).collect();
+            let mut feats: Vec<&str> = w.features.iter().map(String::as_str).collect();
             archs.sort_unstable();
             feats.sort_unstable();
-            (archs, feats, *network)
+            (archs, feats, w.network, w.build)
         })
         .collect();
     workers.sort_unstable();
@@ -97,7 +97,12 @@ mod tests {
     use super::*;
 
     fn caps() -> Vec<BuildWorkerCaps> {
-        vec![(vec!["x86_64-linux".to_owned()], vec![], true)]
+        vec![BuildWorkerCaps {
+            architectures: vec!["x86_64-linux".to_owned()],
+            features: Vec::new(),
+            network: true,
+            build: true,
+        }]
     }
 
     #[test]
@@ -134,9 +139,15 @@ mod tests {
 
     #[test]
     fn the_fingerprint_ignores_worker_order() {
+        let worker = |arch: &str, features: Vec<String>, network: bool| BuildWorkerCaps {
+            architectures: vec![arch.to_owned()],
+            features,
+            network,
+            build: true,
+        };
         let a = vec![
-            (vec!["a".to_owned()], vec!["kvm".to_owned()], true),
-            (vec!["b".to_owned()], vec![], false),
+            worker("a", vec!["kvm".to_owned()], true),
+            worker("b", vec![], false),
         ];
         let b = vec![a[1].clone(), a[0].clone()];
         assert_eq!(caps_fingerprint(&a), caps_fingerprint(&b));

backend/gradient-scheduler/src/buildability.rs

@@ -19,7 +19,24 @@ use crate::count::clamp_to_u32;
 use gradient_pool::build_needs_network;
 use gradient_wire::types::BuildSpecKind;
 
-pub(crate) type BuildWorkerCaps = (Vec<String>, Vec<String>, bool);
+#[derive(Debug, Clone, PartialEq, Eq)]
+pub(crate) struct BuildWorkerCaps {
+    pub architectures: Vec<String>,
+    pub features: Vec<String>,
+    pub network: bool,
+    pub build: bool,
+}
+
+impl From<&gradient_pool::WorkerInfo> for BuildWorkerCaps {
+    fn from(worker: &gradient_pool::WorkerInfo) -> Self {
+        Self {
+            architectures: worker.architectures.clone(),
+            features: worker.system_features.clone(),
+            network: worker.capabilities.network,
+            build: worker.capabilities.build,
+        }
+    }
+}
 
 pub(crate) struct BuildabilityChecker {
     drv_by_id: HashMap<DerivationId, MDerivation>,
@@ -99,17 +116,15 @@ impl BuildabilityChecker {
             drv.is_fixed_output,
         );
         let needs_network = build_needs_network(kind, drv.is_fixed_output);
-        let mut able = worker_caps
-            .iter()
-            .filter(|(_, _, network)| *network || !needs_network);
+        let mut able = worker_caps.iter().filter(|w| w.network || !needs_network);
         match kind {
             BuildSpecKind::Substitute | BuildSpecKind::Download => able.next().is_some(),
             BuildSpecKind::Build => {
                 let required = self.required_features_for(&build.derivation);
-                able.any(|(arch, feats, _)| {
+                able.filter(|w| w.build).any(|w| {
                     let arch_ok = drv.architecture == gradient_types::BUILTIN_ARCH
-                        || arch.iter().any(|a| a == &drv.architecture);
-                    let feats_ok = required.iter().all(|f| feats.iter().any(|sf| sf == f));
+                        || w.architectures.iter().any(|a| a == &drv.architecture);
+                    let feats_ok = required.iter().all(|f| w.features.iter().any(|sf| sf == f));
                     arch_ok && feats_ok
                 })
             }
@@ -195,7 +210,8 @@ impl BuildabilityChecker {
 
         let mut available_architectures: Vec<String> = worker_caps
             .iter()
-            .flat_map(|(archs, _, _)| archs.iter().cloned())
+            .filter(|w| w.build)
+            .flat_map(|w| w.architectures.iter().cloned())
             .collect();
         available_architectures.sort_unstable();
         available_architectures.dedup();
@@ -295,7 +311,7 @@ mod tests {
         let builds = vec![build_for(d_x86.id, eval_id), build_for(d_arm.id, eval_id)];
         let checker = checker_with(vec![d_x86, d_arm], vec![]);
 
-        let caps: Vec<BuildWorkerCaps> = vec![(vec!["x86_64-linux".into()], vec![], true)];
+        let caps: Vec<BuildWorkerCaps> = vec![builder("x86_64-linux", true)];
         let reason = checker.compute_waiting_reason(&builds, &caps);
         let (unmet, connected_workers, available_architectures) = workers_view(&reason);
 
@@ -315,7 +331,7 @@ mod tests {
         let builds = vec![build_for(drv_id, eval_id)];
         let checker = checker_with(vec![d], vec![(drv_id, feat_id, "kvm")]);
 
-        let caps: Vec<BuildWorkerCaps> = vec![(vec!["x86_64-linux".into()], vec![], true)];
+        let caps: Vec<BuildWorkerCaps> = vec![builder("x86_64-linux", true)];
         let reason = checker.compute_waiting_reason(&builds, &caps);
         let (unmet, _, _) = workers_view(&reason);
 
@@ -353,7 +369,7 @@ mod tests {
         let builds = vec![build_for(d.id, eval_id)];
         let checker = checker_with(vec![d], vec![]);
 
-        let caps: Vec<BuildWorkerCaps> = vec![(vec!["x86_64-linux".into()], vec![], true)];
+        let caps: Vec<BuildWorkerCaps> = vec![builder("x86_64-linux", true)];
         let reason = checker.compute_waiting_reason(&builds, &caps);
         let (unmet, _, _) = workers_view(&reason);
 
@@ -377,7 +393,7 @@ mod tests {
         let build = cache_available_build(d.id, eval_id);
         let checker = checker_with(vec![d], vec![]);
 
-        let caps: Vec<BuildWorkerCaps> = vec![(vec!["x86_64-linux".into()], vec![], true)];
+        let caps: Vec<BuildWorkerCaps> = vec![builder("x86_64-linux", true)];
         let builds = [build];
         assert!(checker.any_buildable(&builds, &caps));
         let reason = checker.compute_waiting_reason(&builds, &caps);
@@ -413,7 +429,7 @@ mod tests {
         build.cache_available = false;
         let checker = checker_with(vec![d], vec![]);
 
-        let caps: Vec<BuildWorkerCaps> = vec![(vec!["x86_64-linux".into()], vec![], true)];
+        let caps: Vec<BuildWorkerCaps> = vec![builder("x86_64-linux", true)];
         let builds = [build];
         assert!(!checker.any_buildable(&builds, &caps));
         let reason = checker.compute_waiting_reason(&builds, &caps);
@@ -422,15 +438,68 @@ mod tests {
         assert_eq!(unmet[0].architecture, "aarch64-linux");
     }
 
+    fn builder(architecture: &str, network: bool) -> BuildWorkerCaps {
+        BuildWorkerCaps {
+            architectures: vec![architecture.to_owned()],
+            features: Vec::new(),
+            network,
+            build: true,
+        }
+    }
+
+    fn network_only() -> BuildWorkerCaps {
+        BuildWorkerCaps {
+            build: false,
+            ..builder("x86_64-linux", true)
+        }
+    }
+
+    #[test]
+    fn a_transfer_runs_on_a_network_worker_without_build() {
+        let eval_id = EvaluationId::now_v7();
+        let passthrough = drv(DerivationId::now_v7(), "aarch64-linux");
+        let mut source = drv(DerivationId::now_v7(), gradient_types::BUILTIN_ARCH);
+        source.is_fixed_output = true;
+        let substitute = [cache_available_build(passthrough.id, eval_id)];
+        let download = [build_for(source.id, eval_id)];
+        let checker = checker_with(vec![passthrough, source], vec![]);
+
+        assert!(checker.any_buildable(&substitute, &[network_only()]));
+        assert!(checker.any_buildable(&download, &[network_only()]));
+    }
+
+    #[test]
+    fn a_build_needs_a_build_worker() {
+        let mut fod = drv(DerivationId::now_v7(), "x86_64-linux");
+        fod.is_fixed_output = true;
+        let plain = drv(DerivationId::now_v7(), "x86_64-linux");
+        let builds = [
+            build_for(fod.id, EvaluationId::now_v7()),
+            build_for(plain.id, EvaluationId::now_v7()),
+        ];
+        let checker = checker_with(vec![fod, plain], vec![]);
+
+        assert!(!checker.any_buildable(&builds, &[network_only()]));
+        let reason = checker.compute_waiting_reason(&builds, &[network_only()]);
+        let (unmet, connected_workers, available_architectures) = workers_view(&reason);
+        assert_eq!(
+            unmet.len(),
+            1,
+            "both builds still need a builder: {unmet:?}"
+        );
+        assert_eq!(connected_workers, 1);
+        assert!(available_architectures.is_empty());
+    }
+
     #[test]
     fn a_fixed_output_build_needs_a_network_worker() {
         let mut d = drv(DerivationId::now_v7(), "x86_64-linux");
         d.is_fixed_output = true;
         let builds = [build_for(d.id, EvaluationId::now_v7())];
         let checker = checker_with(vec![d], vec![]);
 
-        assert!(!checker.any_buildable(&builds, &[(vec!["x86_64-linux".into()], vec![], false)]));
-        assert!(checker.any_buildable(&builds, &[(vec!["x86_64-linux".into()], vec![], true)]));
+        assert!(!checker.any_buildable(&builds, &[builder("x86_64-linux", false)]));
+        assert!(checker.any_buildable(&builds, &[builder("x86_64-linux", true)]));
     }
 
     #[test]
@@ -440,7 +509,7 @@ mod tests {
         let mut b = build_for(d.id, eval_id);
         b.status = BuildStatus::Created;
         let checker = checker_with(vec![d], vec![]);
-        let caps = vec![(vec!["x86_64-linux".to_string()], vec![], true)];
+        let caps: Vec<BuildWorkerCaps> = vec![builder("x86_64-linux", true)];
         assert!(!checker.any_buildable(&[b], &caps));
     }
 
@@ -451,14 +520,8 @@ mod tests {
         let b = build_for(d.id, eval_id);
         let checker = checker_with(vec![d], vec![]);
         let shared_builds = std::slice::from_ref(&b);
-        assert!(checker.any_buildable(
-            shared_builds,
-            &[(vec!["x86_64-linux".to_string()], vec![], true)]
-        ));
-        assert!(!checker.any_buildable(
-            shared_builds,
-            &[(vec!["aarch64-linux".to_string()], vec![], true)]
-        ));
+        assert!(checker.any_buildable(shared_builds, &[builder("x86_64-linux", true)]));
+        assert!(!checker.any_buildable(shared_builds, &[builder("aarch64-linux", true)]));
     }
 
     #[test]
@@ -475,7 +538,7 @@ mod tests {
             pending(linux.id, BuildStatus::Queued),
         );
         let checker = checker_with(vec![darwin, linux], vec![]);
-        let caps = vec![(vec!["x86_64-linux".to_string()], vec![], true)];
+        let caps: Vec<BuildWorkerCaps> = vec![builder("x86_64-linux", true)];
 
         let unbuildable = checker.unbuildable_imports(&[stuck.clone(), building, buildable], &caps);
 

backend/gradient-scheduler/src/cases/workers.rs

@@ -652,7 +652,7 @@ async fn w10_a_worker_without_build_gets_no_substitute_job() {
 }
 
 #[tokio::test]
-async fn w11_a_worker_without_build_gets_no_download_job() {
+async fn w11_a_network_worker_without_build_gets_the_download_job() {
     let seed = shape()
         .one(EvaluationId::now_v7(), A, A_OUT)
         .system("builtin")
@@ -663,14 +663,17 @@ async fn w11_a_worker_without_build_gets_no_download_job() {
 
     case.admit_startable_moves().await;
     let offer = case.offers_full(W1).await;
-    let nothing = case.request_job(W1, JobKind::Build).await;
     assert_eq!(
         offered(&offer),
-        Vec::<String>::new(),
+        vec![a.clone()],
         "after 1: offers result.candidates"
     );
-    assert!(nothing.is_none(), "after 1: request result == nothing");
-    assert_eq!(case.assignments(&a).await, vec![], "after 1: dj == []");
+    case.score_all(W1).await;
+    let assignment = case
+        .request_job(W1, JobKind::Build)
+        .await
+        .expect("after 1: W1 gets download:A");
+    assert_eq!(assignment.job_id(), a, "after 1: result.job_id");
     case.check("step 1").await;
 
     case.finish().await;