//! The gateway metering real calls on the upstream server's own process //! tree. Each test spawns an MCP-ish server of its own (a process group of //! its own, killed at the end) that answers one JSON-RPC `tools/call` per //! line, burning the loop count its arguments name before it replies, and //! relays `tools/call`s to it through [`Gateway::relay`]. //! //! The oracle shares no code with the meter or the gateway: the server's //! `proc_pid_rusage(pid, RUSAGE_INFO_V6)` read into a raw buffer and taken //! at the SDK's byte offsets (energy at 336, reaped-descendant CPU time at //! 96 and 104, billed CPU time at 216, billed and serviced energy at 264 //! and 272), read before and after each relay once the server is quiet //! on the oracle's own terms: no thread on a core or queued //! (`proc_pidinfo(PROC_PIDTASKINFO)` into a raw buffer, `pti_numrunning` at //! byte 88) and none in a wait no signal ends (each thread's `pth_run_state`, //! read raw as well) at two reads 20 ms apart whose energy agrees. A server blocked //! waiting for its next request does not move its count, so between the //! meter's reads and the oracle's nothing is spent: the call's figure must //! be the oracle's advance across the relay, in nanojoules rounded UP to //! the receipt's microjoules (an enforcement point never debits less than //! was read). That is the tolerance — exact to the µJ: the figure is at //! least the oracle's advance and less than 1 µJ over it. //! //! What a call is sealed as is read back from its receipt, verified, and //! compared with the plaintext beside the response. The tests log what they //! saw (`upstream tree:`), and run one at a time (a lock) so one test's //! server is not loading the host under another's. #![cfg(all(target_os = "macos", target_arch = "aarch64"))] use ed25519_dalek::SigningKey; use jcp_gateway::{CallMeter, Gateway, Relayed, Upstream, JCP_META_NS}; use serde_json::{json, Value}; use std::io::{BufRead, BufReader, Write}; use std::os::unix::process::CommandExt; use std::process::{Child, ChildStdin, ChildStdout, Command, Stdio}; use std::sync::{Mutex, MutexGuard}; use std::time::{Duration, Instant}; static ONE_AT_A_TIME: Mutex<()> = Mutex::new(()); fn alone() -> MutexGuard<'static, ()> { ONE_AT_A_TIME.lock().unwrap_or_else(|e| e.into_inner()) } /// How long the gateway's tree meter may wait for the server to go quiet in /// these tests: far past the default bound, because the host runs at a load /// of 14 to 260 and a reply's tail can wait that long for a core. A server /// still running after it would be a gap and the call `Partial`, failing a /// test that expects it whole, which is the point: a whole reading is never /// a stale one. const TEST_SETTLE: Duration = Duration::from_secs(2); /// The JSON-RPC result for the request line in `l`: its id, and its /// `arguments.n` as a loop count burned before replying. const REPLY: &str = r#"id = "null"; n = 0 if (match(l, /"id":[0-9]+/)) id = substr(l, RSTART + 5, RLENGTH - 5) if (match(l, /"n":[0-9]+/)) n = substr(l, RSTART + 4, RLENGTH - 4) + 0 for (i = 0; i < n; i++) x += i printf "{\"jsonrpc\":\"2.0\",\"id\":%s,\"result\":{\"content\":[{\"type\":\"text\",\"text\":\"burned %d\"}]}}\n", id, n"#; const STEWARD: &str = "did:jcp:gateway:burner"; fn steward_sk() -> SigningKey { SigningKey::from_bytes(&[0x2c; 32]) } /// `pid`'s `rusage_info_v6`, raw: `proc_pid_rusage(pid, 6)` into a buffer /// larger than the struct's 464 bytes, as 64-bit words. fn oracle_rusage(pid: i32) -> Option<[u64; 64]> { let mut buf = [0u64; 64]; // SAFETY: `buf` is writable and larger than `rusage_info_v6`. let rc = unsafe { libc::proc_pid_rusage(pid, 6, buf.as_mut_ptr() as *mut libc::rusage_info_t) }; (rc == 0).then_some(buf) } fn energy_nj(ru: &[u64; 64]) -> u64 { ru[336 / 8] } fn reaped_cpu_ticks(ru: &[u64; 64]) -> u64 { ru[96 / 8] + ru[104 / 8] } /// `ri_billed_system_time`, mach ticks: the CPU time of work other /// processes did on the server's behalf, billed to it. fn billed_ticks(ru: &[u64; 64]) -> u64 { ru[216 / 8] } /// `ri_billed_energy`, nJ: that work's energy. fn billed_nj(ru: &[u64; 64]) -> u64 { ru[264 / 8] } /// `ri_serviced_energy`, nJ: the energy of work the server did for others. fn serviced_nj(ru: &[u64; 64]) -> u64 { ru[272 / 8] } /// `pti_numrunning` of `pid`: `proc_pidinfo(pid, PROC_PIDTASKINFO = 4)` into /// a raw 96-byte buffer, the int32 at byte 88. fn oracle_running(pid: i32) -> Option { let mut buf = [0u8; 96]; // SAFETY: `buf` is writable for the 96 bytes passed. let n = unsafe { libc::proc_pidinfo(pid, 4, 0, buf.as_mut_ptr().cast(), 96) }; (n == 96).then(|| i32::from_ne_bytes(buf[88..92].try_into().unwrap())) } /// Whether every thread of `pid` is at rest by its own run state: none on a /// core or queued (`TH_STATE_RUNNING`, 1) and none in a wait that no signal /// ends (`TH_STATE_UNINTERRUPTIBLE`, 4: waiting on the disk, the pager or a /// kernel lock, to run again without anything arriving), the rule the meter /// settles by. `proc_pidinfo(pid, PROC_PIDLISTTHREADS = 6)` into a raw buffer /// of 64-bit thread handles, then `proc_pidinfo(pid, PROC_PIDTHREADINFO = 5, /// handle)` into a raw 112-byte buffer, `pth_run_state` the int32 at byte 24. /// `None` if the listing fails (an exited process has no task). fn oracle_at_rest(pid: i32) -> Option { let mut handles = [0u64; 512]; // SAFETY: `handles` is writable for the bytes passed. let n = unsafe { libc::proc_pidinfo(pid, 6, 0, handles.as_mut_ptr().cast(), (handles.len() * 8) as libc::c_int) }; if n <= 0 { return None; } let n = n as usize / 8; assert!(n < handles.len(), "{pid} has more threads than the oracle's buffer holds"); Some(handles[..n].iter().all(|&handle| { let mut buf = [0u8; 112]; // SAFETY: `buf` is writable for the 112 bytes passed. let got = unsafe { libc::proc_pidinfo(pid, 5, handle, buf.as_mut_ptr().cast(), 112) }; // A thread gone since the listing waits on nothing. got != 112 || !matches!(i32::from_ne_bytes(buf[24..28].try_into().unwrap()), 1 | 4) })) } /// `pid`'s rusage once it is quiet by the oracle's rule: no thread running, /// queued or in a wait no signal ends at two reads 20 ms apart, with the /// same energy at both. Panics /// after 20 s. fn quiet(pid: i32) -> [u64; 64] { let deadline = Instant::now() + Duration::from_secs(20); loop { let first = oracle_rusage(pid).expect("oracle read"); let before = oracle_running(pid) == Some(0) && oracle_at_rest(pid) == Some(true); std::thread::sleep(Duration::from_millis(20)); let second = oracle_rusage(pid).expect("oracle read"); let after = oracle_running(pid) == Some(0) && oracle_at_rest(pid) == Some(true); if before && after && energy_nj(&first) == energy_nj(&second) { return second; } assert!(Instant::now() < deadline, "the oracle never saw {pid} quiet"); } } /// An MCP-ish server: a child in a process group of its own, one JSON-RPC /// request per line on its stdin and one response per line on its stdout, /// killed with its whole group on drop. It names its pid to the gateway /// (`named`, normally its own), and is never waited on while it serves, so /// its pid stays its own. struct Server { child: Child, stdin: ChildStdin, stdout: BufReader, named: Option, } impl Server { fn spawn(program: &str, args: &[&str]) -> Server { let mut child = Command::new(program) .args(args) .stdin(Stdio::piped()) .stdout(Stdio::piped()) .stderr(Stdio::null()) .process_group(0) .spawn() .expect("spawn the server"); let stdin = child.stdin.take().unwrap(); let stdout = BufReader::new(child.stdout.take().unwrap()); let named = Some(child.id()); Server { child, stdin, stdout, named } } /// awk answering every request itself: a single-process server. fn single_process() -> Server { Server::spawn("/usr/bin/awk", &[&format!("{{ l = $0; {REPLY}; fflush() }}")]) } /// A shell answering each request with an awk worker it forks, waits on /// and reaps before it replies: every call reaps a descendant. fn reaping_a_worker_per_call() -> Server { Server::spawn( "/bin/sh", &[ "-c", r#"while IFS= read -r l; do r=$(L="$l" /usr/bin/awk "$1"); printf '%s\n' "$r"; done"#, "sh", &format!(r#"BEGIN {{ l = ENVIRON["L"]; {REPLY} }}"#), ], ) } fn pid(&self) -> i32 { self.child.id() as i32 } } impl Upstream for Server { fn call(&mut self, line: &str) -> std::io::Result { writeln!(self.stdin, "{line}")?; self.stdin.flush()?; let mut reply = String::new(); self.stdout.read_line(&mut reply)?; Ok(reply.trim_end().to_string()) } fn pid(&self) -> Option { self.named } } impl Drop for Server { fn drop(&mut self) { // SAFETY: signals this server's own process group. unsafe { libc::kill(-self.pid(), libc::SIGKILL) }; let _ = self.child.wait(); } } /// A gateway granting `burn` a budget no test reaches. fn gateway(declared: bool) -> Gateway { let mut g = Gateway::new("burner", &[("burn", 50_000_000_000)], steward_sk()); g.set_upstream_settle_bound(TEST_SETTLE); if declared { g.declare_upstream_self_contained(); } g } fn burn(id: i64, n: u64) -> String { json!({"jsonrpc":"2.0","id":id,"method":"tools/call","params":{"name":"burn","arguments":{"n":n}}}).to_string() } /// A `burn` call that also asks the server to hand `k` lookups to a system /// service (the delegating server below; others ignore `k`). fn burn_delegating(id: i64, n: u64, k: u64) -> String { json!({"jsonrpc":"2.0","id":id,"method":"tools/call","params":{"name":"burn","arguments":{"n":n,"k":k}}}).to_string() } /// What one allowed call came back as. struct Call { cost_uj: u64, /// The call's sealed receipt, verified and decoded. sealed: Value, /// The plaintext beside the response, without the receipt's bytes. meta: Value, } fn relay(g: &mut Gateway, up: &mut Server, id: i64, n: u64) -> Call { relay_line(g, up, id, n, &burn(id, n)) } /// [`relay`] for a request line of the caller's: its id is `id`, and the /// server answers it `burned `. fn relay_line(g: &mut Gateway, up: &mut Server, id: i64, n: u64, line: &str) -> Call { let Relayed::Allowed { response, cost_uj, .. } = g.relay(line, up).expect("relay") else { panic!("the call was not allowed") }; let pubkey = steward_sk().verifying_key().to_bytes(); let core = jcp_receipt::jcr1::verify_cose_sign1(g.receipts().last().unwrap(), STEWARD, &pubkey).expect("the receipt verifies"); let sealed: Value = jcp_schema::from_canonical_cbor(&core).expect("the receipt decodes"); let response: Value = serde_json::from_str(&response).expect("a JSON response"); let mut meta = response.pointer(&format!("/result/_meta/{}", JCP_META_NS.replace('/', "~1"))).expect("the jcp _meta").clone(); meta.as_object_mut().unwrap().remove("receipt_b64"); assert_eq!(response["id"].as_i64(), Some(id), "the server's own response, relayed: {response}"); assert_eq!(response["result"]["content"][0]["text"].as_str(), Some(format!("burned {n}").as_str()), "{response}"); assert_eq!(sealed["cost_uj"].as_u64(), Some(cost_uj), "{sealed}"); assert_eq!(meta["cost_uj"].as_u64(), Some(cost_uj), "{meta}"); // The invariant: the plaintext grade is the sealed grade. assert_eq!(meta.get("energy_provenance"), sealed.get("energy_provenance"), "{meta} vs {sealed}"); assert_eq!(meta.get("energy_coverage"), sealed.get("energy_coverage"), "{meta} vs {sealed}"); Call { cost_uj, sealed, meta } } fn grade(c: &Call) -> String { match c.sealed.get("energy_coverage").and_then(Value::as_str) { Some(coverage) => format!("{}, {coverage}", c.sealed["energy_provenance"].as_str().unwrap_or("?")), None => c.sealed["energy_provenance"].as_str().unwrap_or("?").to_string(), } } fn the_log_verifies(g: &Gateway) { let pubkey = steward_sk().verifying_key().to_bytes(); assert_eq!(jcp_receipt::verify_log(g.receipts(), STEWARD, &pubkey), Ok(g.receipts().len())); } /// A single-process server whose operator declared it self-contained: each /// call, from a bare round trip to a few milliseconds of CPU, is sealed /// `HwShunt` and whole, read on the server's own process tree, and its /// figure is the oracle's advance across the relay, exact to the µJ. #[test] fn a_declared_self_contained_server_is_read_whole_on_its_own_process() { let _one = alone(); let mut s = Server::single_process(); let pid = s.pid(); let mut g = gateway(true); assert_eq!( g.call_meter(&s), CallMeter::UpstreamTree { pid: pid as u32 }, "a child of this process, on a host whose per-process counter is live (every host this runs on)" ); for (rep, n) in [0u64, 20_000, 150_000, 300_000, 0].into_iter().enumerate() { let before = energy_nj(&quiet(pid)); let c = relay(&mut g, &mut s, rep as i64, n); let oracle = energy_nj(&quiet(pid)) - before; eprintln!( "upstream tree: declared single process, loop {n} rep {rep}: {} µJ [{}], oracle {oracle} nJ, _meta {}", c.cost_uj, grade(&c), c.meta ); assert!(oracle > 0, "the server's count advanced over the call"); assert_eq!(c.sealed["energy_provenance"].as_str(), Some("HwShunt"), "{}", c.sealed); assert!(c.sealed.get("energy_coverage").is_none(), "whole: {}", c.sealed); assert_eq!(c.meta["energy_domain"].as_str(), Some("upstream-process-tree"), "{}", c.meta); assert!(c.meta.get("energy_partial_because").is_none(), "{}", c.meta); assert_eq!(c.cost_uj, oracle.div_ceil(1_000), "the figure against the oracle's advance, to the µJ"); } the_log_verifies(&g); } /// The same server with no declaration: the same reading, to the µJ, but /// sealed `Partial`, because nothing in the tree can show that the server /// hands no work to a process outside it; the plaintext says that is why. #[test] fn an_undeclared_server_is_read_partial_for_want_of_the_declaration() { let _one = alone(); let mut s = Server::single_process(); let pid = s.pid(); let mut g = gateway(false); assert_eq!(g.call_meter(&s), CallMeter::UpstreamTree { pid: pid as u32 }); for (rep, n) in [0u64, 150_000, 300_000].into_iter().enumerate() { let before = energy_nj(&quiet(pid)); let c = relay(&mut g, &mut s, rep as i64, n); let oracle = energy_nj(&quiet(pid)) - before; eprintln!( "upstream tree: undeclared single process, loop {n} rep {rep}: {} µJ [{}], oracle {oracle} nJ, _meta {}", c.cost_uj, grade(&c), c.meta ); assert_eq!(c.sealed["energy_provenance"].as_str(), Some("HwShunt"), "{}", c.sealed); assert_eq!(c.sealed["energy_coverage"].as_str(), Some("Partial"), "{}", c.sealed); assert_eq!(c.meta["energy_domain"].as_str(), Some("upstream-process-tree"), "{}", c.meta); assert_eq!(c.meta["energy_partial_because"], json!(["not-declared-self-contained"]), "{}", c.meta); assert_eq!(c.cost_uj, oracle.div_ceil(1_000), "the figure against the oracle's advance, to the µJ"); } the_log_verifies(&g); } /// A server that answers each call with a worker it forks and reaps (a /// shell loop running awk) is `Partial` even when declared: the worker's /// energy is in no counter once it is reaped. The figure is the shell's own /// advance, exactly, and the plaintext names the reaped descendant as why. #[test] fn a_server_that_reaps_a_worker_per_call_is_partial_even_when_declared() { let _one = alone(); let mut s = Server::reaping_a_worker_per_call(); let pid = s.pid(); let mut g = gateway(true); assert_eq!(g.call_meter(&s), CallMeter::UpstreamTree { pid: pid as u32 }); for (rep, n) in [20_000u64, 200_000, 200_000].into_iter().enumerate() { let before = quiet(pid); let c = relay(&mut g, &mut s, rep as i64, n); let after = quiet(pid); let oracle = energy_nj(&after) - energy_nj(&before); let reaped = reaped_cpu_ticks(&after) - reaped_cpu_ticks(&before); eprintln!( "upstream tree: declared shell reaping a worker, loop {n} rep {rep}: {} µJ [{}], oracle (the shell's own) {oracle} nJ, reaped {reaped} ticks, _meta {}", c.cost_uj, grade(&c), c.meta ); assert!(reaped > 0, "setup: the worker ran and was reaped in the call"); assert_eq!(c.sealed["energy_provenance"].as_str(), Some("HwShunt"), "{}", c.sealed); assert_eq!(c.sealed["energy_coverage"].as_str(), Some("Partial"), "{}", c.sealed); assert_eq!(c.meta["energy_domain"].as_str(), Some("upstream-process-tree"), "{}", c.meta); assert_eq!(c.meta["energy_partial_because"], json!(["reaped-descendant"]), "{}", c.meta); assert_eq!(c.cost_uj, oracle.div_ceil(1_000), "the shell's own advance, to the µJ"); } the_log_verifies(&g); } /// A declared server whose worker writes the reply ITSELF, keeps computing /// for a few milliseconds and exits, and is reaped by the server after the /// reply has gone (perl, with 60 idle children beside it, so a walk of its /// tree takes long enough for the reap to land inside one). The worker was /// born in the call and spent energy in it, and is in no counter once /// reaped, so no call is sealed whole — not even one whose stop walk read /// the server before the reap and looked for the worker after it. #[test] fn a_worker_reaped_after_writing_the_reply_never_leaves_a_call_whole() { let _one = alone(); let mut s = Server::spawn( "/usr/bin/perl", &[ "-e", r#"$| = 1; for (1 .. 60) { if (fork() == 0) { sleep 1000 while 1 } } while (my $l = ) { my ($id) = $l =~ /"id":(\d+)/; my ($n) = $l =~ /"n":(\d+)/; $n //= 0; my $p = fork(); if ($p == 0) { print qq({"jsonrpc":"2.0","id":$id,"result":{"content":[{"type":"text","text":"burned $n"}]}}\n); my $x = 0; $x += $_ for 1 .. $n; exit 0 } waitpid($p, 0); }"#, ], ); let pid = s.pid(); let mut g = gateway(true); assert_eq!(g.call_meter(&s), CallMeter::UpstreamTree { pid: pid as u32 }); let _ = relay(&mut g, &mut s, 0, 0); let mut reasons: Vec = Vec::new(); for rep in 1..31 { let c = relay(&mut g, &mut s, rep, 150_000); eprintln!("upstream tree: declared server, worker reaped after its reply, rep {rep}: {} µJ [{}], _meta {}", c.cost_uj, grade(&c), c.meta); assert_eq!(c.sealed["energy_provenance"].as_str(), Some("HwShunt"), "{}", c.sealed); assert_eq!(c.sealed["energy_coverage"].as_str(), Some("Partial"), "a call whose worker is in no reading sealed whole: {}", c.meta); let because = c.meta["energy_partial_because"].as_array().expect("why it is partial"); assert!( because.iter().any(|k| matches!(k.as_str(), Some("reaped-descendant" | "exited" | "unsettled-at-start" | "unsettled-at-stop" | "tree-changing"))), "{}", c.meta ); reasons.extend(because.iter().filter_map(Value::as_str).map(str::to_string)); } reasons.sort(); reasons.dedup(); eprintln!("upstream tree: declared server, worker reaped after its reply: reasons seen {reasons:?}"); the_log_verifies(&g); } /// A Metal compute server speaking the tests' JSON-RPC, built from this /// source when the test first runs: each request's `arguments.n` picks what /// it does before it replies — 0 nothing on the GPU, 1 a compute kernel on /// the command queue it keeps, 2 the same kernel on a queue made for the /// request and released after it — and it replies `burned `. const METAL_SERVER: &str = r##"#import #include #include #include static NSString *src = @"#include \nusing namespace metal;\n" "kernel void burn(device float *o [[buffer(0)]], uint i [[thread_position_in_grid]]) {\n" " float x = float(i); for (int k = 0; k < 20000; k++) { x = x * 1.0000001f + 0.5f; } o[i] = x; }\n"; static long field(const char *line, const char *key) { const char *p = strstr(line, key); return p ? strtol(p + strlen(key), NULL, 10) : 0; } int main(void) { @autoreleasepool { id d = MTLCreateSystemDefaultDevice(); if (!d) return 1; NSError *err = nil; id lib = [d newLibraryWithSource:src options:nil error:&err]; if (!lib) return 1; id ps = [d newComputePipelineStateWithFunction:[lib newFunctionWithName:@"burn"] error:&err]; id kept = [d newCommandQueue]; id buf = [d newBufferWithLength:sizeof(float) * 65536 options:MTLResourceStorageModeShared]; char line[4096]; while (fgets(line, sizeof line, stdin)) { long rid = field(line, "\"id\":"), n = field(line, "\"n\":"); if (n == 1 || n == 2) { @autoreleasepool { id q = n == 2 ? [d newCommandQueue] : kept; id b = [q commandBuffer]; id e = [b computeCommandEncoder]; [e setComputePipelineState:ps]; [e setBuffer:buf offset:0 atIndex:0]; [e dispatchThreads:MTLSizeMake(65536, 1, 1) threadsPerThreadgroup:MTLSizeMake(256, 1, 1)]; [e endEncoding]; [b commit]; [b waitUntilCompleted]; } } printf("{\"jsonrpc\":\"2.0\",\"id\":%ld,\"result\":{\"content\":[{\"type\":\"text\",\"text\":\"burned %ld\"}]}}\n", rid, n); fflush(stdout); } } return 0; } "##; /// A declared server that holds a GPU client is never sealed whole, whatever /// a call did: its kept queue's work is recorded GPU time ("gpu"), and a /// call whose work ran on a queue released inside it, or that did no GPU /// work, leaves the server's recorded time unmoved, which does not show it /// used none ("gpu-unknown"). The meter waits only 200 ms for the server to /// go quiet: a Metal process keeps threads that wake on their own, and a /// server still running at an end is a reason of its own. #[test] fn a_declared_server_holding_a_gpu_client_is_never_sealed_whole() { let _one = alone(); let dir = std::path::Path::new(env!("CARGO_TARGET_TMPDIR")); let source = dir.join("upstream_tree_metal_server.m"); let binary = dir.join(format!("upstream_tree_metal_server-{}", std::process::id())); std::fs::write(&source, METAL_SERVER).expect("write the Metal server's source"); let out = Command::new("/usr/bin/cc") .args(["-fobjc-arc", "-O2", "-framework", "Foundation", "-framework", "Metal", "-o"]) .arg(&binary) .arg(&source) .output() .expect("run the C compiler"); assert!(out.status.success(), "the Metal server did not build:\n{}", String::from_utf8_lossy(&out.stderr)); let mut s = Server::spawn(binary.to_str().unwrap(), &[]); let pid = s.pid(); let mut g = gateway(true); g.set_upstream_settle_bound(Duration::from_millis(200)); assert_eq!(g.call_meter(&s), CallMeter::UpstreamTree { pid: pid as u32 }); for (rep, n) in [1u64, 2, 0, 2, 1, 0].into_iter().enumerate() { let c = relay(&mut g, &mut s, rep as i64, n); eprintln!("upstream tree: declared Metal server, mode {n} rep {rep}: {} µJ [{}], _meta {}", c.cost_uj, grade(&c), c.meta); assert_eq!(c.sealed["energy_provenance"].as_str(), Some("HwShunt"), "{}", c.sealed); assert_eq!(c.sealed["energy_coverage"].as_str(), Some("Partial"), "a GPU-holding server sealed whole: {}", c.meta); let because: Vec<&str> = c.meta["energy_partial_because"].as_array().expect("why").iter().filter_map(Value::as_str).collect(); let want = if n == 1 { "gpu" } else { "gpu-unknown" }; assert!(because.contains(&want) || because.contains(&"gpu"), "{want} not among {because:?}"); } the_log_verifies(&g); } /// An upstream naming a process this one may not read (launchd, another /// user's) gets no tree meter, and its calls are metered machine-wide, /// sealed as that meter grades them. #[test] fn an_upstream_naming_a_process_this_one_cannot_read_is_metered_machine_wide() { // SAFETY: no preconditions. if unsafe { libc::geteuid() } == 0 { eprintln!("upstream fallback: running as root, launchd is readable; nothing to check"); return; } let _one = alone(); let mut s = Server::single_process(); s.named = Some(1); let mut g = gateway(true); let machine = if g.metering_is_hardware() { CallMeter::MachineHardware } else { CallMeter::Estimator }; assert_eq!(g.call_meter(&s), machine); let c = relay(&mut g, &mut s, 1, 20_000); eprintln!("upstream fallback: pid 1 named: {} µJ [{}], _meta {}", c.cost_uj, grade(&c), c.meta); assert_eq!(c.meta["energy_domain"].as_str(), Some("machine"), "{}", c.meta); assert!(c.meta.get("energy_partial_because").is_none(), "{}", c.meta); if !g.metering_is_hardware() { assert_eq!(c.sealed["energy_provenance"].as_str(), Some("Estimator"), "{}", c.sealed); } the_log_verifies(&g); } /// A server for the delegated-work test, speaking the tests' JSON-RPC, built /// from this source when the test first runs: each request burns its /// `arguments.n` loop, then looks up `arguments.k` account names no account /// has (`getpwnam`). A system directory service does each lookup, on a /// message that carries this process's voucher, so the kernel bills the /// service's CPU work for it to this process. When it looked any up, it /// then waits — up to 5 s, in 1 ms sleeps — for its own billed counts to /// move before it replies `burned `: a servicing thread is credited when /// it next changes voucher, which can be after the reply, and the test is /// of how a bill that landed in the call is sealed, not of when the kernel /// lands it. One process, no children, no GPU client. const DELEGATING_SERVER: &str = r##"#include #include #include #include #include #include #include static long field(const char *line, const char *key) { const char *p = strstr(line, key); return p ? strtol(p + strlen(key), NULL, 10) : 0; } static void billed(unsigned long long *nj, unsigned long long *ticks) { struct rusage_info_v6 r; memset(&r, 0, sizeof r); proc_pid_rusage(getpid(), RUSAGE_INFO_V6, (rusage_info_t *)&r); *nj = r.ri_billed_energy; *ticks = r.ri_billed_system_time; } int main(void) { char line[4096]; unsigned long seq = 0; while (fgets(line, sizeof line, stdin)) { long rid = field(line, "\"id\":"), n = field(line, "\"n\":"), k = field(line, "\"k\":"); volatile unsigned long long x = 1; for (long i = 0; i < n; i++) x = x * 6364136223846793005ULL + 1; if (k > 0) { unsigned long long nj0, t0, nj1, t1; billed(&nj0, &t0); for (long i = 0; i < k; i++) { char name[96]; snprintf(name, sizeof name, "no-such-account-%d-%lu", (int)getpid(), seq++); (void)getpwnam(name); } for (int waited = 0; waited < 5000; waited++) { billed(&nj1, &t1); if (nj1 > nj0 || t1 > t0) break; usleep(1000); } } printf("{\"jsonrpc\":\"2.0\",\"id\":%ld,\"result\":{\"content\":[{\"type\":\"text\",\"text\":\"burned %ld\"}]}}\n", rid, n); fflush(stdout); } return 0; } "##; /// The server's rusage once its billed counts have held for 300 ms: /// whatever was billed for its start, which can land late, has landed. /// Panics after 20 s. fn billed_quiet(pid: i32) { let deadline = Instant::now() + Duration::from_secs(20); loop { let a = oracle_rusage(pid).expect("oracle read"); std::thread::sleep(Duration::from_millis(300)); let b = oracle_rusage(pid).expect("oracle read"); if (billed_nj(&a), billed_ticks(&a)) == (billed_nj(&b), billed_ticks(&b)) { return; } assert!(Instant::now() < deadline, "the oracle never saw {pid}'s billed counts hold"); } } /// A server whose operator declared it self-contained, and whose calls hand /// work to a system service, is sealed `Partial` with `delegated` among the /// reasons: the kernel billed the service's CPU work to it, which is /// evidence the declaration does not hold for that call, and the evidence /// wins. Its figure holds the bill: at least the oracle's own advance less /// what it serviced plus what it saw billed (the oracle reads inside the /// call's span, so a bill landing late is in the gateway's figure and maybe /// not the oracle's), rounded up to the µJ, and more than the server's own /// advance alone. The same server's calls that delegate nothing come first /// (once anything billed for its start has landed), as the control: whole, /// and exact to the µJ against the oracle. #[test] fn a_declared_server_whose_calls_delegate_is_sealed_partial_for_the_delegation() { let _one = alone(); let dir = std::path::Path::new(env!("CARGO_TARGET_TMPDIR")); let source = dir.join("upstream_tree_delegating_server.c"); let binary = dir.join(format!("upstream_tree_delegating_server-{}", std::process::id())); std::fs::write(&source, DELEGATING_SERVER).expect("write the delegating server's source"); let out = Command::new("/usr/bin/cc").args(["-O2", "-o"]).arg(&binary).arg(&source).output().expect("run the C compiler"); assert!(out.status.success(), "the delegating server did not build:\n{}", String::from_utf8_lossy(&out.stderr)); let mut s = Server::spawn(binary.to_str().unwrap(), &[]); let pid = s.pid(); let mut g = gateway(true); assert_eq!(g.call_meter(&s), CallMeter::UpstreamTree { pid: pid as u32 }); // The first call starts the server's work; its start's bills land after. let _ = relay(&mut g, &mut s, 0, 0); billed_quiet(pid); for rep in 1..4 { let before = quiet(pid); let c = relay(&mut g, &mut s, rep, 150_000); let after = quiet(pid); let own = energy_nj(&after) - energy_nj(&before); eprintln!( "upstream tree: declared delegating server, control rep {rep}: {} µJ [{}], oracle own {own} nJ, billed {} nJ, _meta {}", c.cost_uj, grade(&c), billed_nj(&after) - billed_nj(&before), c.meta ); assert_eq!((billed_nj(&after), billed_ticks(&after)), (billed_nj(&before), billed_ticks(&before)), "setup: the control billed nothing"); assert_eq!(c.sealed["energy_provenance"].as_str(), Some("HwShunt"), "{}", c.sealed); assert!(c.sealed.get("energy_coverage").is_none(), "the control is whole: {}", c.sealed); assert!(c.meta.get("energy_partial_because").is_none(), "{}", c.meta); assert_eq!(c.cost_uj, own.div_ceil(1_000), "the control against the oracle's advance, to the µJ"); } for rep in 4..8 { let before = quiet(pid); let c = relay_line(&mut g, &mut s, rep, 20_000, &burn_delegating(rep, 20_000, 3)); let after = quiet(pid); let own = i128::from(energy_nj(&after) - energy_nj(&before)); let serviced = i128::from(serviced_nj(&after) - serviced_nj(&before)); let billed = billed_nj(&after) - billed_nj(&before); let billed_cpu = billed_ticks(&after) - billed_ticks(&before); eprintln!( "upstream tree: declared delegating server, delegating rep {rep}: {} µJ [{}], oracle own {own} nJ, serviced {serviced} nJ, billed {billed} nJ / {billed_cpu} ticks, _meta {}", c.cost_uj, grade(&c), c.meta ); assert!(billed > 0 || billed_cpu > 0, "setup: the oracle saw no work billed to the server"); assert_eq!(c.sealed["energy_provenance"].as_str(), Some("HwShunt"), "{}", c.sealed); assert_eq!(c.sealed["energy_coverage"].as_str(), Some("Partial"), "a delegating call sealed whole under the declaration: {}", c.sealed); assert_eq!(c.meta["energy_domain"].as_str(), Some("upstream-process-tree"), "{}", c.meta); assert_eq!(c.meta["energy_partial_because"], json!(["delegated"]), "{}", c.meta); let floor = u64::try_from(own - serviced + i128::from(billed)).expect("a positive oracle figure"); assert!(c.cost_uj >= floor.div_ceil(1_000), "the figure ({} µJ) is under the oracle's own − serviced + billed ({floor} nJ)", c.cost_uj); assert!(billed == 0 || i128::from(c.cost_uj) * 1_000 > own - serviced, "the bill was not added to the server's own figure"); } the_log_verifies(&g); }