use std::collections::BTreeMap;
use std::fs;
use std::io::{BufRead, BufReader, Read, Write};
use std::path::{Path, PathBuf};
use std::process::{Child, Command, Stdio};
use std::sync::atomic::{AtomicBool, AtomicU64, Ordering};
use std::sync::{Arc, Mutex};
use std::thread;
use std::time::{Duration, Instant};
use anyhow::{bail, Context, Result};
use serde::{Deserialize, Serialize};
use sha2::{Digest, Sha256};
use crate::config;
use crate::remote;
const SOCKET_FILE: &str = "runtime.sock";
const PID_FILE: &str = "runtime.pid";
const MAX_WORKERS: usize = 16;
const IDLE_TTL_SECS: u64 = 15 * 60;
const WORKER_READY_TIMEOUT: Duration = Duration::from_secs(10);
const JOB_TIMEOUT: Duration = Duration::from_secs(60 * 30);
static STOP_REQUESTED: AtomicBool = AtomicBool::new(false);
static WARM_HITS: AtomicU64 = AtomicU64::new(0);
static COLD_FALLBACKS: AtomicU64 = AtomicU64::new(0);
static JOBS_OK: AtomicU64 = AtomicU64::new(0);
static JOBS_ERR: AtomicU64 = AtomicU64::new(0);
pub fn runtime_enabled() -> bool {
match std::env::var("JAN_RUNTIME") {
Ok(v) => {
let t = v.trim();
!(t == "0" || t.eq_ignore_ascii_case("false") || t.eq_ignore_ascii_case("off"))
}
Err(_) => true,
}
}
pub fn runtime_dir() -> PathBuf {
if let Ok(dir) = std::env::var("XDG_RUNTIME_DIR") {
let dir = dir.trim();
if !dir.is_empty() {
return PathBuf::from(dir).join("jan-cli");
}
}
config::config_dir().join("run")
}
pub fn socket_path() -> PathBuf {
runtime_dir().join(SOCKET_FILE)
}
pub fn pid_path() -> PathBuf {
runtime_dir().join(PID_FILE)
}
fn worker_sock_path(key_hash: &str) -> PathBuf {
runtime_dir().join(format!("runtime-w-{key_hash}.sock"))
}
fn jan_bin() -> Result<PathBuf> {
let exe = std::env::current_exe().context("resolve jan binary path")?;
exe.canonicalize()
.with_context(|| format!("canonicalize {}", exe.display()))
}
fn short_hash(parts: &[&str]) -> String {
let mut h = Sha256::new();
for p in parts {
h.update(p.as_bytes());
h.update([0]);
}
let dig = h.finalize();
hex::encode(&dig[..8])
}
mod hex {
pub fn encode(bytes: &[u8]) -> String {
const HEX: &[u8] = b"0123456789abcdef";
let mut s = String::with_capacity(bytes.len() * 2);
for &b in bytes {
s.push(HEX[(b >> 4) as usize] as char);
s.push(HEX[(b & 0xf) as usize] as char);
}
s
}
}
fn materialize_script(name: &str, body: &str) -> Result<PathBuf> {
let dir = remote::cache_root()?.join("runtime");
fs::create_dir_all(&dir).with_context(|| format!("create {}", dir.display()))?;
let hash = short_hash(&[body]);
let path = dir.join(format!("{name}-{hash}"));
if !path.is_file() {
fs::write(&path, body).with_context(|| format!("write {}", path.display()))?;
#[cfg(unix)]
{
use std::os::unix::fs::PermissionsExt;
let mut perms = fs::metadata(&path)?.permissions();
perms.set_mode(0o755);
fs::set_permissions(&path, perms)?;
}
}
Ok(path)
}
fn python_worker_script() -> Result<PathBuf> {
materialize_script(
"python_worker.py",
include_str!("../runtime/python_worker.py"),
)
}
fn node_worker_script() -> Result<PathBuf> {
materialize_script("node_worker.js", include_str!("../runtime/node_worker.js"))
}
fn shell_worker_script() -> Result<PathBuf> {
materialize_script("shell_worker.py", include_str!("../runtime/shell_worker.py"))
}
fn kotlin_worker_script() -> Result<PathBuf> {
materialize_script(
"kotlin_worker.py",
include_str!("../runtime/kotlin_worker.py"),
)
}
#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)]
#[serde(rename_all = "lowercase")]
pub enum RuntimeLang {
Python,
Node,
Bash,
Sh,
Zsh,
Kotlin,
}
impl RuntimeLang {
pub fn as_str(&self) -> &'static str {
match self {
Self::Python => "python",
Self::Node => "node",
Self::Bash => "bash",
Self::Sh => "sh",
Self::Zsh => "zsh",
Self::Kotlin => "kotlin",
}
}
}
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct WorkerKey {
pub lang: RuntimeLang,
pub interpreter: String,
pub env_root: String,
#[serde(default)]
pub node_path: Option<String>,
}
impl WorkerKey {
pub fn hash(&self) -> String {
short_hash(&[
self.lang.as_str(),
&self.interpreter,
&self.env_root,
self.node_path.as_deref().unwrap_or(""),
])
}
}
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct JobSource {
pub kind: String,
#[serde(default)]
pub value: String,
#[serde(default)]
pub classpath: Option<String>,
#[serde(default)]
pub main_class: Option<String>,
#[serde(default)]
pub java: Option<String>,
}
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct JobRequest {
pub cwd: String,
pub env: BTreeMap<String, String>,
pub source: JobSource,
#[serde(default)]
pub argv: Vec<String>,
#[serde(default)]
pub shell: Option<String>,
#[serde(default)]
pub argv0: Option<String>,
}
#[derive(Debug, Clone, Serialize, Deserialize)]
struct JobResponse {
ok: bool,
#[serde(default)]
exit_code: Option<i32>,
#[serde(default)]
error: Option<String>,
#[serde(default)]
stdout_b64: Option<String>,
#[serde(default)]
stderr_b64: Option<String>,
}
struct WorkerHandle {
key: WorkerKey,
key_hash: String,
child: Child,
sock: PathBuf,
last_used: Instant,
jobs: u64,
}
struct SupervisorState {
workers: Vec<WorkerHandle>,
started: Instant,
}
impl SupervisorState {
fn status_line(&self) -> String {
let warm = WARM_HITS.load(Ordering::Relaxed);
let cold = COLD_FALLBACKS.load(Ordering::Relaxed);
let ok = JOBS_OK.load(Ordering::Relaxed);
let err = JOBS_ERR.load(Ordering::Relaxed);
let up = self.started.elapsed().as_secs();
let mut parts: Vec<String> = self
.workers
.iter()
.map(|w| {
format!(
"{}:{}:jobs={}",
w.key.lang.as_str(),
&w.key_hash[..8.min(w.key_hash.len())],
w.jobs
)
})
.collect();
parts.sort();
format!(
"ok uptime={up}s workers={} warm_hits={warm} cold_fallbacks={cold} jobs_ok={ok} jobs_err={err} [{}]",
self.workers.len(),
parts.join(" ")
)
}
fn reap_idle(&mut self) {
let ttl = Duration::from_secs(IDLE_TTL_SECS);
let mut keep = Vec::new();
for mut w in self.workers.drain(..) {
if w.last_used.elapsed() > ttl {
let _ = w.child.kill();
let _ = w.child.wait();
let _ = fs::remove_file(&w.sock);
} else {
keep.push(w);
}
}
self.workers = keep;
}
fn stop_all(&mut self) {
for mut w in self.workers.drain(..) {
let _ = w.child.kill();
let _ = w.child.wait();
let _ = fs::remove_file(&w.sock);
}
}
fn find_mut(&mut self, hash: &str) -> Option<&mut WorkerHandle> {
self.workers.iter_mut().find(|w| w.key_hash == hash)
}
fn register_worker(&mut self, key: WorkerKey, hash: String, child: Child, sock: PathBuf) {
self.workers.push(WorkerHandle {
key,
key_hash: hash,
child,
sock,
last_used: Instant::now(),
jobs: 0,
});
}
}
fn spawn_worker_process(key: &WorkerKey, sock: &Path) -> Result<Child> {
let _ = fs::remove_file(sock);
let mut cmd = match key.lang {
RuntimeLang::Python => {
let script = python_worker_script()?;
let mut c = Command::new(&key.interpreter);
c.arg("-u").arg(script).arg(sock);
c
}
RuntimeLang::Node => {
let script = node_worker_script()?;
let mut c = Command::new(&key.interpreter);
c.arg(script).arg(sock);
if let Some(np) = &key.node_path {
c.env("NODE_PATH", np);
}
c
}
RuntimeLang::Bash | RuntimeLang::Sh | RuntimeLang::Zsh => {
let script = shell_worker_script()?;
let py = which_python()?;
let mut c = Command::new(py);
c.arg(script).arg(sock);
c
}
RuntimeLang::Kotlin => {
let script = kotlin_worker_script()?;
let py = which_python()?;
let mut c = Command::new(py);
c.arg(script).arg(sock);
c
}
};
cmd.stdin(Stdio::null())
.stdout(Stdio::piped())
.stderr(Stdio::inherit());
if !key.env_root.is_empty() {
cmd.env("JAN_RUNTIME_ENV_ROOT", &key.env_root);
}
cmd.spawn()
.with_context(|| format!("spawn {} worker", key.lang.as_str()))
}
fn which_python() -> Result<PathBuf> {
for name in ["python3", "python"] {
if let Ok(p) = crate::deps::resolve_program(name, &[]) {
return Ok(p);
}
}
bail!("python3 required to host shell/kotlin runtime workers")
}
fn spawn_worker(key: &WorkerKey, hash: &str) -> Result<PathBuf> {
let sock = worker_sock_path(hash);
let mut child = spawn_worker_process(key, &sock)?;
let stdout = child.stdout.take().context("worker stdout")?;
let mut reader = BufReader::new(stdout);
let mut line = String::new();
let start = Instant::now();
loop {
if start.elapsed() > WORKER_READY_TIMEOUT {
let _ = child.kill();
bail!(
"{} worker failed to become ready within {:?}",
key.lang.as_str(),
WORKER_READY_TIMEOUT
);
}
line.clear();
match reader.read_line(&mut line) {
Ok(0) => {
let _ = child.kill();
bail!("{} worker exited before ready", key.lang.as_str());
}
Ok(_) => {
if line.starts_with("ready ") {
break;
}
}
Err(e) => {
let _ = child.kill();
bail!("{} worker ready read: {e}", key.lang.as_str());
}
}
}
thread::spawn(move || {
let mut r = reader;
let mut sink = Vec::new();
let _ = r.read_to_end(&mut sink);
});
PENDING_CHILD
.lock()
.unwrap()
.replace((key.clone(), hash.to_string(), child, sock.clone()));
Ok(sock)
}
static PENDING_CHILD: Mutex<Option<(WorkerKey, String, Child, PathBuf)>> = Mutex::new(None);
fn ensure_worker_locked(state: &mut SupervisorState, key: WorkerKey) -> Result<PathBuf> {
state.reap_idle();
let hash = key.hash();
if let Some(w) = state.find_mut(&hash) {
match w.child.try_wait() {
Ok(None) => {
w.last_used = Instant::now();
return Ok(w.sock.clone());
}
_ => {
let _ = fs::remove_file(&w.sock);
}
}
state.workers.retain(|w| w.key_hash != hash);
}
if state.workers.len() >= MAX_WORKERS {
if let Some(idx) = state
.workers
.iter()
.enumerate()
.min_by_key(|(_, w)| w.last_used)
.map(|(i, _)| i)
{
let mut old = state.workers.remove(idx);
let _ = old.child.kill();
let _ = old.child.wait();
let _ = fs::remove_file(&old.sock);
}
}
let sock = spawn_worker(&key, &hash)?;
if let Some((k, h, child, s)) = PENDING_CHILD.lock().unwrap().take() {
state.register_worker(k, h, child, s);
}
Ok(sock)
}
fn write_pid_file() -> Result<()> {
let dir = runtime_dir();
fs::create_dir_all(&dir).with_context(|| format!("create {}", dir.display()))?;
fs::write(pid_path(), format!("{}\n", std::process::id()))
.with_context(|| format!("write {}", pid_path().display()))?;
Ok(())
}
fn remove_runtime_files() {
let _ = fs::remove_file(socket_path());
let _ = fs::remove_file(pid_path());
}
#[cfg(unix)]
fn accept_control(state: Arc<Mutex<SupervisorState>>) {
use std::os::unix::net::UnixListener;
let listener = match UnixListener::bind(socket_path()) {
Ok(l) => l,
Err(e) => {
eprintln!(
"jan runtime daemon: bind {}: {e:#}",
socket_path().display()
);
STOP_REQUESTED.store(true, Ordering::SeqCst);
return;
}
};
let _ = listener.set_nonblocking(false);
for stream in listener.incoming() {
if STOP_REQUESTED.load(Ordering::SeqCst) {
break;
}
match stream {
Ok(s) => {
let st = Arc::clone(&state);
thread::spawn(move || handle_client(s, st));
}
Err(e) => {
eprintln!("jan runtime daemon: accept: {e:#}");
}
}
}
}
#[cfg(unix)]
fn handle_client(mut stream: std::os::unix::net::UnixStream, state: Arc<Mutex<SupervisorState>>) {
use std::os::unix::net::UnixStream;
let reader = BufReader::new(
stream
.try_clone()
.unwrap_or_else(|_| stream.try_clone().expect("clone runtime control socket")),
);
let line = match reader.lines().next() {
Some(Ok(l)) => l,
_ => return,
};
let line = line.trim();
let (cmd, rest) = match line.split_once(char::is_whitespace) {
Some((c, r)) => (c, r.trim()),
None => (line, ""),
};
let reply = match cmd.to_ascii_lowercase().as_str() {
"ping" => "pong".to_string(),
"status" => state.lock().unwrap().status_line(),
"stop" => {
STOP_REQUESTED.store(true, Ordering::SeqCst);
state.lock().unwrap().stop_all();
"ok stopping".to_string()
}
"ensure" => match serde_json::from_str::<WorkerKey>(rest) {
Ok(key) => match ensure_worker_locked(&mut state.lock().unwrap(), key) {
Ok(sock) => format!("ok sock={}", sock.display()),
Err(e) => format!("error {e:#}"),
},
Err(e) => format!("error bad key json: {e}"),
},
"note_cold" => {
COLD_FALLBACKS.fetch_add(1, Ordering::Relaxed);
"ok".to_string()
}
"note_warm" => {
WARM_HITS.fetch_add(1, Ordering::Relaxed);
JOBS_OK.fetch_add(1, Ordering::Relaxed);
if !rest.is_empty() {
if let Some(w) = state.lock().unwrap().find_mut(rest) {
w.jobs += 1;
w.last_used = Instant::now();
}
}
"ok".to_string()
}
other => format!("error unknown command `{other}`"),
};
let _ = writeln!(stream, "{reply}");
let _ = stream.flush();
if cmd.eq_ignore_ascii_case("stop") {
let path = socket_path();
thread::spawn(move || {
thread::sleep(Duration::from_millis(50));
let _ = UnixStream::connect(path);
});
}
}
pub fn run_foreground() -> Result<i32> {
#[cfg(not(unix))]
{
bail!("jan runtime daemon requires Unix");
}
#[cfg(unix)]
{
STOP_REQUESTED.store(false, Ordering::SeqCst);
fs::create_dir_all(runtime_dir())
.with_context(|| format!("create {}", runtime_dir().display()))?;
remove_runtime_files();
write_pid_file()?;
let state = Arc::new(Mutex::new(SupervisorState {
workers: Vec::new(),
started: Instant::now(),
}));
let state_bg = Arc::clone(&state);
let control = thread::spawn(move || accept_control(state_bg));
while !STOP_REQUESTED.load(Ordering::SeqCst) {
thread::sleep(Duration::from_millis(200));
if let Ok(mut st) = state.try_lock() {
st.reap_idle();
}
}
state.lock().unwrap().stop_all();
remove_runtime_files();
let _ = control.join();
Ok(0)
}
}
pub fn send_command(cmd: &str) -> Result<String> {
#[cfg(not(unix))]
{
let _ = cmd;
bail!("jan runtime requires Unix");
}
#[cfg(unix)]
{
use std::os::unix::net::UnixStream;
let path = socket_path();
if !path.exists() {
bail!(
"jan runtime daemon is not running (no socket at {})",
path.display()
);
}
let mut stream = UnixStream::connect(&path)
.with_context(|| format!("connect to {}", path.display()))?;
stream.set_read_timeout(Some(Duration::from_secs(5)))?;
stream.set_write_timeout(Some(Duration::from_secs(5)))?;
writeln!(stream, "{cmd}").context("write runtime command")?;
let mut reader = BufReader::new(stream);
let mut reply = String::new();
reader.read_line(&mut reply).context("read runtime reply")?;
Ok(reply.trim().to_string())
}
}
pub fn daemon_running() -> bool {
match send_command("ping") {
Ok(r) => r == "pong",
Err(_) => false,
}
}
pub fn start_daemon_background() -> Result<()> {
if !runtime_enabled() {
return Ok(());
}
#[cfg(not(unix))]
{
return Ok(());
}
#[cfg(unix)]
{
if daemon_running() {
return Ok(());
}
let jan = jan_bin()?;
let mut cmd = Command::new(&jan);
cmd.args(["--no-log", "runtime", "daemon"]);
cmd.stdin(Stdio::null())
.stdout(Stdio::null())
.stderr(Stdio::null());
cmd.spawn()
.with_context(|| format!("spawn `{} runtime daemon`", jan.display()))?;
for _ in 0..50 {
if daemon_running() {
return Ok(());
}
thread::sleep(Duration::from_millis(100));
}
bail!(
"jan runtime daemon failed to start (no response on {})",
socket_path().display()
);
}
}
pub fn stop_daemon() -> Result<()> {
if !socket_path().exists() {
println!("(jan runtime daemon is not running)");
return Ok(());
}
let reply = send_command("stop")?;
for _ in 0..30 {
if !socket_path().exists() {
println!("stopped jan runtime daemon");
return Ok(());
}
thread::sleep(Duration::from_millis(100));
}
bail!("jan runtime daemon did not stop: {reply}");
}
pub fn daemon_status() -> Result<i32> {
match send_command("status") {
Ok(reply) if reply.starts_with("ok ") => {
println!("jan runtime daemon running ({reply})");
Ok(0)
}
Ok(reply) => {
println!("jan runtime daemon: {reply}");
Ok(1)
}
Err(e) => {
println!("jan runtime daemon is not running ({e:#})");
Ok(1)
}
}
}
pub fn ensure_worker(key: &WorkerKey) -> Result<PathBuf> {
start_daemon_background()?;
let json = serde_json::to_string(key).context("serialize worker key")?;
let reply = send_command(&format!("ensure {json}"))?;
if let Some(sock) = reply.strip_prefix("ok sock=") {
return Ok(PathBuf::from(sock));
}
if let Some(err) = reply.strip_prefix("error ") {
bail!("{err}");
}
bail!("unexpected ensure reply: {reply}");
}
fn b64_decode(s: &str) -> Result<Vec<u8>> {
fn val(c: u8) -> Option<u8> {
match c {
b'A'..=b'Z' => Some(c - b'A'),
b'a'..=b'z' => Some(c - b'a' + 26),
b'0'..=b'9' => Some(c - b'0' + 52),
b'+' => Some(62),
b'/' => Some(63),
_ => None,
}
}
let bytes = s.as_bytes();
let mut out = Vec::with_capacity(bytes.len() * 3 / 4);
let mut buf = 0u32;
let mut n = 0;
for &c in bytes {
if c == b'=' || c.is_ascii_whitespace() {
continue;
}
let Some(v) = val(c) else {
bail!("invalid base64");
};
buf = (buf << 6) | u32::from(v);
n += 6;
if n >= 8 {
n -= 8;
out.push((buf >> n) as u8);
}
}
Ok(out)
}
pub fn run_job_on_worker(sock: &Path, job: &JobRequest) -> Result<i32> {
#[cfg(not(unix))]
{
let _ = (sock, job);
bail!("warm runtime requires Unix");
}
#[cfg(unix)]
{
use std::os::unix::net::UnixStream;
let mut stream = UnixStream::connect(sock)
.with_context(|| format!("connect to worker {}", sock.display()))?;
stream.set_read_timeout(Some(JOB_TIMEOUT))?;
stream.set_write_timeout(Some(Duration::from_secs(30)))?;
let payload = serde_json::to_string(job).context("serialize job")?;
writeln!(stream, "{payload}").context("write job")?;
let mut reader = BufReader::new(stream);
let mut reply = String::new();
reader.read_line(&mut reply).context("read job reply")?;
let resp: JobResponse =
serde_json::from_str(reply.trim()).with_context(|| format!("parse job reply: {reply}"))?;
if !resp.ok {
JOBS_ERR.fetch_add(1, Ordering::Relaxed);
bail!(
"worker error: {}",
resp.error.unwrap_or_else(|| "unknown".into())
);
}
if let Some(b64) = &resp.stdout_b64 {
let bytes = b64_decode(b64)?;
let _ = std::io::stdout().write_all(&bytes);
let _ = std::io::stdout().flush();
}
if let Some(b64) = &resp.stderr_b64 {
let bytes = b64_decode(b64)?;
let _ = std::io::stderr().write_all(&bytes);
let _ = std::io::stderr().flush();
}
let code = resp.exit_code.unwrap_or(255);
let key_hint = sock
.file_name()
.and_then(|s| s.to_str())
.and_then(|s| s.strip_prefix("runtime-w-"))
.and_then(|s| s.strip_suffix(".sock"))
.unwrap_or("");
let _ = send_command(&format!("note_warm {key_hint}"));
Ok(code)
}
}
pub fn try_run_warm(key: &WorkerKey, job: &JobRequest) -> Result<Option<i32>> {
if !runtime_enabled() {
return Ok(None);
}
#[cfg(not(unix))]
{
let _ = (key, job);
return Ok(None);
}
#[cfg(unix)]
{
match (|| -> Result<i32> {
let sock = ensure_worker(key)?;
run_job_on_worker(&sock, job)
})() {
Ok(code) => Ok(Some(code)),
Err(e) => {
let _ = send_command("note_cold");
if std::env::var_os("JAN_RUNTIME_DEBUG").is_some() {
eprintln!("jan: warm runtime fallback: {e:#}");
}
Ok(None)
}
}
}
}
pub fn dispatch_runtime(args: &[std::ffi::OsString]) -> Result<i32> {
let mut it = args.iter();
let sub = it
.next()
.map(|s| s.to_string_lossy().into_owned())
.unwrap_or_else(|| "status".into());
match sub.as_str() {
"help" | "--help" | "-h" => {
print_runtime_help();
Ok(0)
}
"daemon" => run_foreground(),
"start" => {
start_daemon_background()?;
println!("started jan runtime daemon");
Ok(0)
}
"stop" => {
stop_daemon()?;
Ok(0)
}
"restart" => {
let _ = stop_daemon();
start_daemon_background()?;
println!("restarted jan runtime daemon");
Ok(0)
}
"status" => daemon_status(),
other => {
bail!("unknown runtime subcommand `{other}` (try `jan runtime --help`)");
}
}
}
fn print_runtime_help() {
print!(
"\
jan runtime — warm language interpreter pool (Jan-internal)
Usage:
jan runtime start
jan runtime stop
jan runtime restart
jan runtime status
jan runtime daemon Run supervisor in the foreground
The pool hosts long-lived Python/Node (and shell/Kotlin) workers. Each job runs
in an isolated child. Set JAN_RUNTIME=0 to force cold interpreter spawns.
"
);
}
#[cfg(test)]
mod tests {
use super::*;
#[test]
fn worker_key_hash_stable() {
let k = WorkerKey {
lang: RuntimeLang::Python,
interpreter: "/usr/bin/python3".into(),
env_root: "/tmp/env".into(),
node_path: None,
};
assert_eq!(k.hash(), k.hash());
assert_eq!(k.hash().len(), 16);
}
#[test]
fn runtime_enabled_default() {
let _ = runtime_enabled();
}
}