use std::ffi::OsStr;
use std::io::{Read as _, Write as _};
use std::net::TcpStream;
use std::path::{Path, PathBuf};
use std::process::{Child, Command, Stdio};
use std::time::{Duration, Instant};
use base64::Engine as _;
use crate::measure::{self, Ran, Usage};
pub const OWNER: &str = "did:web:owner.bench.test";
const MASTER_KEY_ENV: &str = "KNOT_STATIC_BENCH_KEY";
const MASTER_KEY: [u8; 32] = [7; 32];
pub struct Knot {
pub root: PathBuf,
pub output: PathBuf,
config: PathBuf,
binary: PathBuf,
nginx: Option<PathBuf>,
port: u16,
server: std::cell::RefCell<Option<Child>>,
}
pub struct Hosted {
pub did: String,
pub path: PathBuf,
}
impl Knot {
pub fn new(root: &Path, binary: &Path, nginx: Option<&Path>) -> Self {
std::fs::create_dir_all(root).unwrap();
let port = std::net::TcpListener::bind("127.0.0.1:0")
.unwrap()
.local_addr()
.unwrap()
.port();
let config = root.join("config.toml");
// with crawl urls empty an activation can't reach a real indexer.
std::fs::write(
&config,
format!(
r#"[knot]
hostname = "knot.bench.test"
owner = "{OWNER}"
[repo]
scan_path = "{}"
default_branch = "main"
object_format = "sha1"
[secrets]
sealed_key_file = "{}"
master_key_env = "{MASTER_KEY_ENV}"
[atproto]
plc_directory = "https://127.0.0.1:9"
[output]
root = "{}"
[activate]
nginx_binary = "{}"
listen = "127.0.0.1:{port}"
[crawl]
urls = []
"#,
root.join("state").display(),
root.join("keys.sealed").display(),
root.join("out").display(),
nginx.map_or("true".into(), |nginx| nginx.display().to_string()),
),
)
.unwrap();
let knot = Self {
output: root.join("out"),
root: root.to_path_buf(),
config,
binary: binary.to_path_buf(),
nginx: nginx.map(Path::to_path_buf),
port,
server: Default::default(),
};
knot.run(["init"], None).ok("init");
knot
}
fn command(&self, args: impl IntoIterator<Item = impl AsRef<OsStr>>) -> Command {
let mut command = Command::new(&self.binary);
command.arg("--config").arg(&self.config).args(args);
self.environment(&mut command);
command
}
fn environment(&self, command: &mut Command) {
command
.env(
MASTER_KEY_ENV,
base64::engine::general_purpose::STANDARD.encode(MASTER_KEY),
)
.env("GIT_CONFIG_GLOBAL", "/dev/null")
.env("GIT_CONFIG_SYSTEM", "/dev/null")
.env("NO_COLOR", "1")
.env("RUST_LOG", "info,knot_static::build=debug")
.env("KNOT_PLAN_TRACE", "1");
}
pub fn run(
&self,
args: impl IntoIterator<Item = impl AsRef<OsStr>>,
stdin: Option<&[u8]>,
) -> Ran {
measure::run(&mut self.command(args), stdin)
}
/// the plc genesis can't be submitted from here, and the command fails after registering.
pub fn import(&self, rkey: &str, source: &Path) -> (Ran, Hosted) {
let ran = self.run(
[
OsStr::new("repo"),
"import".as_ref(),
"--owner".as_ref(),
OWNER.as_ref(),
"--rkey".as_ref(),
rkey.as_ref(),
"--name".as_ref(),
rkey.as_ref(),
"--default-branch".as_ref(),
"main".as_ref(),
source.as_os_str(),
],
None,
);
let hosted = self
.hosted(rkey)
.unwrap_or_else(|| panic!("import of {rkey} registered nothing: {}", ran.stderr()));
(ran, hosted)
}
pub fn create(&self, rkey: &str) -> Hosted {
let ran = self.run(
[
"repo",
"create",
"--owner",
OWNER,
"--rkey",
rkey,
"--name",
rkey,
"--default-branch",
"main",
],
None,
);
self.hosted(rkey)
.unwrap_or_else(|| panic!("create of {rkey} registered nothing: {}", ran.stderr()))
}
pub fn hosted(&self, rkey: &str) -> Option<Hosted> {
let listed = self.run(["repo", "list"], None).ok("repo list").stdout();
listed.lines().find_map(|line| {
let row: Vec<&str> = line.split_whitespace().collect();
row.contains(&rkey).then(|| Hosted {
did: row[0].to_string(),
path: PathBuf::from(row[row.len() - 1]),
})
})
}
pub fn generate(&self, repo: Option<&str>) -> (Ran, Option<String>) {
let mut args = vec!["generate"];
args.extend(repo.map(|did| ["--repo", did]).into_iter().flatten());
let ran = self.run(args, None);
let generation = ran
.stderr()
.lines()
.filter(|line| line.starts_with("generated ") || line.starts_with("reused "))
.find_map(|line| {
line.rsplit_once(" at ")
.map(|(_, dir)| dir.trim().to_string())
})
.and_then(|dir| Some(Path::new(&dir).file_name()?.to_string_lossy().into_owned()));
(ran, generation)
}
pub fn maintenance(&self) -> Ran {
self.run(["maintenance", "--all", "--command", "repack"], None)
.ok("maintenance")
}
pub fn activate(&self, generation: &str) -> Ran {
self.run(["activate", generation], None).ok("activate")
}
pub fn write(&self, method: &str, repo: &str, body: &serde_json::Value) -> Ran {
measure::run_until_answered(
&mut self.command(["write", method, "--repo", repo]),
Some(body.to_string().as_bytes()),
)
.ok(method)
}
/// the bench binary is git's ssh, and it counts the usage itself.
pub fn push(&self, source: &Path, rkey: &str) -> (Ran, Usage) {
let usage_file = self.root.join("ssh-usage.json");
let mut command = crate::common::git_command(source);
command
.args(["push", "-q", &format!("git@knot.bench.test:{rkey}"), "main"])
.env("GIT_SSH", std::env::current_exe().unwrap())
.env("GIT_SSH_VARIANT", "simple")
.env(measure::SSH_USAGE_ENV, &usage_file)
.env("KS_BENCH_BIN", &self.binary)
.env("KS_BENCH_CONFIG", &self.config);
self.environment(&mut command);
let ran = measure::run(&mut command, None).ok("git push");
let usage = measure::read_usage(&usage_file).expect("the ssh stand-in's usage");
(ran, usage)
}
pub fn active_generation(&self) -> Option<String> {
let active: serde_json::Value =
serde_json::from_slice(&std::fs::read(self.output.join("active.json")).ok()?).ok()?;
active["generation"].as_str().map(str::to_string)
}
pub fn activation_record(&self) -> serde_json::Value {
std::fs::read(self.output.join("active.json"))
.ok()
.and_then(|bytes| serde_json::from_slice(&bytes).ok())
.unwrap_or_default()
}
pub fn generation_dir(&self, generation: &str) -> PathBuf {
self.output.join("generations").join(generation)
}
fn active_conf(&self) -> PathBuf {
self.output.join("active.conf")
}
fn nginx_args<'a>(&'a self, command: &'a mut Command) -> &'a mut Command {
command
.arg("-c")
.arg(self.active_conf())
.arg("-e")
.arg(self.output.join("runtime/nginx-error.log"))
}
pub fn nginx_test(&self) -> Option<Ran> {
let nginx = self.nginx.as_ref()?;
Some(measure::run(self.nginx_args(&mut Command::new(nginx)).arg("-t"), None).ok("nginx -t"))
}
pub fn serve(&self) -> bool {
let Some(nginx) = &self.nginx else {
return false;
};
let mut command = Command::new(nginx);
self.nginx_args(&mut command)
.args(["-g", "daemon off;"])
.stdin(Stdio::null())
.stdout(Stdio::null())
.stderr(Stdio::null());
*self.server.borrow_mut() = Some(command.spawn().unwrap());
let deadline = Instant::now() + Duration::from_secs(30);
while TcpStream::connect(("127.0.0.1", self.port)).is_err() {
assert!(Instant::now() < deadline, "nginx never listened");
std::thread::sleep(Duration::from_millis(10));
}
true
}
pub fn reload(&self, repo: &str) -> Option<f64> {
let pid = self.server.borrow().as_ref()?.id();
let token = crate::common::repo_token(repo);
let latest = self
.generation_dir(&self.active_generation()?)
.join("www/_atproto")
.join(token)
.join("latest.json");
let latest: serde_json::Value =
serde_json::from_slice(&std::fs::read(latest).ok()?).ok()?;
let rev = latest["rev"].as_str()?.to_string();
let started = Instant::now();
// SAFETY: plain signal to the nginx master this knot spawned.
unsafe { libc::kill(pid as libc::pid_t, libc::SIGHUP) };
let deadline = started + Duration::from_secs(120);
while Instant::now() < deadline {
if self
.latest_commit(repo)
.is_some_and(|body| body.contains(&rev))
{
return Some(started.elapsed().as_secs_f64());
}
std::thread::sleep(Duration::from_millis(2));
}
None
}
pub fn current_cid(&self, repo: &str, collection: &str, rkey: &str) -> Option<String> {
let generation = self.generation_dir(&self.active_generation()?);
let maps = std::fs::read_to_string(generation.join("nginx/maps/62_records.conf")).ok()?;
let token = crate::common::repo_token(repo);
let key = format!("{token}|{collection}|{rkey}| ");
let file = maps.lines().find_map(|line| {
let (_, file) = line.trim().split_once(&key)?;
file.strip_suffix(';')
})?;
let record = generation.join("www/_atproto").join(token).join(file);
let record: serde_json::Value =
serde_json::from_slice(&std::fs::read(record).ok()?).ok()?;
record["cid"].as_str().map(str::to_string)
}
fn latest_commit(&self, repo: &str) -> Option<String> {
let mut stream = TcpStream::connect(("127.0.0.1", self.port)).ok()?;
stream.set_read_timeout(Some(Duration::from_secs(5))).ok()?;
write!(
stream,
"GET /xrpc/com.atproto.sync.getLatestCommit?did={repo} HTTP/1.1\r\nHost: knot.bench.test\r\nConnection: close\r\n\r\n"
)
.ok()?;
let mut answer = String::new();
stream.read_to_string(&mut answer).ok()?;
Some(answer)
}
}
impl Drop for Knot {
fn drop(&mut self) {
if let Some(mut server) = self.server.get_mut().take() {
// SAFETY: plain signal to the nginx master this knot spawned.
unsafe { libc::kill(server.id() as libc::pid_t, libc::SIGQUIT) };
let _ = server.wait();
}
}
}