From cb03a86cee035c8ca8cd9d856c64d62afaf16ceb Mon Sep 17 00:00:00 2001 From: Anthony Ettinger Date: Thu, 12 Nov 2026 11:24:27 +0000 Subject: [PATCH 1/8] test(container): set an explicit job price cap --- node/crates/c0mpute-core/tests/container_run_e2e.rs | 4 ++++ 1 file changed, 4 insertions(+) diff --git a/node/crates/c0mpute-core/tests/container_run_e2e.rs b/node/crates/c0mpute-core/tests/container_run_e2e.rs index f1fb7468..dcf56a28 100644 --- a/node/crates/c0mpute-core/tests/container_run_e2e.rs +++ b/node/crates/c0mpute-core/tests/container_run_e2e.rs @@ -155,6 +155,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 +207,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 +258,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 +296,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() }, From 33c58e0aa7714c48596226a74d5fbd68578b5f8d Mon Sep 17 00:00:00 2001 From: Anthony Ettinger Date: Thu, 12 Nov 2026 11:56:20 +0000 Subject: [PATCH 2/8] test(hosting): isolate loopback mesh from LAN peers --- node/crates/c0mpute-core/tests/hosting_e2e.rs | 3 +++ 1 file changed, 3 insertions(+) diff --git a/node/crates/c0mpute-core/tests/hosting_e2e.rs b/node/crates/c0mpute-core/tests/hosting_e2e.rs index d6e22c2b..b293d210 100644 --- a/node/crates/c0mpute-core/tests/hosting_e2e.rs +++ b/node/crates/c0mpute-core/tests/hosting_e2e.rs @@ -79,6 +79,9 @@ 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. From de82427d08f4102a5127dd5115199ae599dd0abd Mon Sep 17 00:00:00 2001 From: Anthony Ettinger Date: Thu, 12 Nov 2026 12:01:50 +0000 Subject: [PATCH 3/8] test(container): isolate loopback worker from LAN peers --- node/crates/c0mpute-core/tests/container_run_e2e.rs | 2 ++ 1 file changed, 2 insertions(+) diff --git a/node/crates/c0mpute-core/tests/container_run_e2e.rs b/node/crates/c0mpute-core/tests/container_run_e2e.rs index dcf56a28..389fc2a8 100644 --- a/node/crates/c0mpute-core/tests/container_run_e2e.rs +++ b/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(); From f5a0039d99b2993ee51f95d47221dee7c3874875 Mon Sep 17 00:00:00 2001 From: Anthony Ettinger Date: Thu, 12 Nov 2026 12:12:53 +0000 Subject: [PATCH 4/8] test(inference): set budget above node floor --- node/crates/c0mpute-core/tests/inference_e2e.rs | 5 ++++- 1 file changed, 4 insertions(+), 1 deletion(-) diff --git a/node/crates/c0mpute-core/tests/inference_e2e.rs b/node/crates/c0mpute-core/tests/inference_e2e.rs index 54e91ff7..fdc29a88 100644 --- a/node/crates/c0mpute-core/tests/inference_e2e.rs +++ b/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. From 1692c63aebc41de704566ce0c4472c19ac300626 Mon Sep 17 00:00:00 2001 From: Anthony Ettinger Date: Thu, 12 Nov 2026 12:17:53 +0000 Subject: [PATCH 5/8] test: isolate loopback jobs from node pricing and LAN --- .github/workflows/rust.yml | 2 ++ node/crates/c0mpute-core/tests/ocr_e2e.rs | 4 +++- node/crates/c0mpute-core/tests/render_e2e.rs | 6 ++++-- node/crates/c0mpute-core/tests/workflow_e2e.rs | 4 +++- 4 files changed, 12 insertions(+), 4 deletions(-) diff --git a/.github/workflows/rust.yml b/.github/workflows/rust.yml index bd54bb27..0616f1dd 100644 --- a/.github/workflows/rust.yml +++ b/.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 diff --git a/node/crates/c0mpute-core/tests/ocr_e2e.rs b/node/crates/c0mpute-core/tests/ocr_e2e.rs index 5be968cc..de95b310 100644 --- a/node/crates/c0mpute-core/tests/ocr_e2e.rs +++ b/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() diff --git a/node/crates/c0mpute-core/tests/render_e2e.rs b/node/crates/c0mpute-core/tests/render_e2e.rs index 40911898..a1e2da7f 100644 --- a/node/crates/c0mpute-core/tests/render_e2e.rs +++ b/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() diff --git a/node/crates/c0mpute-core/tests/workflow_e2e.rs b/node/crates/c0mpute-core/tests/workflow_e2e.rs index 8d8b12b5..2531d4cf 100644 --- a/node/crates/c0mpute-core/tests/workflow_e2e.rs +++ b/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(); From 6df7d4177c2a855d0f9b7749b9e5c106482d3f81 Mon Sep 17 00:00:00 2001 From: Anthony Ettinger Date: Thu, 12 Nov 2026 12:35:20 +0000 Subject: [PATCH 6/8] test(hosting): mesh loopback peers before replication --- node/crates/c0mpute-core/tests/hosting_e2e.rs | 32 ++++++++++++++++--- 1 file changed, 28 insertions(+), 4 deletions(-) diff --git a/node/crates/c0mpute-core/tests/hosting_e2e.rs b/node/crates/c0mpute-core/tests/hosting_e2e.rs index b293d210..dc590fc3 100644 --- a/node/crates/c0mpute-core/tests/hosting_e2e.rs +++ b/node/crates/c0mpute-core/tests/hosting_e2e.rs @@ -87,6 +87,11 @@ async fn sites_replicate_to_three_heal_and_serve_from_any_gateway() { // 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 = Vec::new(); let opts = |dir: &std::path::Path| Options { hosts: true, @@ -99,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, @@ -120,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, @@ -133,7 +143,21 @@ 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"); From 063dc017d6d9d73f078f48e10f80785d22886ac7 Mon Sep 17 00:00:00 2001 From: Anthony Ettinger Date: Thu, 12 Nov 2026 13:14:45 +0000 Subject: [PATCH 7/8] test(hosting): shrink loopback fixture for CI --- node/crates/c0mpute-core/tests/hosting_e2e.rs | 6 +++++- 1 file changed, 5 insertions(+), 1 deletion(-) diff --git a/node/crates/c0mpute-core/tests/hosting_e2e.rs b/node/crates/c0mpute-core/tests/hosting_e2e.rs index dc590fc3..069abd0c 100644 --- a/node/crates/c0mpute-core/tests/hosting_e2e.rs +++ b/node/crates/c0mpute-core/tests/hosting_e2e.rs @@ -164,7 +164,11 @@ async fn sites_replicate_to_three_heal_and_serve_from_any_gateway() { std::fs::create_dir_all(site_dir.join("docs")).unwrap(); std::fs::write(site_dir.join("index.html"), b"

hello from the network

").unwrap(); std::fs::write(site_dir.join("docs/index.html"), b"docs page").unwrap(); - let big: Vec = (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 = (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(); From 9f583c9c69f9a3dcb094ac9e011dfe085407954f Mon Sep 17 00:00:00 2001 From: Anthony Ettinger Date: Thu, 12 Nov 2026 13:29:29 +0000 Subject: [PATCH 8/8] test(hosting): await marketplace mesh convergence --- .../c0mpute-core/tests/hosting_market_e2e.rs | 39 ++++++++++++++++--- 1 file changed, 33 insertions(+), 6 deletions(-) diff --git a/node/crates/c0mpute-core/tests/hosting_market_e2e.rs b/node/crates/c0mpute-core/tests/hosting_market_e2e.rs index 4fac99a2..4a23b24b 100644 --- a/node/crates/c0mpute-core/tests/hosting_market_e2e.rs +++ b/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,10 +106,16 @@ 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 = 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()); } @@ -115,7 +123,9 @@ async fn preferred_hosts_hold_pass_challenges_and_get_paid() { 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; }