use std::io::{Read, Write};
use std::net::{Ipv4Addr, SocketAddr, TcpListener, TcpStream, ToSocketAddrs};
use std::path::{Path, PathBuf};
use std::time::{Duration, Instant};
use anyhow::Result;
pub const BIND: &str = "127.0.0.1";
pub const HEALTH_TIMEOUT: Duration = Duration::from_secs(2);
pub const START_TIMEOUT: Duration = Duration::from_secs(20);
pub const POLL_INTERVAL: Duration = Duration::from_millis(250);
pub const BIN_NAME: &str = "rustxng";
#[derive(Debug, Clone, PartialEq, Eq)]
pub enum Error {
BinNotFound,
Spawn(String),
Unresponsive(String),
State(String),
}
impl std::fmt::Display for Error {
fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
match self {
Self::BinNotFound => {
write!(f, "{}", missing_binary_hint(crate::locale::Locale::English))
}
Self::Spawn(e) => write!(f, "could not start rustxng: {e}"),
Self::Unresponsive(url) => write!(
f,
"rustxng did not answer /healthz at {url} within {}s",
START_TIMEOUT.as_secs()
),
Self::State(e) => write!(f, "cannot record the rustxng port: {e}"),
}
}
}
impl std::error::Error for Error {}
pub fn missing_binary_hint(locale: crate::locale::Locale) -> String {
match locale {
crate::locale::Locale::English => {
"rustxng binary not found; set RUSTXNG_BIN=/path/to/rustxng, put rustxng on PATH, or install it to ~/.local/bin/rustxng (rustdsh stays MIT: the AGPL engine runs as a separate process)".to_string()
}
crate::locale::Locale::Chinese => {
"未找到 rustxng 可执行文件;请设置 RUSTXNG_BIN=/path/to/rustxng,或将 rustxng 放入 PATH,或安装到 ~/.local/bin/rustxng(rustdsh 保持 MIT:AGPL 引擎作为独立进程运行)".to_string()
}
crate::locale::Locale::Japanese => {
"rustxng の実行ファイルが見つかりません。RUSTXNG_BIN=/path/to/rustxng を設定するか、rustxng を PATH に置くか、~/.local/bin/rustxng へインストールしてください(rustdsh は MIT のまま: AGPL エンジンは別プロセスとして動きます)".to_string()
}
}
}
#[cfg(test)]
static BIN_MODE: std::sync::atomic::AtomicU8 = std::sync::atomic::AtomicU8::new(0);
fn env(name: &str) -> Option<String> {
std::env::var(name)
.ok()
.map(|value| value.trim().to_string())
.filter(|value| !value.is_empty())
}
fn exe_name(name: &str) -> String {
if cfg!(windows) {
format!("{name}.exe")
} else {
name.to_string()
}
}
pub fn binary() -> Option<PathBuf> {
#[cfg(test)]
{
use std::sync::atomic::Ordering;
match BIN_MODE.load(Ordering::Relaxed) {
1 => {
return env("RUSTXNG_BIN")
.map(PathBuf::from)
.filter(|p| p.is_file())
}
2 => return None,
_ => {}
}
}
binary_with(
&|name| env(name),
&std::env::var_os("PATH"),
home_dir().as_deref(),
)
}
fn binary_with(
lookup: &dyn Fn(&str) -> Option<String>,
path: &Option<std::ffi::OsString>,
home: Option<&Path>,
) -> Option<PathBuf> {
if let Some(explicit) = lookup("RUSTXNG_BIN") {
let candidate = PathBuf::from(explicit);
return candidate.is_file().then_some(candidate);
}
let name = exe_name(BIN_NAME);
if let Some(path) = path {
for dir in std::env::split_paths(path) {
let candidate = dir.join(&name);
if candidate.is_file() {
return Some(candidate);
}
}
}
for dir in fallback_dirs(home) {
let candidate = dir.join(&name);
if candidate.is_file() {
return Some(candidate);
}
}
None
}
pub fn fallback_dirs(home: Option<&Path>) -> Vec<PathBuf> {
let mut dirs = Vec::new();
if let Some(home) = home {
dirs.push(home.join(".local").join("bin"));
}
dirs.push(PathBuf::from("/usr/local/bin"));
dirs
}
fn home_dir() -> Option<PathBuf> {
#[cfg(windows)]
{
if let (Ok(drive), Ok(path)) = (std::env::var("HOMEDRIVE"), std::env::var("HOMEPATH")) {
if !drive.is_empty() && !path.is_empty() {
return Some(PathBuf::from(format!("{drive}{path}")));
}
}
}
env("HOME").map(PathBuf::from)
}
pub fn base_url() -> Option<String> {
if let Some(url) = env("RUSTXNG_URL") {
return Some(url.trim_end_matches('/').to_string());
}
if let Some(port) = configured_port() {
if !port_in_use(port) {
return Some(url_for(port));
}
}
recorded_state().map(|(port, _)| url_for(port))
}
pub fn state_file() -> PathBuf {
PathBuf::from(crate::inspect::dsh_home()).join("rustxng.port")
}
pub fn log_file() -> PathBuf {
PathBuf::from(crate::inspect::dsh_home())
.join("logs")
.join("rustxng.log")
}
pub fn recorded_state() -> Option<(u16, Option<u32>)> {
read_state(&state_file())
}
fn read_state(path: &Path) -> Option<(u16, Option<u32>)> {
let text = std::fs::read_to_string(path).ok()?;
parse_state(&text)
}
fn parse_state(text: &str) -> Option<(u16, Option<u32>)> {
let mut fields = text.split_whitespace();
let port = fields
.next()?
.parse::<u16>()
.ok()
.filter(|port| *port != 0)?;
let pid = fields.next().and_then(|pid| pid.parse::<u32>().ok());
Some((port, pid))
}
fn write_state(path: &Path, port: u16, pid: Option<u32>) -> std::io::Result<()> {
if let Some(parent) = path.parent() {
std::fs::create_dir_all(parent)?;
}
let body = match pid {
Some(pid) => format!("{port}\t{pid}\n"),
None => format!("{port}\n"),
};
std::fs::write(path, body)
}
fn clear_state(path: &Path) {
let _ = std::fs::remove_file(path);
}
pub fn free_port() -> Result<u16> {
let listener = TcpListener::bind((Ipv4Addr::LOCALHOST, 0))?;
Ok(listener.local_addr()?.port())
}
pub fn port_in_use(port: u16) -> bool {
probe_port(port, Duration::from_millis(300))
}
fn probe_port(port: u16, timeout: Duration) -> bool {
let address = SocketAddr::from((Ipv4Addr::LOCALHOST, port));
TcpStream::connect_timeout(&address, timeout).is_ok()
}
pub(crate) fn configured_port() -> Option<u16> {
env("RUSTXNG_PORT")
.and_then(|port| port.parse::<u16>().ok())
.filter(|port| *port != 0)
}
fn url_for(port: u16) -> String {
format!("http://{BIND}:{port}")
}
pub fn health(base: &str) -> bool {
let Ok(response) = http_get(base, "/healthz", HEALTH_TIMEOUT) else {
return false;
};
if !response.healthy() {
return false;
}
match response.json() {
Some(value) => match value.get("status").and_then(|status| status.as_str()) {
Some(status) => {
status.eq_ignore_ascii_case("ok") || status.eq_ignore_ascii_case("healthy")
}
None => true,
},
None => true,
}
}
pub fn ensure() -> Result<String, Error> {
#[cfg(test)]
let _lock = tests::env_lock();
let state_path = state_file();
if let Some(url) = env("RUSTXNG_URL") {
if health(&url) {
return Ok(url);
}
return Err(Error::Unresponsive(url));
}
let known = configured_port().or_else(|| recorded_state().map(|(port, _)| port));
if let Some(port) = known {
let url = url_for(port);
if port_in_use(port) && health(&url) {
return Ok(url);
}
}
let binary = match binary() {
Some(binary) => binary,
None => match self_install() {
Ok(Some(path)) => path,
_ => return Err(Error::BinNotFound),
},
};
let mut ports = Vec::new();
match configured_port() {
Some(port) => ports.push(port),
None => {
if let Some((recorded, _)) = recorded_state() {
if !port_in_use(recorded) {
ports.push(recorded);
}
}
ports.push(free_port().map_err(|e| Error::Spawn(e.to_string()))?);
}
}
let mut last = Error::BinNotFound;
for port in ports {
if port_in_use(port) {
return Err(Error::Spawn(format!(
"{BIND}:{port} is already in use by another process"
)));
}
match start_engine(&binary, port, &state_path) {
Ok(url) => return Ok(url),
Err(e) => last = e,
}
}
Err(last)
}
fn start_engine(binary: &Path, port: u16, path: &Path) -> Result<String, Error> {
let log = log_file();
if let Some(parent) = log.parent() {
std::fs::create_dir_all(parent).map_err(|e| Error::Spawn(e.to_string()))?;
}
let handle = std::fs::OpenOptions::new()
.create(true)
.append(true)
.open(&log)
.map_err(|e| Error::Spawn(e.to_string()))?;
let mut child =
spawn_detached(binary, port, handle).map_err(|e| Error::Spawn(e.to_string()))?;
if let Err(e) = write_state(path, port, Some(child.id())) {
kill_and_reap(&mut child);
clear_state(path);
return Err(Error::State(e.to_string()));
}
let url = url_for(port);
let deadline = Instant::now() + START_TIMEOUT;
while Instant::now() < deadline {
if health(&url) {
return Ok(url);
}
std::thread::sleep(POLL_INTERVAL);
}
kill_and_reap(&mut child);
clear_state(path);
Err(Error::Unresponsive(url))
}
fn kill_and_reap(child: &mut std::process::Child) {
let _ = child.kill();
let _ = child.wait();
}
#[cfg(unix)]
fn spawn_detached(
binary: &Path,
port: u16,
log: std::fs::File,
) -> std::io::Result<std::process::Child> {
use std::os::unix::process::CommandExt;
let mut command = std::process::Command::new(binary);
build_command(&mut command, port);
command
.stdin(std::process::Stdio::null())
.stdout(std::process::Stdio::from(log.try_clone()?))
.stderr(std::process::Stdio::from(log));
unsafe {
command.pre_exec(|| {
libc::setsid();
libc::signal(libc::SIGHUP, libc::SIG_IGN);
Ok(())
});
}
command.spawn()
}
#[cfg(not(unix))]
fn spawn_detached(
binary: &Path,
port: u16,
log: std::fs::File,
) -> std::io::Result<std::process::Child> {
let mut command = std::process::Command::new(binary);
build_command(&mut command, port);
command
.stdin(std::process::Stdio::null())
.stdout(std::process::Stdio::from(log.try_clone()?))
.stderr(std::process::Stdio::from(log));
command.spawn()
}
fn build_command(command: &mut std::process::Command, port: u16) {
if let Some(settings) = env("RUSTXNG_SETTINGS") {
let path = PathBuf::from(settings);
if path.is_file() {
command.arg("--config").arg(path);
}
}
command
.arg("--bind")
.arg(BIND)
.arg("--port")
.arg(port.to_string());
}
pub fn stop() -> Result<bool> {
let Some((_port, pid)) = recorded_state() else {
clear_state(&state_file());
return Ok(false);
};
let mut killed = false;
if let Some(pid) = pid {
if process_is_rustxng(pid) {
killed = terminate(pid);
reap_if_ours(pid);
}
}
clear_state(&state_file());
Ok(killed)
}
#[cfg(unix)]
fn reap_if_ours(pid: u32) {
let mut status = 0;
unsafe {
libc::waitpid(pid as libc::pid_t, &mut status, libc::WNOHANG);
}
}
#[cfg(not(unix))]
fn reap_if_ours(_pid: u32) {}
pub fn process_is_rustxng(pid: u32) -> bool {
#[cfg(target_os = "linux")]
{
std::fs::read_to_string(format!("/proc/{pid}/comm"))
.map(|name| name.trim().contains(BIN_NAME))
.unwrap_or(false)
}
#[cfg(all(unix, not(target_os = "linux")))]
{
std::process::Command::new("ps")
.args(["-p", &pid.to_string(), "-o", "comm="])
.output()
.map(|out| {
String::from_utf8_lossy(&out.stdout)
.trim()
.contains(BIN_NAME)
})
.unwrap_or(false)
}
#[cfg(not(unix))]
{
let _ = pid;
false
}
}
#[cfg(unix)]
fn terminate(pid: u32) -> bool {
let target = pid as libc::pid_t;
if unsafe { libc::kill(target, libc::SIGTERM) } != 0 {
return false;
}
let deadline = Instant::now() + Duration::from_secs(5);
while Instant::now() < deadline {
if !process_running(pid) {
return true;
}
std::thread::sleep(Duration::from_millis(50));
}
unsafe { libc::kill(target, libc::SIGKILL) };
true
}
#[cfg(unix)]
fn process_running(pid: u32) -> bool {
#[cfg(target_os = "linux")]
{
std::fs::read_to_string(format!("/proc/{pid}/stat"))
.map(|stat| {
let state = stat.rsplit(')').next().unwrap_or_default();
!state.trim_start().starts_with('Z')
})
.unwrap_or(false)
}
#[cfg(not(target_os = "linux"))]
{
unsafe { libc::kill(pid as libc::pid_t, 0) == 0 }
}
}
#[cfg(not(unix))]
fn terminate(_pid: u32) -> bool {
false
}
pub fn status() -> serde_json::Value {
let binary = binary();
let recorded = recorded_state();
let url = env("RUSTXNG_URL")
.map(|url| url.trim_end_matches('/').to_string())
.or_else(|| configured_port().map(url_for))
.or_else(|| recorded.map(|(port, _)| url_for(port)));
let running = url.as_deref().is_some_and(health);
serde_json::json!({
"binary": binary.as_ref().map(|path| path.display().to_string()),
"url": url,
"running": running,
"port": recorded.map(|(port, _)| port),
"pid": recorded.and_then(|(_, pid)| pid),
"state_file": state_file().display().to_string(),
"log_file": log_file().display().to_string(),
"hint": if binary.is_none() {
Some(missing_binary_hint(crate::locale::Locale::English))
} else {
None
},
})
}
pub fn cmd_status(json_out: bool, locale: crate::locale::Locale) -> Result<()> {
let value = status();
if json_out {
println!("{}", serde_json::to_string_pretty(&value)?);
return Ok(());
}
let binary = value
.get("binary")
.and_then(|v| v.as_str())
.map(str::to_string);
let running = value
.get("running")
.and_then(|v| v.as_bool())
.unwrap_or(false);
let url = value
.get("url")
.and_then(|v| v.as_str())
.map(str::to_string);
match &binary {
Some(path) => println!(
"{}",
match locale {
crate::locale::Locale::English => format!("rustxng binary: {path}"),
crate::locale::Locale::Chinese => format!("rustxng 可执行文件:{path}"),
crate::locale::Locale::Japanese => format!("rustxng 実行ファイル: {path}"),
}
),
None => println!("{}", missing_binary_hint(locale)),
}
if let Some(url) = url {
let state = if running {
match locale {
crate::locale::Locale::English => "running",
crate::locale::Locale::Chinese => "运行中",
crate::locale::Locale::Japanese => "稼働中",
}
} else {
match locale {
crate::locale::Locale::English => "not answering",
crate::locale::Locale::Chinese => "无响应",
crate::locale::Locale::Japanese => "応答なし",
}
};
println!(
"{}",
match locale {
crate::locale::Locale::English => format!("url: {url} ({state})"),
crate::locale::Locale::Chinese => format!("地址:{url}({state})"),
crate::locale::Locale::Japanese => format!("URL: {url}({state})"),
}
);
} else {
println!(
"{}",
match locale {
crate::locale::Locale::English => "no recorded instance in $DSH_HOME",
crate::locale::Locale::Chinese => "$DSH_HOME 中没有已记录的实例",
crate::locale::Locale::Japanese => "$DSH_HOME に記録されたインスタンスはありません",
}
);
}
Ok(())
}
pub fn cmd_start(locale: crate::locale::Locale) -> Result<()> {
let url = ensure().map_err(anyhow::Error::new)?;
println!(
"{}",
match locale {
crate::locale::Locale::English => format!("rustxng ready at {url}"),
crate::locale::Locale::Chinese => format!("rustxng 已就绪:{url}"),
crate::locale::Locale::Japanese => format!("rustxng は {url} で利用できます"),
}
);
Ok(())
}
pub fn cmd_stop(locale: crate::locale::Locale) -> Result<()> {
let terminated = stop()?;
println!(
"{}",
match (locale, terminated) {
(crate::locale::Locale::English, true) => "stopped the recorded rustxng instance",
(crate::locale::Locale::English, false) => "no rustxng instance was running",
(crate::locale::Locale::Chinese, true) => "已停止记录的 rustxng 实例",
(crate::locale::Locale::Chinese, false) => "没有正在运行的 rustxng 实例",
(crate::locale::Locale::Japanese, true) =>
"記録された rustxng インスタンスを停止しました",
(crate::locale::Locale::Japanese, false) => {
"稼働中の rustxng インスタンスはありませんでした"
}
}
);
Ok(())
}
const INSTALL_REPO: &str = "sahenjp/RustXNG";
const INSTALL_TIMEOUT: Duration = Duration::from_secs(300);
pub(crate) fn target_triple() -> &'static str {
if cfg!(target_os = "linux") {
if cfg!(target_arch = "x86_64") {
"x86_64-unknown-linux-gnu"
} else if cfg!(target_arch = "aarch64") {
"aarch64-unknown-linux-gnu"
} else {
"unknown-linux-generic"
}
} else if cfg!(target_os = "macos") {
if cfg!(target_arch = "aarch64") {
"aarch64-apple-darwin"
} else {
"x86_64-apple-darwin"
}
} else if cfg!(all(target_os = "windows", target_arch = "x86_64")) {
"x86_64-pc-windows-msvc"
} else {
"unknown"
}
}
#[derive(Debug, Clone, PartialEq, Eq)]
pub(crate) struct ReleaseAsset {
pub name: String,
pub url: String,
}
pub(crate) fn parse_latest_assets(json: &str) -> Vec<ReleaseAsset> {
let Ok(value) = serde_json::from_str::<serde_json::Value>(json) else {
return Vec::new();
};
value
.get("assets")
.and_then(serde_json::Value::as_array)
.map(|assets| {
assets
.iter()
.filter_map(|asset| {
let name = asset.get("name")?.as_str()?.to_string();
let url = asset.get("browser_download_url")?.as_str()?.to_string();
Some(ReleaseAsset { name, url })
})
.collect()
})
.unwrap_or_default()
}
fn matches_triple(name: &str, triple: &str) -> bool {
let Some(rest) = name.strip_prefix("rustxng-") else {
return false;
};
let base = rest.strip_suffix(".exe").unwrap_or(rest);
match base.strip_suffix(triple) {
Some(tag) => tag.starts_with('v') && tag.ends_with('-'),
None => false,
}
}
fn choose_asset<'a>(assets: &'a [ReleaseAsset], triple: &str) -> Option<&'a ReleaseAsset> {
assets
.iter()
.find(|asset| matches_triple(&asset.name, triple))
}
fn looks_like_executable(bytes: &[u8]) -> bool {
bytes.starts_with(b"\x7fELF")
|| bytes.starts_with(b"MZ")
|| bytes.starts_with(b"\xcf\xfa\xed\xfe")
|| bytes.starts_with(b"\xca\xfe\xba\xbe")
}
fn install_path(home: &Path) -> PathBuf {
fallback_dirs(Some(home))[0].join(exe_name(BIN_NAME))
}
fn install_failed(locale: crate::locale::Locale, detail: &str) -> anyhow::Error {
anyhow::anyhow!(
"{}",
match locale {
crate::locale::Locale::English => format!("rustxng install failed: {detail}"),
crate::locale::Locale::Chinese => format!("rustxng 安装失败:{detail}"),
crate::locale::Locale::Japanese =>
format!("rustxng のインストールに失敗しました: {detail}"),
}
)
}
async fn download_sidecar(dest: &Path, locale: crate::locale::Locale) -> Result<()> {
let client = reqwest::Client::builder()
.user_agent(format!("rdsh/{}", env!("CARGO_PKG_VERSION")))
.connect_timeout(Duration::from_secs(30))
.timeout(INSTALL_TIMEOUT)
.build()?;
let api = format!("https://api.github.com/repos/{INSTALL_REPO}/releases/latest");
let body = client
.get(&api)
.send()
.await?
.error_for_status()?
.text()
.await?;
let assets = parse_latest_assets(&body);
let triple = target_triple();
let asset = choose_asset(&assets, triple).ok_or_else(|| {
let names = assets
.iter()
.map(|a| a.name.as_str())
.collect::<Vec<_>>()
.join(", ");
install_failed(
locale,
&format!("no release asset matches {triple}; the release has: {names}"),
)
})?;
let bytes = client
.get(&asset.url)
.send()
.await?
.error_for_status()?
.bytes()
.await?;
anyhow::ensure!(
!bytes.is_empty() && looks_like_executable(&bytes),
"{}",
install_failed(
locale,
&format!("{} did not download as an executable", asset.name)
)
);
if let Some(parent) = dest.parent() {
std::fs::create_dir_all(parent)?;
}
let partial = dest.with_extension("part");
std::fs::write(&partial, &bytes)?;
#[cfg(unix)]
{
use std::os::unix::fs::PermissionsExt;
std::fs::set_permissions(&partial, std::fs::Permissions::from_mode(0o755))?;
}
std::fs::rename(&partial, dest)?;
Ok(())
}
fn install_sidecar(dest: &Path, locale: crate::locale::Locale) -> Result<PathBuf> {
tokio::runtime::Builder::new_current_thread()
.enable_all()
.build()?
.block_on(download_sidecar(dest, locale))?;
Ok(dest.to_path_buf())
}
fn self_install() -> Result<Option<PathBuf>, Error> {
if std::env::var("RUSTXNG_SKIP_INSTALL")
.ok()
.is_some_and(|v| v == "1")
{
return Ok(None);
}
#[cfg(test)]
{
Ok(None)
}
#[cfg(not(test))]
{
let dest = match home_dir() {
Some(home) => install_path(&home),
None => return Ok(None),
};
eprintln!(
"rustxng sidecar is missing; installing the latest release into {}…",
dest.display()
);
match install_sidecar(&dest, crate::locale::Locale::English) {
Ok(path) => {
eprintln!("rustxng sidecar installed at {}", path.display());
Ok(Some(path))
}
Err(error) => Err(Error::Spawn(format!("auto-install failed: {error:#}"))),
}
}
}
pub fn cmd_install(locale: crate::locale::Locale) -> Result<()> {
if let Some(path) = binary() {
println!(
"{}",
match locale {
crate::locale::Locale::English =>
format!("rustxng is already installed at {}", path.display()),
crate::locale::Locale::Chinese => format!("rustxng 已安装:{}", path.display()),
crate::locale::Locale::Japanese =>
format!("rustxng は導入済みです: {}", path.display()),
}
);
return Ok(());
}
let Some(home) = home_dir() else {
return Err(install_failed(locale, "no home directory"));
};
let dest = install_path(&home);
let path = install_sidecar(&dest, locale)?;
println!(
"{}",
match locale {
crate::locale::Locale::English => format!(
"rustxng installed at {}; start it with: rdsh rustxng start",
path.display()
),
crate::locale::Locale::Chinese => format!(
"rustxng 已安装到 {};启动:rdsh rustxng start",
path.display()
),
crate::locale::Locale::Japanese => format!(
"rustxng を {} へ導入しました。起動: rdsh rustxng start",
path.display()
),
}
);
Ok(())
}
fn http_get(base: &str, target: &str, timeout: Duration) -> Result<HttpResponse> {
let (host, port, prefix) = split_base(base)?;
let deadline = Instant::now() + timeout;
let mut stream = connect(&host, port, deadline)?;
let request = format!(
"GET {prefix}{target} HTTP/1.0\r\nHost: {host}:{port}\r\nUser-Agent: rdsh/{}\r\nAccept: application/json, text/html\r\nConnection: close\r\n\r\n",
env!("CARGO_PKG_VERSION")
);
let remaining = deadline.saturating_duration_since(Instant::now());
stream.set_write_timeout(Some(remaining.max(Duration::from_millis(1))))?;
stream.write_all(request.as_bytes())?;
let mut raw = Vec::new();
stream.set_read_timeout(Some(remaining.max(Duration::from_millis(1))))?;
let mut chunk = [0u8; 4096];
loop {
match stream.read(&mut chunk) {
Ok(0) => break,
Ok(n) => {
if raw.len().saturating_add(n) > 1024 * 1024 {
break;
}
raw.extend_from_slice(&chunk[..n]);
}
Err(e) if e.kind() == std::io::ErrorKind::WouldBlock => break,
Err(e) => return Err(e.into()),
}
}
parse_response(&String::from_utf8_lossy(&raw))
}
fn connect(host: &str, port: u16, deadline: Instant) -> Result<TcpStream> {
let bare = host
.strip_prefix('[')
.and_then(|host| host.strip_suffix(']'))
.unwrap_or(host);
let mut last = None;
for address in (bare, port).to_socket_addrs()? {
let remaining = deadline.saturating_duration_since(Instant::now());
if remaining.is_zero() {
break;
}
match TcpStream::connect_timeout(&address, remaining) {
Ok(stream) => return Ok(stream),
Err(e) => last = Some(e),
}
}
Err(match last {
Some(e) => e.into(),
None => anyhow::anyhow!("no address for {host}:{port}"),
})
}
fn split_base(base: &str) -> Result<(String, u16, String)> {
anyhow::ensure!(
!base.chars().any(|c| c.is_control() || c.is_whitespace())
&& !base.contains(['@', '?', '#']),
"invalid rustxng URL: {base}"
);
let rest = base
.strip_prefix("http://")
.ok_or_else(|| anyhow::anyhow!("only http:// rustxng URLs are supported: {base}"))?;
let (hostport, prefix) = match rest.find('/') {
Some(i) => (&rest[..i], rest[i..].to_string()),
None => (rest, String::new()),
};
let (host, port) = match hostport.rsplit_once(':') {
Some((host, port)) => (
host.to_string(),
port.parse::<u16>()
.ok()
.filter(|port| *port != 0)
.ok_or_else(|| anyhow::anyhow!("invalid rustxng port in {base}"))?,
),
None => (hostport.to_string(), 80),
};
anyhow::ensure!(!host.is_empty(), "invalid rustxng host in {base}");
Ok((host, port, prefix))
}
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct HttpResponse {
pub status: u16,
pub body: String,
}
impl HttpResponse {
pub fn healthy(&self) -> bool {
(200..300).contains(&self.status)
}
pub fn json(&self) -> Option<serde_json::Value> {
serde_json::from_str(&self.body).ok()
}
}
fn parse_response(text: &str) -> Result<HttpResponse> {
let (head, body) = text
.split_once("\r\n\r\n")
.or_else(|| text.split_once("\n\n"))
.ok_or_else(|| anyhow::anyhow!("malformed HTTP response"))?;
let status = head
.lines()
.next()
.and_then(|line| line.split_whitespace().nth(1))
.and_then(|code| code.parse::<u16>().ok())
.ok_or_else(|| anyhow::anyhow!("malformed HTTP status line"))?;
let chunked = head.lines().skip(1).any(|line| {
let lower = line.to_ascii_lowercase();
lower.starts_with("transfer-encoding:") && lower.contains("chunked")
});
let body = if chunked {
decode_chunked(body)
} else {
body.to_string()
};
Ok(HttpResponse { status, body })
}
fn decode_chunked(body: &str) -> String {
let mut out = String::new();
let mut rest = body;
while let Some(eol) = rest.find("\r\n") {
let size = usize::from_str_radix(rest[..eol].trim(), 16).unwrap_or(0);
if size == 0 {
break;
}
let start = eol + 2;
if start + size > rest.len() {
out.push_str(&rest[start..]);
break;
}
out.push_str(&rest[start..start + size]);
rest = &rest[(start + size).min(rest.len())..];
rest = rest.strip_prefix("\r\n").unwrap_or(rest);
}
out
}
#[cfg(test)]
pub(crate) mod tests {
use super::*;
use std::sync::atomic::{AtomicU64, Ordering};
use std::sync::OnceLock;
pub(crate) fn env_lock() -> crate::test_support::EnvLock {
crate::test_support::env_lock()
}
pub(crate) struct BinGuard {
previous_mode: u8,
previous: Option<String>,
}
impl BinGuard {
pub(crate) fn install(path: Option<&Path>) -> Self {
use std::sync::atomic::Ordering;
let previous_mode =
BIN_MODE.swap(if path.is_some() { 1 } else { 2 }, Ordering::Relaxed);
let previous = std::env::var("RUSTXNG_BIN").ok();
match path {
Some(path) => std::env::set_var("RUSTXNG_BIN", path),
None => std::env::remove_var("RUSTXNG_BIN"),
}
Self {
previous_mode,
previous,
}
}
}
impl Drop for BinGuard {
fn drop(&mut self) {
use std::sync::atomic::Ordering;
match &self.previous {
Some(value) => std::env::set_var("RUSTXNG_BIN", value),
None => std::env::remove_var("RUSTXNG_BIN"),
}
BIN_MODE.store(self.previous_mode, Ordering::Relaxed);
}
}
struct EnvGuard {
saved: Vec<(String, Option<String>)>,
}
impl EnvGuard {
fn apply(pairs: &[(&str, Option<&str>)]) -> Self {
let mut saved = Vec::new();
for (name, value) in pairs {
saved.push(((*name).to_string(), std::env::var(name).ok()));
match value {
Some(value) => std::env::set_var(name, value),
None => std::env::remove_var(name),
}
}
Self { saved }
}
}
impl Drop for EnvGuard {
fn drop(&mut self) {
for (name, value) in &self.saved {
match value {
Some(value) => std::env::set_var(name, value),
None => std::env::remove_var(name),
}
}
}
}
struct KillOnDrop(std::process::Child);
impl Drop for KillOnDrop {
fn drop(&mut self) {
let _ = self.0.kill();
let _ = self.0.wait();
}
}
fn next_nonce() -> u64 {
static NEXT: AtomicU64 = AtomicU64::new(0);
NEXT.fetch_add(1, Ordering::Relaxed)
}
fn test_dir(tag: &str) -> PathBuf {
let dir = std::env::temp_dir().join(format!(
"rdsh-rustxng-test-{}-{}-{tag}",
std::process::id(),
next_nonce()
));
let _ = std::fs::remove_dir_all(&dir);
std::fs::create_dir_all(&dir).unwrap();
dir
}
fn write_executable(path: &Path, body: &str) {
std::fs::write(path, body).unwrap();
#[cfg(unix)]
{
use std::os::unix::fs::PermissionsExt;
std::fs::set_permissions(path, std::fs::Permissions::from_mode(0o755)).unwrap();
}
}
fn dirs(path: &Path) -> Option<std::ffi::OsString> {
Some(std::ffi::OsString::from(path.display().to_string()))
}
#[test]
fn explicit_binary_environment_variable_wins_over_path() {
let dir = test_dir("bin-env");
let explicit = dir.join("explicit-rustxng");
write_executable(&explicit, "#!/bin/sh\nexit 0\n");
let path_dir = dir.join("path");
std::fs::create_dir_all(&path_dir).unwrap();
write_executable(&path_dir.join(exe_name(BIN_NAME)), "#!/bin/sh\nexit 0\n");
let lookup = |name: &str| (name == "RUSTXNG_BIN").then(|| explicit.display().to_string());
assert_eq!(
binary_with(&lookup, &dirs(&path_dir), Some(&dir)),
Some(explicit)
);
}
#[test]
fn only_a_real_file_is_accepted() {
let dir = test_dir("bin-none");
let as_dir = dir.join(exe_name(BIN_NAME));
std::fs::create_dir_all(&as_dir).unwrap();
std::fs::create_dir_all(dir.join("nope")).unwrap();
let lookup = |name: &str| (name == "RUSTXNG_BIN").then(|| as_dir.display().to_string());
assert_eq!(
binary_with(&lookup, &None, Some(&dir)),
None,
"a directory named rustxng must not be accepted"
);
let absent =
|name: &str| (name == "RUSTXNG_BIN").then(|| dir.join("nope").display().to_string());
assert_eq!(
binary_with(&absent, &None, Some(&dir)),
None,
"a missing RUSTXNG_BIN path falls through to the other locations"
);
}
#[test]
fn path_lookup_precedes_the_fallback_directories() {
let dir = test_dir("bin-path");
let home = dir.join("home");
std::fs::create_dir_all(home.join(".local/bin")).unwrap();
let path_dir = dir.join("path");
std::fs::create_dir_all(&path_dir).unwrap();
let on_path = path_dir.join(exe_name(BIN_NAME));
write_executable(&on_path, "#!/bin/sh\nexit 0\n");
write_executable(
&home.join(".local/bin").join(exe_name(BIN_NAME)),
"#!/bin/sh\nexit 0\n",
);
assert_eq!(
binary_with(&|_: &str| None, &dirs(&path_dir), Some(&home)),
Some(on_path)
);
}
#[test]
fn home_local_bin_is_the_first_fallback_directory() {
let dir = test_dir("bin-home");
let home = dir.join("home");
let bin = home.join(".local/bin");
std::fs::create_dir_all(&bin).unwrap();
let candidate = bin.join(exe_name(BIN_NAME));
write_executable(&candidate, "#!/bin/sh\nexit 0\n");
assert_eq!(
binary_with(&|_: &str| None, &None, Some(&home)),
Some(candidate)
);
assert!(fallback_dirs(Some(&home))
.iter()
.any(|dir| dir == Path::new("/usr/local/bin")));
}
#[test]
fn state_round_trips_port_and_pid() {
let dir = test_dir("state");
let path = dir.join("rustxng.port");
write_state(&path, 45123, Some(4242)).unwrap();
assert_eq!(read_state(&path), Some((45123, Some(4242))));
write_state(&path, 45124, None).unwrap();
assert_eq!(read_state(&path), Some((45124, None)));
clear_state(&path);
assert_eq!(read_state(&path), None);
}
#[test]
fn malformed_state_is_ignored_or_kept_safe() {
for text in ["", "not-a-port", "0", "70000", "\t", "45123\tnot-a-pid"] {
match parse_state(text) {
None => {}
Some((port, pid)) => {
assert!((1..=65535).contains(&port), "{text:?}");
assert!(pid.is_none(), "{text:?}");
}
}
}
assert_eq!(parse_state("45123"), Some((45123, None)));
assert_eq!(
parse_state("45123\t4242\ttrailing"),
Some((45123, Some(4242)))
);
}
#[test]
fn free_port_is_bindable_and_in_use_reports_a_live_listener() {
let reserved = TcpListener::bind((Ipv4Addr::LOCALHOST, 0)).unwrap();
let port = reserved.local_addr().unwrap().port();
assert_ne!(port, 0);
assert!(port_in_use(port), "the reserved listener owns the port");
drop(reserved);
let fresh = free_port().unwrap();
assert_ne!(fresh, 0);
let listener = TcpListener::bind((Ipv4Addr::LOCALHOST, fresh)).unwrap();
assert!(port_in_use(fresh));
drop(listener);
}
#[test]
fn plain_chunked_and_bad_health_responses_are_understood() {
let plain = parse_response("HTTP/1.1 200 OK\r\nContent-Length: 2\r\n\r\nok").unwrap();
assert_eq!(plain.status, 200);
assert_eq!(plain.body, "ok");
assert!(plain.healthy());
let chunked = parse_response(
"HTTP/1.1 200 OK\r\nTransfer-Encoding: chunked\r\n\r\n2\r\nok\r\n0\r\n\r\n",
)
.unwrap();
assert_eq!(chunked.body, "ok");
assert!(!parse_response("HTTP/1.1 404 Not Found\r\n\r\nnope")
.unwrap()
.healthy());
assert!(parse_response("<html>not http</html>").is_err());
assert!(split_base("http://127.0.0.1:8888/x\r\nInjected: yes").is_err());
assert!(split_base("https://127.0.0.1:8888").is_err());
assert_eq!(
split_base("http://127.0.0.1:8888/searx").unwrap(),
("127.0.0.1".to_string(), 8888, "/searx".to_string())
);
}
fn start_stub() -> (String, u16, KillOnDrop) {
let program = stub_program();
let port = free_port().unwrap();
let child = std::process::Command::new(&program)
.arg(port.to_string())
.stdout(std::process::Stdio::null())
.stderr(std::process::Stdio::null())
.spawn()
.unwrap();
let guard = KillOnDrop(child);
let base = url_for(port);
let deadline = Instant::now() + Duration::from_secs(10);
while Instant::now() < deadline && !health(&base) {
std::thread::sleep(Duration::from_millis(25));
}
(base, port, guard)
}
fn compiler() -> PathBuf {
static RESOLVED: OnceLock<PathBuf> = OnceLock::new();
RESOLVED
.get_or_init(|| {
if let Ok(sysroot) = std::process::Command::new("rustc")
.arg("--print")
.arg("sysroot")
.output()
{
let bin = PathBuf::from(String::from_utf8_lossy(&sysroot.stdout).trim())
.join("bin")
.join(exe_name("rustc"));
if bin.is_file() {
return bin;
}
}
match std::env::var("RUSTC") {
Ok(value) if Path::new(&value).is_absolute() => PathBuf::from(value),
_ => PathBuf::from("rustc"),
}
})
.clone()
}
fn stub_program() -> PathBuf {
static BUILT: OnceLock<PathBuf> = OnceLock::new();
BUILT
.get_or_init(|| build_stub_program(&test_dir("stub-bin")))
.clone()
}
fn build_stub_program(dir: &Path) -> PathBuf {
let source = dir.join("stub.rs");
std::fs::write(&source, STUB_SOURCE).unwrap();
let program = dir.join(exe_name("stub"));
let rustc = compiler();
let output = std::process::Command::new(rustc)
.arg(&source)
.arg("-o")
.arg(&program)
.output()
.expect("rustc must run: cargo test ships one");
assert!(
output.status.success(),
"rustc failed: {}",
String::from_utf8_lossy(&output.stderr)
);
program
}
#[test]
fn health_and_search_answer_from_a_local_tcp_stub() {
let _lock = env_lock();
let (base, port, _guard) = start_stub();
assert!(health(&base), "/healthz must answer");
assert!(port_in_use(port));
let response = http_get(&base, "/search?q=x", Duration::from_secs(5)).unwrap();
assert_eq!(response.status, 200);
assert!(response.body.contains("DUMMY_HIT"));
assert!(response.json().is_none(), "the stub answers HTML, not JSON");
let hits = crate::websearch::fetch_via_stub(&base, "x", crate::locale::Locale::English)
.expect("the stub's /search page must parse");
assert_eq!(hits.len(), 1);
assert_eq!(hits[0].title, "DUMMY_HIT");
assert!(!health(&url_for(free_port().unwrap())));
}
#[test]
fn ensure_reuses_a_healthy_instance() {
let _lock = env_lock();
let (base, port, _guard) = start_stub();
let dir = test_dir("ensure-reuse");
let home = dir.join("home");
std::fs::create_dir_all(&home).unwrap();
let state = home.join("rustxng.port");
write_state(&state, port, Some(999_999)).unwrap();
let _env = EnvGuard::apply(&[
("DSH_HOME", Some(home.to_string_lossy().as_ref())),
("RUSTXNG_BIN", None),
("RUSTXNG_URL", None),
("RUSTXNG_PORT", None),
("RUSTXNG_SETTINGS", None),
]);
let _bin = BinGuard::install(Some(&stub_program()));
let url = ensure().expect("a healthy recorded instance is reused");
assert_eq!(url, base);
assert_eq!(read_state(&state), Some((port, Some(999_999))));
}
#[test]
fn base_url_skips_an_occupied_configured_port() {
let _lock = env_lock();
let dir = test_dir("base-url-port");
let home = dir.join("home");
std::fs::create_dir_all(&home).unwrap();
let occupied = std::net::TcpListener::bind((std::net::Ipv4Addr::LOCALHOST, 0)).unwrap();
let occupied_port = occupied.local_addr().unwrap().port();
let recorded = free_port().unwrap();
write_state(&home.join("rustxng.port"), recorded, Some(4242)).unwrap();
let _env = EnvGuard::apply(&[
("DSH_HOME", Some(home.to_string_lossy().as_ref())),
("RUSTXNG_URL", None),
("RUSTXNG_PORT", Some(occupied_port.to_string().as_str())),
]);
assert_eq!(base_url(), Some(url_for(recorded)));
}
#[test]
fn base_url_is_none_without_a_configured_or_recorded_port() {
let _lock = env_lock();
let dir = test_dir("base-url-none");
let home = dir.join("home");
std::fs::create_dir_all(&home).unwrap();
let _env = EnvGuard::apply(&[
("DSH_HOME", Some(home.to_string_lossy().as_ref())),
("RUSTXNG_URL", None),
("RUSTXNG_PORT", None),
]);
assert_eq!(base_url(), None);
}
#[test]
fn configured_url_is_used_without_spawning() {
let _lock = env_lock();
let (base, _port, _guard) = start_stub();
let dir = test_dir("ensure-url");
let home = dir.join("home");
std::fs::create_dir_all(&home).unwrap();
let dead = free_port().unwrap();
let _env = EnvGuard::apply(&[
("DSH_HOME", Some(home.to_string_lossy().as_ref())),
("RUSTXNG_BIN", Some("/nonexistent/rdsh-test-rustxng")),
("RUSTXNG_URL", Some(base.as_str())),
("RUSTXNG_PORT", None),
]);
assert_eq!(ensure(), Ok(base));
std::env::set_var("RUSTXNG_URL", url_for(dead));
let failure = ensure().unwrap_err();
assert!(matches!(failure, Error::Unresponsive(_)), "{failure:?}");
assert!(!home.join("rustxng.port").exists());
}
#[test]
fn ensure_reports_the_missing_binary_hint() {
let _lock = env_lock();
let dir = test_dir("ensure-missing");
let home = dir.join("home");
std::fs::create_dir_all(&home).unwrap();
let _env = EnvGuard::apply(&[
("DSH_HOME", Some(home.to_string_lossy().as_ref())),
("RUSTXNG_BIN", None),
("RUSTXNG_URL", None),
("RUSTXNG_PORT", None),
("RUSTXNG_SETTINGS", None),
]);
let _bin = BinGuard::install(None);
let failure = ensure().unwrap_err();
assert_eq!(failure, Error::BinNotFound);
let hint = missing_binary_hint(crate::locale::Locale::English);
assert!(hint.contains("RUSTXNG_BIN"), "{hint}");
assert!(hint.contains("PATH"), "{hint}");
assert!(!home.join("rustxng.port").exists());
assert!(!fallback_dirs(Some(&home)).is_empty());
}
#[cfg(unix)]
#[test]
fn ensure_spawns_and_stop_terminates_the_recorded_instance() {
let _lock = env_lock();
let dir = test_dir("ensure-spawn");
let home = dir.join("home");
std::fs::create_dir_all(&home).unwrap();
let program = dir.join(exe_name(BIN_NAME));
std::fs::copy(stub_program(), &program).unwrap();
let _env = EnvGuard::apply(&[
("DSH_HOME", Some(home.to_string_lossy().as_ref())),
("RUSTXNG_BIN", Some(program.to_string_lossy().as_ref())),
("RUSTXNG_URL", None),
("RUSTXNG_PORT", None),
("RUSTXNG_SETTINGS", None),
]);
let url = ensure().expect("the stub must start");
let (port, pid) = read_state(&state_file()).expect("ensure records the port");
assert_eq!(url, url_for(port));
assert!(health(&url));
assert!(log_file().starts_with(&home));
assert!(stop().unwrap(), "stop must terminate the recorded stub");
let deadline = Instant::now() + Duration::from_secs(5);
while Instant::now() < deadline && port_in_use(port) {
std::thread::sleep(Duration::from_millis(25));
}
assert!(!port_in_use(port), "the spawned process must be gone");
assert_eq!(read_state(&state_file()), None);
assert!(pid.is_some());
assert!(!stop().unwrap(), "a second stop has nothing to terminate");
}
#[cfg(unix)]
#[test]
fn process_is_rustxng_rejects_a_foreign_pid() {
assert!(!process_is_rustxng(std::process::id()));
assert!(!process_is_rustxng(999_999));
}
#[test]
fn spawn_arguments_follow_the_documented_cli() {
let _lock = env_lock();
let dir = test_dir("config");
let config = dir.join("settings.yml");
std::fs::write(&config, "search: {}\n").unwrap();
let _env = EnvGuard::apply(&[("RUSTXNG_SETTINGS", None)]);
let mut command = std::process::Command::new(BIN_NAME);
build_command(&mut command, 45123);
let args: Vec<String> = command
.get_args()
.map(|arg| arg.to_string_lossy().to_string())
.collect();
assert_eq!(args, vec!["--bind", "127.0.0.1", "--port", "45123"]);
std::env::set_var("RUSTXNG_SETTINGS", &config);
let mut command = std::process::Command::new(BIN_NAME);
build_command(&mut command, 45124);
let args: Vec<String> = command
.get_args()
.map(|arg| arg.to_string_lossy().to_string())
.collect();
assert_eq!(
args,
vec![
"--config",
config.to_string_lossy().as_ref(),
"--bind",
"127.0.0.1",
"--port",
"45124"
]
);
}
const STUB_SOURCE: &str = r#"
use std::io::{Read, Write};
use std::net::TcpListener;
fn main() {
// The documented rustxng CLI contract: --bind ADDRESS --port PORT. A
// bare positional port is accepted too, so the fixture can also be run
// by hand.
let mut positional = None;
let mut args = std::env::args().skip(1);
while let Some(arg) = args.next() {
if arg == "--port" {
positional = Some(args.next().unwrap());
} else if !arg.starts_with('-') {
positional.get_or_insert(arg);
}
}
let port: u16 = positional.expect("--port is required").parse().unwrap();
let listener = TcpListener::bind(("127.0.0.1", port)).unwrap();
for stream in listener.incoming() {
let Ok(mut stream) = stream else { continue };
let mut request = Vec::new();
let mut byte = [0u8; 1];
while let Ok(1) = stream.read(&mut byte) {
request.push(byte[0]);
if request.ends_with(b"\r\n\r\n") || request.len() > 16384 {
break;
}
}
let request = String::from_utf8_lossy(&request).to_string();
let target = request
.lines()
.next()
.and_then(|line| line.split_whitespace().nth(1))
.unwrap_or("/")
.to_string();
let (status, body) = if target.starts_with("/healthz") {
("200 OK", "{\"status\":\"ok\"}".to_string())
} else if target.starts_with("/search") {
(
"200 OK",
"<html><head><meta name=\"generator\" content=\"searxng/test\"></head><body><div id=\"results\"><article class=\"result\"><a href=\"https://example.invalid/\">DUMMY_HIT</a></article></div></body></html>".to_string(),
)
} else {
("404 Not Found", "no".to_string())
};
let response = format!(
"HTTP/1.0 {status}\r\nContent-Type: text/html; charset=utf-8\r\nContent-Length: {}{}\r\n\r\n{body}",
body.len(),
"\r\n\r\n"
);
let _ = stream.write_all(response.as_bytes());
let _ = stream.flush();
}
}
"#;
}
#[cfg(test)]
mod install_tests {
use super::*;
#[test]
fn parses_latest_release_assets() {
let json = r#"{
"tag_name": "v0.2.0",
"assets": [
{"name": "SOURCE.tar.gz", "browser_download_url": "https://example/SOURCE.tar.gz"},
{"name": "rustxng-v0.2.0-x86_64-unknown-linux-gnu",
"browser_download_url": "https://example/rustxng-linux"},
{"name": "rustxng-v0.2.0-x86_64-pc-windows-msvc.exe",
"browser_download_url": "https://example/rustxng-win.exe"}
]
}"#;
let assets = parse_latest_assets(json);
assert_eq!(assets.len(), 3);
match target_triple() {
"x86_64-unknown-linux-gnu" => {
let picked = choose_asset(&assets, target_triple()).unwrap();
assert_eq!(picked.name, "rustxng-v0.2.0-x86_64-unknown-linux-gnu");
assert_eq!(picked.url, "https://example/rustxng-linux");
}
"x86_64-pc-windows-msvc" => {
let picked = choose_asset(&assets, target_triple()).unwrap();
assert_eq!(picked.name, "rustxng-v0.2.0-x86_64-pc-windows-msvc.exe");
}
other => {
assert!(choose_asset(&assets, other).is_none());
}
}
}
#[test]
fn asset_contract_requires_tag_prefix() {
assert!(matches_triple(
"rustxng-v0.2.0-x86_64-unknown-linux-gnu",
"x86_64-unknown-linux-gnu"
));
assert!(!matches_triple(
"rustxng-x86_64-unknown-linux-gnu",
"x86_64-unknown-linux-gnu"
));
assert!(!matches_triple(
"rustxng-v0.2.0-aarch64-apple-darwin",
"x86_64-unknown-linux-gnu"
));
assert!(parse_latest_assets("not json").is_empty());
}
#[test]
fn windows_assets_use_exe_suffix() {
let name = "rustxng-v0.2.0-x86_64-pc-windows-msvc.exe";
assert!(matches_triple(name, "x86_64-pc-windows-msvc"));
assert!(!looks_like_executable(b"<html>rate limited</html>"));
assert!(looks_like_executable(b"ELFpayload"));
assert!(looks_like_executable(b"MZpayload"));
}
}