test: isolate pricing and hosting E2E assumptions

.github/workflows/rust.yml

@@ -78,6 +78,8 @@ jobs:
       - name: Test
         env:
           C0MPUTE_REQUIRE_FUSE: "1"
+          # Keep loopback E2Es isolated when the runner is near real workers.
+          C0MPUTE_DISABLE_MDNS: "1"
         run: cargo test --workspace --locked
 
       # Blocking, unlike the lint job: this invariant has no legacy debt to

node/crates/c0mpute-core/tests/container_run_e2e.rs

@@ -72,6 +72,8 @@ fn docker(args: &[&str]) -> String {
 
 #[tokio::test(flavor = "multi_thread", worker_threads = 4)]
 async fn builds_runs_and_returns_results() {
+    // Run only against this test's loopback worker, not dev2 workers found by mDNS.
+    unsafe { std::env::set_var("C0MPUTE_DISABLE_MDNS", "1") };
     let rt = tokio::task::spawn_blocking(|| container_run::runtime().clone())
         .await
         .unwrap();
@@ -155,6 +157,7 @@ async fn builds_runs_and_returns_results() {
             context: Some(ctx_tar.clone()),
             build: Some(build.clone()),
             builders: vec![worker_id.clone()],
+            max_price_usd: 1.0,
             timeout_secs: Some(600),
             ..Default::default()
         },
@@ -206,6 +209,7 @@ async fn builds_runs_and_returns_results() {
             input: Some(in_tar.clone()),
             outputs: vec!["out".into(), "nope".into()],
             builders: vec![worker_id.clone()],
+            max_price_usd: 1.0,
             timeout_secs: Some(600),
             memory_mb: Some(256),
             cpus: Some(1.0),
@@ -256,6 +260,7 @@ async fn builds_runs_and_returns_results() {
             build: Some(build.clone()),
             cmd: vec!["sh".into(), "-c".into(), "echo boom; exit 3".into()],
             builders: vec![worker_id.clone()],
+            max_price_usd: 1.0,
             timeout_secs: Some(600),
             ..Default::default()
         },
@@ -293,6 +298,7 @@ async fn builds_runs_and_returns_results() {
             context: Some(ctx_tar.clone()),
             build: Some(build),
             builders: vec![worker_id.clone()],
+            max_price_usd: 1.0,
             timeout_secs: Some(60),
             ..Default::default()
         },

node/crates/c0mpute-core/tests/hosting_e2e.rs

@@ -79,11 +79,19 @@ async fn sites_replicate_to_three_heal_and_serve_from_any_gateway() {
         .with_env_filter("c0mpute_core=info,c0mpute_net=warn")
         .with_test_writer()
         .try_init();
+    // The integration test runs beside live dev2 workers, so keep its local
+    // mesh isolated from LAN peers and real hosting traffic.
+    unsafe { std::env::set_var("C0MPUTE_DISABLE_MDNS", "1") };
     let tmp = tempdir().unwrap();
 
     // A hub everyone dials (it also hosts), then four more hosting nodes.
     let (hub_net, hub_store) = node(&tmp.path().join("h0"), vec![]).await;
     let hub_addr = listen_addr(&hub_net).await;
+    // The test disables mDNS, so explicitly connect each new node to all
+    // existing loopback peers. This keeps DHT provider lookups deterministic
+    // on CI, where multicast discovery is unavailable.
+    let mut mesh_addrs = vec![hub_addr.clone()];
+    let mut mesh_nets = vec![hub_net.clone()];
     let mut hosts: Vec<Hosting> = Vec::new();
     let opts = |dir: &std::path::Path| Options {
         hosts: true,
@@ -96,13 +104,17 @@ async fn sites_replicate_to_three_heal_and_serve_from_any_gateway() {
     hosts.push(Hosting::start(hub_net.clone(), hub_store, opts(&tmp.path().join("h0"))).await.unwrap());
     for i in 1..5 {
         let d = tmp.path().join(format!("h{i}"));
-        let (n, s) = node(&d, vec![hub_addr.clone()]).await;
+        let (n, s) = node(&d, mesh_addrs.clone()).await;
+        mesh_addrs.push(listen_addr(&n).await);
+        mesh_nets.push(n.clone());
         hosts.push(Hosting::start(n, s, opts(&d)).await.unwrap());
     }
 
     // A gateway that hosts nothing.
     let gd = tmp.path().join("gw");
-    let (gn, gs) = node(&gd, vec![hub_addr.clone()]).await;
+    let (gn, gs) = node(&gd, mesh_addrs.clone()).await;
+    mesh_addrs.push(listen_addr(&gn).await);
+    mesh_nets.push(gn.clone());
     let gateway = Hosting::start(
         gn,
         gs,
@@ -117,7 +129,8 @@ async fn sites_replicate_to_three_heal_and_serve_from_any_gateway() {
 
     // The publisher: seeds its own site, does not count as a copy.
     let pd = tmp.path().join("pub");
-    let (pn, ps) = node(&pd, vec![hub_addr.clone()]).await;
+    let (pn, ps) = node(&pd, mesh_addrs).await;
+    mesh_nets.push(pn.clone());
     let publisher = Hosting::start(
         pn,
         ps,
@@ -130,14 +143,32 @@ async fn sites_replicate_to_three_heal_and_serve_from_any_gateway() {
     .await
     .unwrap();
 
-    // gossipsub mesh warm-up
+    // Wait for the explicit mesh to be ready before advertising the site.
+    let mesh = mesh_nets.clone();
+    wait_for("full loopback mesh", 30, || {
+        let mesh = mesh.clone();
+        async move {
+            let expected = mesh.len() - 1;
+            let mut connected = true;
+            for n in &mesh {
+                connected &= n.connected_peers().await.len() >= expected;
+            }
+            connected
+        }
+    })
+    .await;
+    // Let gossipsub subscriptions settle across the connected peers.
     tokio::time::sleep(Duration::from_secs(3)).await;
 
     let site_dir = tmp.path().join("site");
     std::fs::create_dir_all(site_dir.join("docs")).unwrap();
     std::fs::write(site_dir.join("index.html"), b"<meta property=\"og:image\" content=\"img/hero.png\"><h1>hello from the network</h1>").unwrap();
     std::fs::write(site_dir.join("docs/index.html"), b"docs page").unwrap();
-    let big: Vec<u8> = (0..(5 * 1024 * 1024)).map(|i| (i % 251) as u8).collect();
+    // Three chunks keep this multi-chunk while limiting loopback traffic on
+    // slower CI runners.
+    let big: Vec<u8> = (0..(2 * 1024 * 1024 + 97))
+        .map(|i| (i % 251) as u8)
+        .collect();
     std::fs::write(site_dir.join("big.bin"), &big).unwrap();
     let built = build_from_dir(&site_dir, "demo").unwrap();
     let owner = libp2p::identity::Keypair::generate_ed25519();

node/crates/c0mpute-core/tests/hosting_market_e2e.rs

@@ -86,6 +86,8 @@ async fn preferred_hosts_hold_pass_challenges_and_get_paid() {
         .with_env_filter("c0mpute_core=info,c0mpute_net=warn")
         .with_test_writer()
         .try_init();
+    // Keep this local mesh isolated from LAN peers and real hosting traffic.
+    unsafe { std::env::set_var("C0MPUTE_DISABLE_MDNS", "1") };
     let tmp = tempdir().unwrap();
 
     // Five hosting nodes, each asking $0.004/GiB-month with a payout wallet.
@@ -104,18 +106,26 @@ async fn preferred_hosts_hold_pass_challenges_and_get_paid() {
     };
     let (hub_net, hub_store) = node(&tmp.path().join("h0"), vec![]).await;
     let hub_addr = listen_addr(&hub_net).await;
+    // CI does not provide reliable multicast discovery. Connect each node to
+    // all existing loopback peers so provider lookups see the same mesh.
+    let mut mesh_addrs = vec![hub_addr.clone()];
+    let mut mesh_nets = vec![hub_net.clone()];
     let mut hosts: Vec<Hosting> = vec![Hosting::start(hub_net, hub_store, host_opts(&tmp.path().join("h0"), 0)).await.unwrap()];
     for i in 1..5 {
         let d = tmp.path().join(format!("h{i}"));
-        let (n, s) = node(&d, vec![hub_addr.clone()]).await;
+        let (n, s) = node(&d, mesh_addrs.clone()).await;
+        mesh_addrs.push(listen_addr(&n).await);
+        mesh_nets.push(n.clone());
         hosts.push(Hosting::start(n, s, host_opts(&d, i)).await.unwrap());
     }
 
     // The publisher: owns and pays for the site, challenges its holders.
     let owner = libp2p::identity::Keypair::generate_ed25519();
     let owner_key = owner_hex(&owner).unwrap();
     let pd = tmp.path().join("pub");
-    let (pn, ps) = node(&pd, vec![hub_addr.clone()]).await;
+    let (pn, ps) = node(&pd, mesh_addrs.clone()).await;
+    mesh_addrs.push(listen_addr(&pn).await);
+    mesh_nets.push(pn.clone());
     let publisher = Hosting::start(
         pn,
         ps,
@@ -136,7 +146,8 @@ async fn preferred_hosts_hold_pass_challenges_and_get_paid() {
     // A gateway that challenges everything with its own key: its receipts
     // travel on gossip but do not count for the owner's settlement.
     let gd = tmp.path().join("gw");
-    let (gn, gs) = node(&gd, vec![hub_addr.clone()]).await;
+    let (gn, gs) = node(&gd, mesh_addrs.clone()).await;
+    mesh_nets.push(gn.clone());
     let gateway = Hosting::start(
         gn,
         gs,
@@ -152,6 +163,21 @@ async fn preferred_hosts_hold_pass_challenges_and_get_paid() {
     .await
     .unwrap();
 
+    // Wait for explicit connections before publishing or challenging.
+    let mesh = mesh_nets.clone();
+    wait_for("full loopback mesh", 30, || {
+        let mesh = mesh.clone();
+        async move {
+            let expected = mesh.len() - 1;
+            let mut connected = true;
+            for n in &mesh {
+                connected &= n.connected_peers().await.len() >= expected;
+            }
+            connected
+        }
+    })
+    .await;
+    // Let gossipsub subscriptions settle across the connected peers.
     tokio::time::sleep(Duration::from_secs(3)).await;
 
     // Two of the five are preferred, at $1/GiB-month.
@@ -191,9 +217,9 @@ async fn preferred_hosts_hold_pass_challenges_and_get_paid() {
         }
     })
     .await;
-    // Converged: every host sees the same three copies (both preferred) and
-    // keeps seeing them for longer than the trim grace (one holder TTL), so a
-    // copy that was briefly extra has been released and none comes back.
+    // Converged: the publisher and every host see the same three copies (both
+    // preferred) for longer than the trim grace (one holder TTL), so a copy
+    // that was briefly extra has been released and none comes back.
     // Bounded, rather than a fixed sleep that races the trim.
     let stable_for = fast().holder_ttl + Duration::from_secs(2);
     let deadline = tokio::time::Instant::now() + Duration::from_secs(60);
@@ -202,6 +228,7 @@ async fn preferred_hosts_hold_pass_challenges_and_get_paid() {
         publisher.regossip(&rec).await;
         let view = hosts[0].holders(&site).await;
         let mut agree = view.len() == 3 && preferred.iter().all(|x| view.contains(x));
+        agree &= publisher.holders(&site).await == view;
         for h in &hosts[1..] {
             agree &= h.holders(&site).await == view;
         }

node/crates/c0mpute-core/tests/inference_e2e.rs

@@ -79,6 +79,8 @@ async fn fake_runtime() -> String {
 #[tokio::test(flavor = "multi_thread", worker_threads = 4)]
 async fn chat_turns_are_answered_peer_to_peer() {
     let _ = tracing_subscriber::fmt().with_env_filter("c0mpute_core=info,c0mpute_net=warn").with_test_writer().try_init();
+    // Keep the loopback buyer and worker off the live LAN network.
+    unsafe { std::env::set_var("C0MPUTE_DISABLE_MDNS", "1") };
     let live = std::env::var("C0MPUTE_INFERENCE_LIVE_URL").ok().filter(|s| !s.is_empty());
     let base = match &live {
         Some(u) => u.trim_end_matches('/').to_string(),
@@ -108,7 +110,8 @@ async fn chat_turns_are_answered_peer_to_peer() {
 
     // ── buyer ──────────────────────────────────────────────────────────
     let (buyer, buyer_store) = node(&tmp.path().join("b"), vec![worker_addr]).await;
-    let bridge = JobsBridge::new(buyer.clone(), buyer_store, "test-token-0123456789abcdef".into(), 0.0, vec![]).unwrap().with_bid_window(Duration::from_secs(2));
+    // This paid inference scenario needs a cap above the default node floor.
+    let bridge = JobsBridge::new(buyer.clone(), buyer_store, "test-token-0123456789abcdef".into(), 1.0, vec![]).unwrap().with_bid_window(Duration::from_secs(2));
     tokio::time::sleep(Duration::from_secs(2)).await;
 
     // A multi-turn conversation, any model.

node/crates/c0mpute-core/tests/ocr_e2e.rs

@@ -117,6 +117,8 @@ async fn pdf_ocr_round_trips_through_the_bridge() {
         .with_env_filter("c0mpute_core=info,c0mpute_net=warn")
         .with_test_writer()
         .try_init();
+    // Keep this loopback OCR job off the live LAN network.
+    unsafe { std::env::set_var("C0MPUTE_DISABLE_MDNS", "1") };
     let tmp = tempdir().unwrap();
 
     // ── worker: the real OCR runner, with its bid filter ───────────────
@@ -151,7 +153,7 @@ async fn pdf_ocr_round_trips_through_the_bridge() {
         buyer.clone(),
         buyer_store,
         "test-token-0123456789abcdef".into(),
-        0.0,
+        1.0,
         vec![],
     )
     .unwrap()

node/crates/c0mpute-core/tests/render_e2e.rs

@@ -114,6 +114,8 @@ async fn frames_render_across_workers() {
         .with_env_filter("c0mpute_core=info,c0mpute_net=warn")
         .with_test_writer()
         .try_init();
+    // Keep these loopback render jobs off the live LAN network.
+    unsafe { std::env::set_var("C0MPUTE_DISABLE_MDNS", "1") };
     let tmp = tempdir().unwrap();
     let blend = tmp.path().join("scene.blend");
     make_scene(&blend);
@@ -139,7 +141,7 @@ async fn frames_render_across_workers() {
         resolution_pct: None,
         samples: Some(4),
         frame_timeout: Duration::from_secs(300),
-        max_price_usd: 0.0,
+        max_price_usd: 1.0,
         concurrency: 3,
         attempts: 3,
         out_dir: out.clone(),
@@ -179,7 +181,7 @@ async fn frames_render_across_workers() {
         buyer.clone(),
         buyer_store,
         "test-token-0123456789abcdef".into(),
-        0.0,
+        1.0,
         vec![],
     )
     .unwrap()

node/crates/c0mpute-core/tests/workflow_e2e.rs

@@ -63,6 +63,8 @@ async fn transform_step_round_trips_through_the_bridge() {
         .with_env_filter("c0mpute_core=info,c0mpute_net=warn")
         .with_test_writer()
         .try_init();
+    // Keep loopback jobs off the live LAN network.
+    unsafe { std::env::set_var("C0MPUTE_DISABLE_MDNS", "1") };
     let tmp = tempdir().unwrap();
 
     // ── worker: the real workflow.step runner ──────────────────────────
@@ -84,7 +86,7 @@ async fn transform_step_round_trips_through_the_bridge() {
 
     // ── buyer: the bridge, dialing the worker directly ─────────────────
     let (buyer, buyer_store) = node(&tmp.path().join("b"), vec![worker_addr]).await;
-    let bridge = JobsBridge::new(buyer.clone(), buyer_store, TOKEN.into(), 0.0, vec![])
+    let bridge = JobsBridge::new(buyer.clone(), buyer_store, TOKEN.into(), 1.0, vec![])
         .unwrap()
         .with_bid_window(Duration::from_secs(2));
     let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap();