use std::{
collections::BTreeMap,
fs,
io::Write,
net::{SocketAddr, TcpListener, TcpStream},
path::{Path, PathBuf},
sync::{
Arc,
atomic::{AtomicBool, Ordering},
},
thread::JoinHandle,
time::{Duration, Instant, SystemTime, UNIX_EPOCH},
};
use base64::Engine as _;
use serde_json::{Value, json};
use crate::{
integrity::{FrontendIntegrityInputs, create_explicit_run_integrity},
lifecycle::{note_kept_evidence, remove_stored_tree_deferred},
run_store::{RunIntegrity, RunMetadata},
rust_project::PreparedRustProject,
served_tests::ServedTest,
};
const ENTRY_FILE: &str = "supercov-worker-entry.mjs";
const REGISTRY_FILE: &str = "supercov-worker-registry.mjs";
const REGISTRY: &str = r#"// Generated by Supercov: remembers this isolate's wasm instances.
const instances = (globalThis.__supercovWasm ??= []);
const Instance = WebAssembly.Instance;
function Recorded(module, imports) {
const instance = new Instance(module, imports);
instances.push(instance);
return instance;
}
Recorded.prototype = Instance.prototype;
WebAssembly.Instance = Recorded;
const instantiate = WebAssembly.instantiate;
WebAssembly.instantiate = async function (source, imports) {
const result = await instantiate.call(WebAssembly, source, imports);
instances.push(result instanceof Instance ? result : result.instance);
return result;
};
"#;
const TAKE: &str = r#"(() => {
const seen = (globalThis.__supercovTaken ??= new WeakSet());
const out = [];
for (const instance of globalThis.__supercovWasm ?? []) {
const exports = instance.exports;
for (const name of Object.keys(exports)) {
if (!name.startsWith('__supercov_region_')) continue;
const crate = name.slice('__supercov_region_'.length);
const packed = BigInt.asUintN(64, BigInt(exports[name]()));
const at = Number(packed >> 32n), length = Number(packed & 0xffffffffn);
if (!length || !exports.memory) continue;
const view = new DataView(exports.memory.buffer, at, length);
const first = !seen.has(exports);
if (!first && view.getUint32(40, true) === 0 && view.getUint32(44, true) === 0) continue;
seen.add(exports);
const bytes = new Uint8Array(exports.memory.buffer, at, length);
const encode = (from, to) => {
let text = '';
for (let index = from; index < to; index += 0x8000)
text += String.fromCharCode.apply(null, bytes.subarray(index, Math.min(to, index + 0x8000)));
return btoa(text);
};
const bodyAt = view.getUint32(32, true);
const segments = [[0, encode(0, bodyAt)]];
for (let page = 0; bodyAt + page * 4096 < length; page++) {
if (bytes[128 + (page >> 3)] & (1 << (page & 7))) {
const from = bodyAt + page * 4096;
segments.push([from, encode(from, Math.min(length, from + 4096))]);
}
}
out.push([crate, length, segments]);
exports['__supercov_clear_' + crate]();
}
}
return out;
})()"#;
const POLL_INTERVAL: Duration = Duration::from_millis(5);
#[derive(Debug, Clone, PartialEq, Eq)]
struct WorkerConfig {
path: PathBuf,
main: String,
inspector_port: Option<u16>,
}
fn worker_configs(root: &Path) -> Vec<PathBuf> {
const SKIPPED: &[&str] = &[
"node_modules",
"target",
"build",
"dist",
".git",
".supercov",
".wrangler",
];
let mut found = Vec::new();
let mut directories = vec![(root.to_path_buf(), 0)];
while let Some((directory, depth)) = directories.pop() {
let Ok(entries) = fs::read_dir(&directory) else {
continue;
};
for entry in entries.flatten() {
let path = entry.path();
let name = entry.file_name();
let Some(name) = name.to_str() else {
continue;
};
let Ok(kind) = entry.file_type() else {
continue;
};
if kind.is_dir() {
if depth < 6 && !SKIPPED.contains(&name) {
directories.push((path, depth + 1));
}
continue;
}
if !kind.is_file()
|| !matches!(name, "wrangler.toml" | "wrangler.json" | "wrangler.jsonc")
{
continue;
}
let builds_rust = read_config(&path).is_some_and(|(_, document)| {
build_command(&document).is_some_and(|command| command.contains("worker-build"))
});
if builds_rust && path.with_file_name("Cargo.toml").is_file() {
found.push(path);
}
}
}
found.sort();
found
}
fn read_config(path: &Path) -> Option<(String, Value)> {
let text = fs::read_to_string(path).ok()?;
let document = if path
.extension()
.is_some_and(|extension| extension == "toml")
{
let table = text.parse::<toml::Table>().ok()?;
serde_json::to_value(table).ok()?
} else {
serde_json::from_str(&strip_jsonc(&text)).ok()?
};
Some((text, document))
}
fn build_command(document: &Value) -> Option<String> {
match document.get("build")?.get("command")? {
Value::String(command) => Some(command.clone()),
Value::Array(parts) => Some(
parts
.iter()
.filter_map(Value::as_str)
.collect::<Vec<_>>()
.join(" "),
),
_ => None,
}
}
fn parse_config(path: &Path) -> Option<WorkerConfig> {
let (_, document) = read_config(path)?;
Some(WorkerConfig {
path: path.to_owned(),
main: document.get("main")?.as_str()?.to_owned(),
inspector_port: document
.get("dev")
.and_then(|dev| dev.get("inspector_port"))
.and_then(Value::as_u64)
.and_then(|port| u16::try_from(port).ok()),
})
}
fn strip_jsonc(text: &str) -> String {
let mut out = String::with_capacity(text.len());
let mut characters = text.chars().peekable();
while let Some(character) = characters.next() {
match character {
'"' => {
out.push('"');
while let Some(inner) = characters.next() {
out.push(inner);
match inner {
'\\' => {
if let Some(escaped) = characters.next() {
out.push(escaped);
}
}
'"' => break,
_ => {}
}
}
}
'/' if characters.peek() == Some(&'/') => {
for inner in characters.by_ref() {
if inner == '\n' {
out.push('\n');
break;
}
}
}
'/' if characters.peek() == Some(&'*') => {
characters.next();
let mut previous = '\0';
for inner in characters.by_ref() {
if previous == '*' && inner == '/' {
break;
}
previous = inner;
}
}
_ => out.push(character),
}
}
let mut cleaned = String::with_capacity(out.len());
let mut in_string = false;
let mut escaped = false;
let bytes = out.as_bytes();
for (index, character) in out.char_indices() {
if in_string {
cleaned.push(character);
if escaped {
escaped = false;
} else if character == '\\' {
escaped = true;
} else if character == '"' {
in_string = false;
}
continue;
}
if character == '"' {
in_string = true;
} else if character == ',' {
let next = bytes[index + 1..]
.iter()
.find(|byte| !byte.is_ascii_whitespace());
if matches!(next, Some(b'}' | b']')) {
continue;
}
}
cleaned.push(character);
}
cleaned
}
fn rewrite_config(config: &WorkerConfig, entry: &str, port: Option<u16>) -> Result<(), String> {
let path = &config.path;
let text = fs::read_to_string(path).map_err(|error| format!("{}: {error}", path.display()))?;
let rewritten = if path
.extension()
.is_some_and(|extension| extension == "toml")
{
let mut document = text
.parse::<toml_edit::DocumentMut>()
.map_err(|error| format!("{}: {error}", path.display()))?;
document["main"] = toml_edit::value(entry);
if let (None, Some(port)) = (config.inspector_port, port) {
if !document.contains_key("dev") {
document["dev"] = toml_edit::table();
}
document["dev"]["inspector_port"] = toml_edit::value(i64::from(port));
}
document.to_string()
} else {
let mut document: Value = serde_json::from_str(&strip_jsonc(&text))
.map_err(|error| format!("{}: {error}", path.display()))?;
document["main"] = json!(entry);
if let (None, Some(port)) = (config.inspector_port, port) {
if !document.get("dev").is_some_and(Value::is_object) {
document["dev"] = json!({});
}
document["dev"]["inspector_port"] = json!(port);
}
serde_json::to_string_pretty(&document).map_err(|error| error.to_string())?
};
fs::write(path, rewritten).map_err(|error| format!("{}: {error}", path.display()))
}
fn point_at_entry(config: &WorkerConfig, port: Option<u16>) -> Result<(), String> {
if config.main.trim_start_matches("./") == ENTRY_FILE {
return Ok(());
}
let directory = config.path.parent().unwrap_or(Path::new("."));
fs::write(directory.join(REGISTRY_FILE), REGISTRY).map_err(|error| error.to_string())?;
let main = config.main.trim_start_matches("./");
let entry = format!(
"// Generated by Supercov: the project's entry, with its wasm instances remembered.\nimport './{REGISTRY_FILE}';\nexport * from './{main}';\nexport {{ default }} from './{main}';\n"
);
fs::write(directory.join(ENTRY_FILE), entry).map_err(|error| error.to_string())?;
rewrite_config(config, ENTRY_FILE, port)
}
fn free_port() -> Result<u16, String> {
let listener = TcpListener::bind("127.0.0.1:0").map_err(|error| error.to_string())?;
listener
.local_addr()
.map(|address| address.port())
.map_err(|error| error.to_string())
}
pub(crate) fn wasm_target_cfg(
crate_directory: &Path,
) -> Option<crate::rust_instrumenter::TargetCfg> {
let output = std::process::Command::new("rustc")
.args(["--print", "cfg", "--target", "wasm32-unknown-unknown"])
.current_dir(crate_directory)
.output()
.ok()?;
if !output.status.success() {
return None;
}
let values = String::from_utf8_lossy(&output.stdout)
.lines()
.filter_map(|line| line.parse::<cargo_platform::Cfg>().ok())
.collect::<Vec<_>>();
Some(crate::rust_instrumenter::TargetCfg::new(&values))
}
fn now_ms() -> i64 {
SystemTime::now()
.duration_since(UNIX_EPOCH)
.map(|elapsed| elapsed.as_millis() as i64)
.unwrap_or(0)
}
struct Poller {
stop: Arc<AtomicBool>,
thread: Option<JoinHandle<()>>,
}
impl Poller {
fn start(port: u16, windows: PathBuf) -> Self {
let stop = Arc::new(AtomicBool::new(false));
let flag = Arc::clone(&stop);
let thread = std::thread::Builder::new()
.name("supercov-worker-poller".into())
.spawn(move || poll(port, &windows, &flag))
.ok();
Self { stop, thread }
}
fn finish(mut self) {
self.stop.store(true, Ordering::Release);
if let Some(thread) = self.thread.take() {
let _ = thread.join();
}
}
}
impl Drop for Poller {
fn drop(&mut self) {
self.stop.store(true, Ordering::Release);
}
}
fn poll(port: u16, windows: &Path, stop: &AtomicBool) {
let Ok(mut file) = fs::OpenOptions::new()
.create(true)
.append(true)
.open(windows)
else {
return;
};
let address = SocketAddr::from(([127, 0, 0, 1], port));
while !stop.load(Ordering::Acquire) {
let Ok(stream) = TcpStream::connect_timeout(&address, Duration::from_millis(200)) else {
std::thread::sleep(Duration::from_millis(100));
continue;
};
let _ = stream.set_nodelay(true);
let _ = stream.set_read_timeout(Some(Duration::from_millis(20)));
let Ok((mut socket, _)) =
tungstenite::client::client(format!("ws://127.0.0.1:{port}/ws"), stream)
else {
std::thread::sleep(Duration::from_millis(100));
continue;
};
session(&mut socket, &mut file, stop);
let _ = socket.close(None);
}
}
fn session(socket: &mut tungstenite::WebSocket<TcpStream>, file: &mut fs::File, stop: &AtomicBool) {
use tungstenite::{Error, Message};
let send = |socket: &mut tungstenite::WebSocket<TcpStream>, message: Value| {
socket
.send(Message::text(message.to_string()))
.map_err(|_| ())
};
if send(socket, json!({"id": 1, "method": "Runtime.enable"})).is_err() {
return;
}
let mut contexts = Vec::<i64>::new();
let mut next_id = 2_u64;
let mut awaiting: Option<(u64, Instant)> = None;
let mut window_start = now_ms();
let mut last_take = Instant::now() - POLL_INTERVAL;
let mut final_sent = false;
loop {
let stopping = stop.load(Ordering::Acquire);
if stopping && awaiting.is_none() && (final_sent || contexts.is_empty()) {
return;
}
if awaiting.is_none()
&& let Some(context) = contexts.last()
&& (stopping || last_take.elapsed() >= POLL_INTERVAL)
{
let id = next_id;
next_id += 1;
let request = json!({
"id": id,
"method": "Runtime.evaluate",
"params": {"expression": TAKE, "contextId": context, "returnByValue": true},
});
if send(socket, request).is_err() {
return;
}
awaiting = Some((id, Instant::now()));
last_take = Instant::now();
final_sent = stopping;
}
match socket.read() {
Ok(Message::Text(text)) => {
let Ok(message) = serde_json::from_str::<Value>(text.as_str()) else {
continue;
};
match message.get("method").and_then(Value::as_str) {
Some("Runtime.executionContextCreated") => {
if let Some(id) = message["params"]["context"]["id"].as_i64() {
contexts.push(id);
}
}
Some("Runtime.executionContextDestroyed") => {
if let Some(id) = message["params"]["executionContextId"].as_i64() {
contexts.retain(|context| *context != id);
}
}
Some("Runtime.executionContextsCleared") => contexts.clear(),
_ => {}
}
if let Some((id, _)) = awaiting
&& message.get("id").and_then(Value::as_u64) == Some(id)
{
awaiting = None;
let end = now_ms();
let regions = &message["result"]["result"]["value"];
if regions
.as_array()
.is_some_and(|regions| !regions.is_empty())
{
let line = json!({"start": window_start, "end": end, "regions": regions});
let _ = writeln!(file, "{line}");
}
window_start = end;
}
}
Ok(Message::Close(_)) => return,
Ok(_) => {}
Err(Error::Io(error))
if matches!(
error.kind(),
std::io::ErrorKind::WouldBlock | std::io::ErrorKind::TimedOut
) => {}
Err(_) => return,
}
if awaiting.is_some_and(|(_, sent)| sent.elapsed() > Duration::from_secs(5)) {
awaiting = None;
}
}
}
struct Window {
start: i64,
end: i64,
regions: Vec<(String, Vec<u8>)>,
}
fn read_windows(directory: &Path) -> Result<Vec<Window>, String> {
let mut files = fs::read_dir(directory)
.map(|entries| {
entries
.filter_map(Result::ok)
.map(|entry| entry.path())
.filter(|path| {
path.extension()
.is_some_and(|extension| extension == "windows")
})
.collect::<Vec<_>>()
})
.unwrap_or_default();
files.sort();
let mut windows = Vec::new();
for file in &files {
let text =
fs::read_to_string(file).map_err(|error| format!("{}: {error}", file.display()))?;
for line in text.lines() {
let Ok(value) = serde_json::from_str::<Value>(line) else {
continue;
};
if let Some(window) = window_from(&value) {
windows.push(window);
}
}
}
Ok(windows)
}
fn window_from(value: &Value) -> Option<Window> {
let mut regions = Vec::new();
for region in value.get("regions")?.as_array()? {
let crate_key = region.get(0)?.as_str()?.to_owned();
let length = usize::try_from(region.get(1)?.as_u64()?).ok()?;
let mut bytes = vec![0_u8; length];
for segment in region.get(2)?.as_array()? {
let offset = usize::try_from(segment.get(0)?.as_u64()?).ok()?;
let data = base64::engine::general_purpose::STANDARD
.decode(segment.get(1)?.as_str()?)
.ok()?;
bytes
.get_mut(offset..offset + data.len())?
.copy_from_slice(&data);
}
regions.push((crate_key, bytes));
}
Some(Window {
start: value.get("start")?.as_i64()?,
end: value.get("end")?.as_i64()?,
regions,
})
}
pub struct RustWorkerServer {
project: PreparedRustProject,
pollers: Vec<Poller>,
integrity: RunIntegrity,
source_inputs: crate::source_manifest::Inputs,
source_roots: Option<Vec<String>>,
directory: PathBuf,
}
impl RustWorkerServer {
pub fn present(root: &Path) -> bool {
!worker_configs(root).is_empty()
}
pub fn prepare(
root: &Path,
workspace: &Path,
run_id: &str,
command: &[String],
diagnostics: &mut dyn Write,
) -> Result<Vec<Self>, String> {
let configs = worker_configs(workspace);
let mut by_workspace = BTreeMap::<PathBuf, Vec<WorkerConfig>>::new();
for path in configs {
let Some(config) = parse_config(&path) else {
writeln!(
diagnostics,
"[supercov] {} names no `main`; its Worker is not measured",
path.display()
)
.map_err(|error| error.to_string())?;
continue;
};
let crate_directory = path.parent().unwrap_or(workspace).to_owned();
let cargo_root = cargo_workspace_root(&crate_directory).unwrap_or(crate_directory);
by_workspace.entry(cargo_root).or_default().push(config);
}
let mut servers = Vec::new();
for (index, (cargo_root, configs)) in by_workspace.into_iter().enumerate() {
let Ok(relative) = cargo_root.strip_prefix(workspace) else {
continue;
};
let source_roots = crate::source_discovery::configured_source_roots(
&std::env::vars().collect::<BTreeMap<_, _>>(),
);
let mut integrity_inputs = crate::rust_run::collect_integrity_inputs(root, command)?;
crate::source_discovery::fold_roots_into_configuration(
&mut integrity_inputs.execution_configuration,
source_roots.as_deref(),
);
let source_inputs =
crate::source_capture::capture(root, "rust", integrity_inputs.assertion_paths())?;
let integrity = create_explicit_run_integrity(
root,
&integrity_inputs,
&FrontendIntegrityInputs::embedded_rust(),
)
.map_err(|error| error.to_string())?;
let cfg = wasm_target_cfg(&cargo_root);
let mut project =
crate::rust_project::prepare_rust_project_for(&cargo_root, None, cfg.as_ref())
.map_err(|error| error.to_string())?;
crate::rust_wasm_evidence::root_paths(&mut project, relative);
let directory = root
.join(".supercov/work")
.join(run_id)
.join(format!("rust-worker-{index}"));
let evidence = directory.join("evidence");
fs::create_dir_all(&evidence).map_err(|error| error.to_string())?;
let mut pollers = Vec::new();
for (number, config) in configs.iter().enumerate() {
let port = match config.inspector_port {
Some(port) => port,
None => free_port()?,
};
point_at_entry(config, Some(port))?;
pollers.push(Poller::start(
port,
evidence.join(format!("{number:04}.windows")),
));
}
servers.push(Self {
project,
pollers,
integrity,
source_inputs,
source_roots,
directory,
});
}
Ok(servers)
}
pub fn source_files(&self) -> usize {
self.project.source_files.len()
}
pub fn source_fingerprint(&self) -> String {
self.integrity.fingerprint.source.clone()
}
pub fn environment(&self) -> BTreeMap<String, String> {
let mut rustflags = std::env::var("RUSTFLAGS").unwrap_or_default();
if !rustflags.is_empty() {
rustflags.push(' ');
}
rustflags.push_str("--cap-lints=warn -A unused_parens");
BTreeMap::from([("RUSTFLAGS".into(), rustflags)])
}
pub fn publish(
mut self,
root: &Path,
_evidence_run_id: &str,
metadata: RunMetadata,
served: &[ServedTest],
diagnostics: &mut dyn Write,
) -> Result<Option<(PathBuf, RunMetadata)>, String> {
for poller in std::mem::take(&mut self.pollers) {
poller.finish();
}
let evidence_directory = self.directory.join("evidence");
let result = read_windows(&evidence_directory).and_then(|windows| {
crate::rust_wasm_evidence::publish(
root,
crate::rust_wasm_evidence::WasmRun {
language: "Rust Worker",
archive: "rust-worker-evidence.raw.gz",
project: &self.project,
integrity: &self.integrity,
source_inputs: &self.source_inputs,
source_roots: self.source_roots.clone(),
},
windows
.into_iter()
.map(|window| crate::rust_wasm_evidence::WasmRecord {
owner: crate::rust_wasm_evidence::Owner::Window(window.start, window.end),
regions: window.regions,
})
.collect(),
metadata,
served,
diagnostics,
)
});
let result = result.map_err(|error| {
let run = self
.directory
.parent()
.and_then(Path::file_name)
.map(|name| name.to_string_lossy().into_owned())
.unwrap_or_default();
note_kept_evidence(
root,
&evidence_directory,
&format!("{run}-rust-worker"),
error,
)
});
let _ = remove_stored_tree_deferred(root, &self.directory);
result
}
}
pub const REAL_WRANGLER: &str = "SUPERCOV_WRANGLER";
pub const WRANGLER_COMMAND: &str = "__wrangler";
pub const WRANGLER_POLL_COMMAND: &str = "__wrangler-poll";
pub struct CargoTestWorkers {
environment: Vec<(String, std::ffi::OsString)>,
}
impl CargoTestWorkers {
pub fn prepare(workspace: &Path, tools: &Path) -> Result<Option<Self>, String> {
let configs = worker_configs(workspace)
.iter()
.filter_map(|path| parse_config(path))
.collect::<Vec<_>>();
if configs.is_empty() {
return Ok(None);
}
let path = std::env::var_os("PATH").unwrap_or_default();
let Some(real) = on_path(&path, "wrangler") else {
return Ok(None);
};
for config in &configs {
point_at_entry(config, None)?;
}
let supercov = std::env::current_exe().map_err(|error| error.to_string())?;
fs::create_dir_all(tools).map_err(|error| format!("{}: {error}", tools.display()))?;
write_stand_in(tools, &supercov)?;
let mut directories = vec![tools.to_owned()];
directories.extend(std::env::split_paths(&path));
let joined = std::env::join_paths(directories).map_err(|error| error.to_string())?;
Ok(Some(Self {
environment: vec![
("PATH".into(), joined),
(REAL_WRANGLER.into(), real.into_os_string()),
],
}))
}
pub fn environment(&self) -> impl Iterator<Item = (&str, &std::ffi::OsStr)> {
self.environment
.iter()
.map(|(name, value)| (name.as_str(), value.as_os_str()))
}
}
fn on_path(path: &std::ffi::OsStr, name: &str) -> Option<PathBuf> {
let names = if cfg!(windows) {
vec![
format!("{name}.cmd"),
format!("{name}.exe"),
name.to_owned(),
]
} else {
vec![name.to_owned()]
};
std::env::split_paths(path).find_map(|directory| {
names
.iter()
.map(|name| directory.join(name))
.find(|candidate| fs::metadata(candidate).is_ok_and(|metadata| metadata.is_file()))
})
}
fn write_stand_in(tools: &Path, supercov: &Path) -> Result<(), String> {
#[cfg(unix)]
{
use std::os::unix::fs::PermissionsExt;
let quoted = supercov.to_string_lossy().replace('\'', r"'\''");
let script = tools.join("wrangler");
fs::write(
&script,
format!("#!/bin/sh\nexec '{quoted}' {WRANGLER_COMMAND} \"$@\"\n"),
)
.map_err(|error| format!("{}: {error}", script.display()))?;
fs::set_permissions(&script, fs::Permissions::from_mode(0o755))
.map_err(|error| format!("{}: {error}", script.display()))?;
}
#[cfg(windows)]
{
let script = tools.join("wrangler.cmd");
fs::write(
&script,
format!("@\"{}\" {WRANGLER_COMMAND} %*\r\n", supercov.display()),
)
.map_err(|error| format!("{}: {error}", script.display()))?;
}
Ok(())
}
fn starts_dev(arguments: &[std::ffi::OsString]) -> bool {
const VALUED: &[&str] = &["--cwd", "-c", "--config", "-e", "--env", "--env-file"];
let mut arguments = arguments.iter().map(|argument| argument.to_string_lossy());
while let Some(argument) = arguments.next() {
if VALUED.contains(&argument.as_ref()) {
arguments.next();
} else if !argument.starts_with('-') {
return argument == "dev";
}
}
false
}
fn runs_remote(arguments: &[std::ffi::OsString]) -> bool {
arguments.iter().any(|argument| {
matches!(
argument.to_string_lossy().as_ref(),
"--remote" | "-r" | "--remote=true"
)
})
}
const REMOTE_NOTE: &str = "worker-remote-";
pub fn started_remote(directory: &Path) -> bool {
fs::read_dir(directory).is_ok_and(|entries| {
entries.filter_map(Result::ok).any(|entry| {
entry
.file_name()
.to_str()
.is_some_and(|name| name.starts_with(REMOTE_NOTE))
})
})
}
fn named_inspector_port(arguments: &[std::ffi::OsString]) -> Option<u16> {
let mut arguments = arguments.iter().map(|argument| argument.to_string_lossy());
while let Some(argument) = arguments.next() {
if argument == "--inspector-port" {
return arguments.next()?.parse().ok();
}
if let Some(port) = argument.strip_prefix("--inspector-port=") {
return port.parse().ok();
}
}
None
}
pub fn launch_wrangler(arguments: Vec<std::ffi::OsString>) -> i32 {
let Some(real) = std::env::var_os(REAL_WRANGLER) else {
eprintln!("[supercov] {WRANGLER_COMMAND} runs only as the wrangler of a measured test");
return 127;
};
let mut arguments = arguments;
let evidence = std::env::var_os("SUPERCOV_RUST_EVIDENCE_DIR").map(PathBuf::from);
if starts_dev(&arguments)
&& let Some(evidence) = evidence
{
if runs_remote(&arguments) {
let _ = fs::create_dir_all(&evidence);
let _ = fs::write(
evidence.join(format!("{REMOTE_NOTE}{}", std::process::id())),
"",
);
return exec_wrangler(&real, &arguments);
}
let port = match named_inspector_port(&arguments) {
Some(port) => Some(port),
None => free_port().ok().inspect(|port| {
arguments.push("--inspector-port".into());
arguments.push(port.to_string().into());
}),
};
if let (Some(port), Ok(supercov)) = (port, std::env::current_exe()) {
let windows = evidence.join(format!("worker-{}.windows", std::process::id()));
let _ = std::process::Command::new(supercov)
.arg(WRANGLER_POLL_COMMAND)
.arg(std::process::id().to_string())
.arg(port.to_string())
.arg(&windows)
.stdin(std::process::Stdio::null())
.stdout(std::process::Stdio::null())
.stderr(std::process::Stdio::null())
.spawn();
}
}
exec_wrangler(&real, &arguments)
}
fn exec_wrangler(real: &std::ffi::OsStr, arguments: &[std::ffi::OsString]) -> i32 {
let rustflags = worker_rustflags(&std::env::var("RUSTFLAGS").unwrap_or_default());
let mut command = std::process::Command::new(real);
command.args(arguments).env("RUSTFLAGS", rustflags);
#[cfg(unix)]
{
use std::os::unix::process::CommandExt;
let error = command.exec();
eprintln!(
"[supercov] could not start {}: {error}",
Path::new(&real).display()
);
126
}
#[cfg(not(unix))]
{
match command.status() {
Ok(status) => status.code().unwrap_or(1),
Err(error) => {
eprintln!(
"[supercov] could not start {}: {error}",
Path::new(&real).display()
);
126
}
}
}
}
fn worker_rustflags(inherited: &str) -> String {
let mut kept = Vec::new();
let mut words = inherited.split_whitespace();
while let Some(word) = words.next() {
if word == "--cap-lints" {
words.next();
} else if !word.starts_with("--cap-lints=") {
kept.push(word);
}
}
kept.extend(["--cap-lints=warn", "-A", "unused_parens"]);
kept.join(" ")
}
pub fn poll_wrangler(arguments: Vec<std::ffi::OsString>) -> i32 {
let [pid, port, windows] = arguments.as_slice() else {
return 2;
};
let (Ok(pid), Ok(port)) = (
pid.to_string_lossy().parse::<u32>(),
port.to_string_lossy().parse::<u16>(),
) else {
return 2;
};
let windows = PathBuf::from(windows);
let part = partial(&windows);
let poller = Poller::start(port, part.clone());
while process_alive(pid) {
std::thread::sleep(Duration::from_millis(20));
}
poller.finish();
if fs::symlink_metadata(&part).is_ok() {
let _ = fs::rename(&part, &windows);
}
0
}
fn partial(windows: &Path) -> PathBuf {
let mut name = windows.as_os_str().to_owned();
name.push(".part");
PathBuf::from(name)
}
#[cfg(unix)]
fn process_alive(pid: u32) -> bool {
let Ok(pid) = libc::pid_t::try_from(pid) else {
return false;
};
let answered = unsafe { libc::kill(pid, 0) };
answered == 0 || std::io::Error::last_os_error().raw_os_error() == Some(libc::EPERM)
}
#[cfg(not(unix))]
fn process_alive(_pid: u32) -> bool {
false
}
const POLLER_SETTLE: Duration = Duration::from_secs(3);
pub fn materialize_worker_windows(
directory: &Path,
token: &str,
) -> Result<std::collections::BTreeSet<String>, String> {
let listed = |extension: &str| -> Vec<PathBuf> {
let mut files = fs::read_dir(directory)
.map(|entries| {
entries
.filter_map(Result::ok)
.map(|entry| entry.path())
.filter(|path| {
path.file_name()
.and_then(|name| name.to_str())
.is_some_and(|name| {
name.starts_with("worker-") && name.ends_with(extension)
})
})
.collect::<Vec<_>>()
})
.unwrap_or_default();
files.sort();
files
};
let started = Instant::now();
while !listed(".windows.part").is_empty() && started.elapsed() < POLLER_SETTLE {
std::thread::sleep(Duration::from_millis(20));
}
let files = listed(".windows")
.into_iter()
.chain(listed(".windows.part"))
.collect::<Vec<_>>();
let mut crates = std::collections::BTreeSet::new();
for (number, file) in files.iter().enumerate() {
let text =
fs::read_to_string(file).map_err(|error| format!("{}: {error}", file.display()))?;
for (line, value) in text
.lines()
.filter_map(|line| serde_json::from_str::<Value>(line).ok())
.enumerate()
{
let Some(window) = window_from(&value) else {
continue;
};
for (index, (crate_key, bytes)) in window.regions.into_iter().enumerate() {
if region_token(&bytes) != Some(token) {
continue;
}
let region = directory.join(format!(
"{token}-worker-{number:04}-{line:06}-{index}.region"
));
fs::write(®ion, &bytes)
.map_err(|error| format!("{}: {error}", region.display()))?;
crates.insert(crate_key);
}
}
fs::remove_file(file).map_err(|error| format!("{}: {error}", file.display()))?;
}
Ok(crates)
}
pub(crate) fn region_token(bytes: &[u8]) -> Option<&str> {
let field = bytes.get(56..72)?;
let end = field
.iter()
.position(|byte| *byte == 0)
.unwrap_or(field.len());
std::str::from_utf8(&field[..end]).ok()
}
pub(crate) fn worker_crate_directories(workspace: &Path) -> Vec<String> {
worker_configs(workspace)
.iter()
.filter_map(|config| config.parent()?.strip_prefix(workspace).ok())
.map(|directory| directory.to_string_lossy().replace('\\', "/"))
.collect()
}
pub(crate) fn cargo_workspace_root(directory: &Path) -> Option<PathBuf> {
let output = std::process::Command::new("cargo")
.args(["locate-project", "--workspace", "--message-format", "plain"])
.current_dir(directory)
.output()
.ok()?;
if !output.status.success() {
return None;
}
let manifest = PathBuf::from(String::from_utf8_lossy(&output.stdout).trim());
manifest.parent().map(Path::to_path_buf)
}
#[cfg(test)]
mod tests {
use super::*;
#[test]
fn jsonc_comments_and_trailing_commas_are_dropped_and_strings_kept() {
let text = r#"{
// the Worker
"name": "a // not a comment",
/* block */ "main": "build/worker/shim.mjs",
"build": { "command": "worker-build --release", },
"list": [1, 2, ],
"quote": "say \"hi\", ]"
}"#;
let value: Value = serde_json::from_str(&strip_jsonc(text)).unwrap();
assert_eq!(value["name"], "a // not a comment");
assert_eq!(value["main"], "build/worker/shim.mjs");
assert_eq!(value["build"]["command"], "worker-build --release");
assert_eq!(value["list"], json!([1, 2]));
assert_eq!(value["quote"], "say \"hi\", ]");
}
#[test]
fn the_worker_build_caps_lints_at_warn_whatever_it_inherited() {
assert_eq!(worker_rustflags(""), "--cap-lints=warn -A unused_parens");
assert_eq!(
worker_rustflags("-C debuginfo=1 --cap-lints=allow"),
"-C debuginfo=1 --cap-lints=warn -A unused_parens"
);
assert_eq!(
worker_rustflags("--cap-lints allow -Dwarnings"),
"-Dwarnings --cap-lints=warn -A unused_parens"
);
}
#[test]
fn the_stand_in_reads_dev_and_an_inspector_port_wherever_they_stand() {
let os = |arguments: &[&str]| {
arguments
.iter()
.map(std::ffi::OsString::from)
.collect::<Vec<_>>()
};
assert!(runs_remote(&os(&["dev", "--remote"])));
assert!(runs_remote(&os(&["dev", "-r", "--port", "8787"])));
assert!(!runs_remote(&os(&["dev", "--remote=false"])));
assert!(!runs_remote(&os(&["dev", "--local"])));
assert!(starts_dev(&os(&["dev", "--local"])));
assert!(starts_dev(&os(&["--cwd", "worker", "dev"])));
assert!(starts_dev(&os(&[
"-c",
"wrangler.toml",
"--env=staging",
"dev"
])));
assert!(starts_dev(&os(&["--local", "dev"])));
assert!(!starts_dev(&os(&["deploy"])));
assert!(!starts_dev(&os(&[])));
assert_eq!(
named_inspector_port(&os(&["dev", "--inspector-port", "9230"])),
Some(9230)
);
assert_eq!(
named_inspector_port(&os(&["dev", "--inspector-port=9231"])),
Some(9231)
);
assert_eq!(named_inspector_port(&os(&["dev", "--port", "8787"])), None);
assert_eq!(
named_inspector_port(&os(&["dev", "--inspector-port"])),
None
);
}
#[test]
fn a_take_becomes_a_region_file_of_the_test_that_started_the_worker() {
let directory = temporary("materialize");
let region = |token: &str| {
let mut bytes = vec![0_u8; 4096 + 256];
bytes[56..56 + token.len()].copy_from_slice(token.as_bytes());
bytes[200] = 7;
bytes
};
let segment =
|bytes: &[u8]| json!([0, base64::engine::general_purpose::STANDARD.encode(bytes)]);
let ours = region("abc123");
let theirs = region("fff999");
let line = json!({
"start": 1,
"end": 2,
"regions": [
["abc123k1", ours.len(), [segment(&ours)]],
["fff999k2", theirs.len(), [segment(&theirs)]],
],
});
fs::write(
directory.join("worker-41.windows"),
format!("{line}\n{{\"start\": 3, \"end\""),
)
.unwrap();
let crates = materialize_worker_windows(&directory, "abc123").unwrap();
assert_eq!(
crates,
std::collections::BTreeSet::from(["abc123k1".to_owned()])
);
let mut names = fs::read_dir(&directory)
.unwrap()
.map(|entry| entry.unwrap().file_name().to_string_lossy().into_owned())
.collect::<Vec<_>>();
names.sort();
assert_eq!(names, ["abc123-worker-0000-000000-0.region"]);
assert_eq!(fs::read(directory.join(&names[0])).unwrap(), ours);
assert!(
materialize_worker_windows(&directory, "abc123")
.unwrap()
.is_empty()
);
assert_eq!(region_token(&ours), Some("abc123"));
assert_eq!(region_token(&[0_u8; 10]), None);
fs::remove_dir_all(&directory).unwrap();
}
#[cfg(unix)]
#[test]
fn the_stand_in_is_put_first_only_where_a_wrangler_is_installed() {
let workspace = temporary("stand-in");
fs::create_dir_all(workspace.join("worker/src")).unwrap();
fs::write(
workspace.join("worker/Cargo.toml"),
"[package]\nname = \"w\"\n",
)
.unwrap();
fs::write(workspace.join("worker/src/lib.rs"), "").unwrap();
fs::write(
workspace.join("worker/wrangler.toml"),
"name = \"w\"\nmain = \"build/worker/shim.mjs\"\n\n[build]\ncommand = \"worker-build --release\"\n",
)
.unwrap();
let bin = workspace.join("bin");
fs::create_dir_all(&bin).unwrap();
let real = bin.join("wrangler");
fs::write(&real, "#!/bin/sh\n").unwrap();
let found = on_path(bin.as_os_str(), "wrangler");
assert_eq!(found.as_deref(), Some(real.as_path()));
assert_eq!(on_path(std::ffi::OsStr::new(""), "wrangler"), None);
let tools = workspace.join("tools");
let config = parse_config(&workspace.join("worker/wrangler.toml")).unwrap();
point_at_entry(&config, None).unwrap();
let pointed = fs::read_to_string(workspace.join("worker/wrangler.toml")).unwrap();
assert!(
pointed.contains(&format!("main = \"{ENTRY_FILE}\"")),
"{pointed}"
);
assert!(!pointed.contains("inspector_port"), "{pointed}");
let entry = fs::read_to_string(workspace.join("worker").join(ENTRY_FILE)).unwrap();
point_at_entry(
&parse_config(&workspace.join("worker/wrangler.toml")).unwrap(),
None,
)
.unwrap();
assert_eq!(
fs::read_to_string(workspace.join("worker").join(ENTRY_FILE)).unwrap(),
entry
);
assert!(entry.contains("./build/worker/shim.mjs"), "{entry}");
fs::create_dir_all(&tools).unwrap();
write_stand_in(&tools, Path::new("/opt/it's/supercov")).unwrap();
let script = fs::read_to_string(tools.join("wrangler")).unwrap();
assert_eq!(
script,
format!("#!/bin/sh\nexec '/opt/it'\\''s/supercov' {WRANGLER_COMMAND} \"$@\"\n")
);
fs::remove_dir_all(&workspace).unwrap();
}
fn temporary(name: &str) -> PathBuf {
let path = std::env::temp_dir().join(format!(
"supercov-rust-worker-{}-{}-{name}",
std::process::id(),
SystemTime::now()
.duration_since(UNIX_EPOCH)
.unwrap()
.as_nanos()
));
fs::create_dir_all(&path).unwrap();
path
}
#[test]
fn only_a_configuration_that_builds_with_worker_build_beside_a_crate_is_a_rust_worker() {
let root = temporary("configs");
let write = |relative: &str, text: &str| {
let path = root.join(relative);
fs::create_dir_all(path.parent().unwrap()).unwrap();
fs::write(path, text).unwrap();
};
write(
"worker/wrangler.toml",
"main = \"build/worker/shim.mjs\"\n[build]\ncommand = \"cargo install -q worker-build && worker-build --release\"\n",
);
write("worker/Cargo.toml", "[package]\nname = \"w\"\n");
write("js/wrangler.toml", "main = \"src/index.js\"\n");
write(
"orphan/wrangler.jsonc",
"{ \"main\": \"x.mjs\", // c\n \"build\": { \"command\": \"worker-build\" } }",
);
write(
"json/wrangler.jsonc",
"{ \"main\": \"build/worker/shim.mjs\", /* c */ \"build\": { \"command\": \"worker-build --dev\", }, }",
);
write("json/Cargo.toml", "[package]\nname = \"j\"\n");
write(
"node_modules/dep/wrangler.toml",
"main = \"m\"\n[build]\ncommand = \"worker-build\"\n",
);
write("node_modules/dep/Cargo.toml", "");
let found = worker_configs(&root)
.into_iter()
.map(|path| path.strip_prefix(&root).unwrap().to_owned())
.collect::<Vec<_>>();
assert_eq!(
found,
[
PathBuf::from("json/wrangler.jsonc"),
PathBuf::from("worker/wrangler.toml")
]
);
fs::remove_dir_all(root).unwrap();
}
#[test]
fn the_copy_points_at_the_entry_and_an_inspector_port_and_keeps_the_rest() {
let root = temporary("rewrite");
let toml_path = root.join("wrangler.toml");
fs::write(
&toml_path,
"# the Worker\nname = \"grades\"\nmain = \"build/worker/shim.mjs\"\n\n[build]\ncommand = \"worker-build --release\"\n",
)
.unwrap();
let config = parse_config(&toml_path).unwrap();
rewrite_config(&config, ENTRY_FILE, Some(9444)).unwrap();
let rewritten = fs::read_to_string(&toml_path).unwrap();
assert!(rewritten.contains("# the Worker"), "{rewritten}");
let (_, document) = read_config(&toml_path).unwrap();
assert_eq!(document["main"], ENTRY_FILE);
assert_eq!(document["dev"]["inspector_port"], 9444);
assert_eq!(document["build"]["command"], "worker-build --release");
let json_path = root.join("wrangler.jsonc");
fs::write(
&json_path,
"{\n // mine\n \"main\": \"./build/worker/shim.mjs\",\n \"dev\": { \"inspector_port\": 9555 },\n}\n",
)
.unwrap();
let config = parse_config(&json_path).unwrap();
assert_eq!(config.inspector_port, Some(9555));
assert_eq!(config.main, "./build/worker/shim.mjs");
rewrite_config(&config, ENTRY_FILE, Some(1)).unwrap();
let (_, document) = read_config(&json_path).unwrap();
assert_eq!(document["main"], ENTRY_FILE);
assert_eq!(document["dev"]["inspector_port"], 9555);
fs::remove_dir_all(root).unwrap();
}
#[test]
fn a_window_is_rebuilt_from_its_header_and_touched_pages() {
let encode = |bytes: &[u8]| base64::engine::general_purpose::STANDARD.encode(bytes);
let value = json!({
"start": 10,
"end": 15,
"regions": [["abc", 8192, [[0, encode(&[1, 2, 3])], [4096, encode(&[9; 4])]]]],
});
let window = window_from(&value).unwrap();
assert_eq!((window.start, window.end), (10, 15));
let (crate_key, bytes) = &window.regions[0];
assert_eq!(crate_key, "abc");
assert_eq!(bytes.len(), 8192);
assert_eq!(&bytes[..4], &[1, 2, 3, 0]);
assert_eq!(&bytes[4096..4101], &[9, 9, 9, 9, 0]);
let bad = json!({"start": 0, "end": 1, "regions": [["a", 4, [[2, encode(&[1, 2, 3])]]]]});
assert!(window_from(&bad).is_none());
}
}