zc2 0.0.29

P2P compute broker with credit-based billing, WAL, and broker mesh support
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112

//! Self-exec entry points for the agent integration tests. The test binary
//! re-runs itself as `<exe> agent::testbin::<entry> --exact --ignored` to become
//! a fake worker or a real broker in a separate process. Skipped in normal runs.

use std::sync::atomic::{AtomicU64, Ordering};
use std::sync::Arc;
use std::time::Duration;

pub fn self_exec_argv(entry: &str) -> Vec<String> {
    vec![
        std::env::current_exe()
            .unwrap()
            .to_string_lossy()
            .to_string(),
        format!("agent::testbin::{entry}"),
        "--exact".into(),
        "--ignored".into(),
        "--nocapture".into(),
        "--test-threads=1".into(),
    ]
}

/// Exit as soon as the parent (the agent under test) is gone. Only reached
/// once `child_port()` has confirmed this really is a spawned child — the
/// main test harness process must never fall into this, since a container's
/// PID 1 (where `getppid()` can read as 0) would otherwise exit the whole
/// run immediately. The ppid is sampled immediately, at entry, before
/// anything else runs: if the parent is already gone by then (reparented to
/// init, ppid 1) this exits right away instead of waiting for the first poll
/// to notice.
fn exit_with_parent() {
    let ppid = unsafe { libc::getppid() };
    if ppid <= 1 {
        std::process::exit(0);
    }
    std::thread::spawn(move || loop {
        std::thread::sleep(Duration::from_millis(300));
        let now = unsafe { libc::getppid() };
        if now != ppid || now <= 1 {
            std::process::exit(0);
        }
    });
}

fn child_port() -> Option<u16> {
    std::env::var("ZAKURO_AGENT_CHILD_PORT").ok()?.parse().ok()
}

#[test]
#[ignore = "self-exec entry point for the agent integration tests"]
fn fake_worker_main() {
    let Some(port) = child_port() else { return };
    exit_with_parent();
    let name = std::env::var("ZAKURO_AGENT_CHILD_NAME").unwrap_or_else(|_| "fake".into());
    let server = tiny_http::Server::http(("127.0.0.1", port)).unwrap();
    // Counts every request this worker has ever received on /execute and on
    // /info, so tests can observe real dispatch (a positive control) and a
    // real, counted number of discovery probes, instead of assuming timing.
    let execute_count = Arc::new(AtomicU64::new(0));
    let info_count = Arc::new(AtomicU64::new(0));
    for req in server.incoming_requests() {
        // The `_count` endpoints are checked before their shorter prefixes
        // (`/info_count` also starts with `/info`, `/execute_count` also
        // starts with `/execute`), so a count query is never miscounted as
        // the request it is reporting on.
        let body = if req.url().starts_with("/execute_count") {
            format!(r#"{{"count":{}}}"#, execute_count.load(Ordering::SeqCst))
        } else if req.url().starts_with("/info_count") {
            format!(r#"{{"count":{}}}"#, info_count.load(Ordering::SeqCst))
        } else if req.url().starts_with("/info") {
            info_count.fetch_add(1, Ordering::SeqCst);
            format!(
                r#"{{"name":"{name}","worker_type":"zakuro","resources":{{"cpus_total":1.0,"cpus_available":1.0,"memory_total":1073741824,"memory_available":1073741824,"gpus_total":0,"gpus_available":0}},"pricing":{{"price_per_hour":3.6,"min_charge":0.001}}}}"#
            )
        } else if req.url().starts_with("/execute") {
            execute_count.fetch_add(1, Ordering::SeqCst);
            r#"{"result":"ok"}"#.to_string()
        } else {
            r#"{"status":"healthy"}"#.to_string()
        };
        let _ = req.respond(tiny_http::Response::from_string(body).with_header(
            tiny_http::Header::from_bytes("Content-Type", "application/json").unwrap(),
        ));
    }
}

#[test]
#[ignore = "self-exec entry point for the agent integration tests"]
fn broker_main() {
    let Some(port) = child_port() else { return };
    exit_with_parent();
    let config = crate::broker::BrokerConfig {
        host: "127.0.0.1".into(),
        port,
        verbose: false,
        daemon: true,
        enable_discovery: true,
        health_check_interval: 1,
        worker_timeout: 5,
        enable_p2p: false,
        peer_key: None,
        owner_user_id: None,
        node_name: Some("agent-it".into()),
        worker_key: None,
        api_url: None,
        api_key: None,
        // Reads ZAKURO_SCAN_RANGE / ZAKURO_SCAN_INTERVAL, which the agent sets.
        discovery: crate::broker::DiscoveryConfig::default(),
        ..crate::broker::BrokerConfig::default()
    };
    let _ = crate::broker::start_server(config);
}