use std::fs::{self, OpenOptions};
use std::net::SocketAddr;
use std::path::{Path, PathBuf};
use std::process::{Child, Command, Stdio};
use std::time::Duration;
use anyhow::{bail, Context};
use rand::{rngs::OsRng, RngCore};
use serde::{Deserialize, Serialize};
use sha2::{Digest, Sha256};
use tokio::time::{sleep, Instant};
use super::options::ServeOptions;
const READY_FILE_ENV: &str = "A3S_INTERNAL_WEB_READY_FILE";
pub(super) const INSTANCE_NONCE_ENV: &str = "A3S_INTERNAL_WEB_INSTANCE_NONCE";
pub(super) const INSTANCE_FILE_ENV: &str = "A3S_INTERNAL_WEB_INSTANCE_FILE";
const STARTUP_TIMEOUT: Duration = Duration::from_secs(15);
const POLL_INTERVAL: Duration = Duration::from_millis(50);
#[derive(Clone, Debug, Serialize, Deserialize)]
#[serde(rename_all = "camelCase")]
pub(crate) struct WebInstanceRecord {
pub schema_version: u32,
pub pid: u32,
pub nonce: String,
pub address: SocketAddr,
pub workspace: PathBuf,
pub log_path: PathBuf,
pub executable: PathBuf,
pub started_at_ms: u128,
pub api_only: bool,
}
#[derive(Clone, Debug, Serialize)]
#[serde(rename_all = "camelCase")]
pub(crate) struct WebInstanceStatus {
pub running: bool,
pub stale: bool,
pub instance: Option<WebInstanceRecord>,
}
pub(super) async fn start(
args: &[String],
options: &ServeOptions,
) -> anyhow::Result<WebInstanceRecord> {
let executable = std::env::current_exe()
.map_err(|error| anyhow::anyhow!("could not locate the current a3s executable: {error}"))?;
let workspace = canonical_workspace(&options.workspace)?;
let instance_path = instance_path(&workspace)?;
if let Some(existing) = read_instance(&instance_path)? {
if probe_instance(&existing).await {
bail!(
"A3S Web is already running for {} at http://{}",
workspace.display(),
existing.address
);
}
quarantine_stale_instance(&instance_path)?;
}
let log_path = log_path(&workspace)?;
if let Some(parent) = log_path.parent() {
std::fs::create_dir_all(parent).map_err(|error| {
anyhow::anyhow!(
"could not create A3S Web log directory {}: {error}",
parent.display()
)
})?;
}
let log = OpenOptions::new()
.create(true)
.append(true)
.open(&log_path)
.map_err(|error| {
anyhow::anyhow!("could not open A3S Web log {}: {error}", log_path.display())
})?;
let ready_path = ready_file_path();
let _ = std::fs::remove_file(&ready_path);
let nonce = random_nonce();
let mut command = Command::new(executable);
command
.arg("web")
.args(foreground_args(args))
.env(READY_FILE_ENV, &ready_path)
.env(INSTANCE_NONCE_ENV, &nonce)
.env(INSTANCE_FILE_ENV, &instance_path)
.stdin(Stdio::null())
.stdout(Stdio::from(log.try_clone()?))
.stderr(Stdio::from(log));
configure_detached(&mut command);
let mut child = command
.spawn()
.map_err(|error| anyhow::anyhow!("failed to start A3S Web in the background: {error}"))?;
let pid = child.id();
let ready = wait_until_ready(&mut child, &ready_path, &log_path).await;
let _ = std::fs::remove_file(&ready_path);
let address = ready?;
let record = WebInstanceRecord {
schema_version: 1,
pid,
nonce,
address,
workspace,
log_path: log_path.clone(),
executable: std::env::current_exe()?,
started_at_ms: unix_millis(),
api_only: options.api_only,
};
wait_until_control_ready(&record, &mut child).await?;
write_instance(&instance_path, &record)?;
Ok(record)
}
pub(crate) async fn status(workspace: &Path) -> anyhow::Result<WebInstanceStatus> {
let workspace = canonical_workspace(workspace)?;
let path = instance_path(&workspace)?;
let Some(instance) = read_instance(&path)? else {
return Ok(WebInstanceStatus {
running: false,
stale: false,
instance: None,
});
};
let running = probe_instance(&instance).await;
Ok(WebInstanceStatus {
running,
stale: !running,
instance: Some(instance),
})
}
pub(crate) async fn stop(workspace: &Path) -> anyhow::Result<Option<WebInstanceRecord>> {
let workspace = canonical_workspace(workspace)?;
let path = instance_path(&workspace)?;
let Some(instance) = read_instance(&path)? else {
return Ok(None);
};
if !probe_instance(&instance).await {
quarantine_stale_instance(&path)?;
bail!(
"the A3S Web instance record for {} is stale; it was quarantined without signaling any process",
workspace.display()
);
}
let url = control_url(&instance, "stop");
let response = control_client()?
.post(url)
.send()
.await
.context("failed to request A3S Web shutdown")?;
if !response.status().is_success() {
bail!("A3S Web rejected the authenticated shutdown request");
}
let deadline = Instant::now() + Duration::from_secs(10);
while Instant::now() < deadline {
if !probe_instance(&instance).await {
remove_instance_if_owned(&path, &instance.nonce);
return Ok(Some(instance));
}
sleep(POLL_INTERVAL).await;
}
bail!("A3S Web did not stop within 10 seconds; no force signal was sent")
}
pub(crate) async fn open(workspace: &Path) -> anyhow::Result<WebInstanceRecord> {
let status = status(workspace).await?;
match (status.running, status.instance) {
(true, Some(instance)) => Ok(instance),
_ => bail!("A3S Web is not running for {}", workspace.display()),
}
}
pub(crate) fn read_log_tail(path: &Path, lines: usize) -> anyhow::Result<String> {
let content = fs::read_to_string(path)
.with_context(|| format!("could not read A3S Web log {}", path.display()))?;
let lines = content.lines().rev().take(lines).collect::<Vec<_>>();
let mut output = lines.into_iter().rev().collect::<Vec<_>>().join("\n");
if !output.is_empty() {
output.push('\n');
}
Ok(output)
}
pub(crate) fn remove_instance_if_owned(path: &Path, nonce: &str) {
let owned = read_instance(path)
.ok()
.flatten()
.is_some_and(|record| record.nonce == nonce);
if owned {
let _ = fs::remove_file(path);
}
}
pub(super) fn notify_ready(address: SocketAddr) -> anyhow::Result<()> {
let Some(path) = std::env::var_os(READY_FILE_ENV) else {
return Ok(());
};
std::fs::write(&path, address.to_string()).map_err(|error| {
anyhow::anyhow!(
"could not report A3S Web background startup through {}: {error}",
PathBuf::from(path).display()
)
})
}
async fn wait_until_ready(
child: &mut Child,
ready_path: &Path,
log_path: &Path,
) -> anyhow::Result<SocketAddr> {
let deadline = Instant::now() + STARTUP_TIMEOUT;
loop {
if let Ok(value) = std::fs::read_to_string(ready_path) {
if let Ok(address) = value.trim().parse::<SocketAddr>() {
return Ok(address);
}
}
if let Some(status) = child.try_wait()? {
anyhow::bail!(
"A3S Web exited before becoming ready ({status}); see {}",
log_path.display()
);
}
if Instant::now() >= deadline {
let _ = child.kill();
let _ = child.wait();
anyhow::bail!(
"A3S Web did not become ready within {} seconds; see {}",
STARTUP_TIMEOUT.as_secs(),
log_path.display()
);
}
sleep(POLL_INTERVAL).await;
}
}
fn foreground_args(args: &[String]) -> impl Iterator<Item = &String> {
args.iter()
.filter(|argument| !matches!(argument.as_str(), "-d" | "--detach"))
}
fn log_path(workspace: &Path) -> anyhow::Result<PathBuf> {
Ok(state_root()?
.join("logs/web")
.join(format!("{}.log", workspace_key(workspace))))
}
fn ready_file_path() -> PathBuf {
std::env::temp_dir().join(format!(
"a3s-web-ready-{}-{:016x}",
std::process::id(),
rand::random::<u64>()
))
}
fn instance_path(workspace: &Path) -> anyhow::Result<PathBuf> {
Ok(state_root()?
.join("web/instances")
.join(format!("{}.json", workspace_key(workspace))))
}
fn state_root() -> anyhow::Result<PathBuf> {
if let Some(path) = std::env::var_os("A3S_STATE_HOME").filter(|value| !value.is_empty()) {
return Ok(absolute(PathBuf::from(path)));
}
if let Some(path) = std::env::var_os("XDG_STATE_HOME").filter(|value| !value.is_empty()) {
return Ok(absolute(PathBuf::from(path)).join("a3s"));
}
if let Some(home) = std::env::var_os("HOME") {
return Ok(PathBuf::from(home).join(".local/state/a3s"));
}
#[cfg(windows)]
if let Some(path) = std::env::var_os("LOCALAPPDATA") {
return Ok(PathBuf::from(path).join("a3s/state"));
}
bail!("A3S_STATE_HOME is not set and no home directory is available")
}
fn absolute(path: PathBuf) -> PathBuf {
if path.is_absolute() {
path
} else {
std::env::current_dir()
.map(|directory| directory.join(&path))
.unwrap_or(path)
}
}
fn canonical_workspace(workspace: &Path) -> anyhow::Result<PathBuf> {
fs::canonicalize(workspace)
.with_context(|| format!("could not resolve workspace {}", workspace.display()))
}
fn workspace_key(workspace: &Path) -> String {
let digest = Sha256::digest(workspace.to_string_lossy().as_bytes());
format!("{digest:x}")
}
fn random_nonce() -> String {
let mut bytes = [0u8; 32];
OsRng.fill_bytes(&mut bytes);
bytes.iter().map(|byte| format!("{byte:02x}")).collect()
}
fn unix_millis() -> u128 {
std::time::SystemTime::now()
.duration_since(std::time::UNIX_EPOCH)
.unwrap_or_default()
.as_millis()
}
fn read_instance(path: &Path) -> anyhow::Result<Option<WebInstanceRecord>> {
let bytes = match fs::read(path) {
Ok(bytes) => bytes,
Err(error) if error.kind() == std::io::ErrorKind::NotFound => return Ok(None),
Err(error) => {
return Err(error).with_context(|| format!("could not read {}", path.display()))
}
};
let record = serde_json::from_slice(&bytes)
.with_context(|| format!("invalid A3S Web instance record {}", path.display()))?;
Ok(Some(record))
}
fn write_instance(path: &Path, record: &WebInstanceRecord) -> anyhow::Result<()> {
let parent = path.parent().unwrap_or_else(|| Path::new("."));
fs::create_dir_all(parent)?;
let temp = path.with_extension(format!("{}.tmp", std::process::id()));
let bytes = serde_json::to_vec_pretty(record)?;
fs::write(&temp, bytes)?;
#[cfg(unix)]
{
use std::os::unix::fs::PermissionsExt;
fs::set_permissions(&temp, fs::Permissions::from_mode(0o600))?;
}
fs::rename(&temp, path)?;
Ok(())
}
fn quarantine_stale_instance(path: &Path) -> anyhow::Result<()> {
if !path.exists() {
return Ok(());
}
let stale = path.with_extension(format!("stale-{}", unix_millis()));
fs::rename(path, &stale).with_context(|| {
format!(
"could not quarantine stale A3S Web state {}",
path.display()
)
})
}
fn control_url(instance: &WebInstanceRecord, action: &str) -> String {
format!(
"http://{}/.a3s/web/{}/{}",
instance.address, instance.nonce, action
)
}
fn control_client() -> anyhow::Result<reqwest::Client> {
reqwest::Client::builder()
.timeout(Duration::from_secs(2))
.no_proxy()
.build()
.context("could not create the A3S Web control client")
}
async fn probe_instance(instance: &WebInstanceRecord) -> bool {
let Ok(client) = control_client() else {
return false;
};
let Ok(response) = client.get(control_url(instance, "status")).send().await else {
return false;
};
if !response.status().is_success() {
return false;
}
response
.json::<serde_json::Value>()
.await
.ok()
.is_some_and(|value| {
value.get("pid").and_then(serde_json::Value::as_u64) == Some(instance.pid as u64)
&& value.get("nonce").and_then(serde_json::Value::as_str)
== Some(instance.nonce.as_str())
})
}
async fn wait_until_control_ready(
instance: &WebInstanceRecord,
child: &mut Child,
) -> anyhow::Result<()> {
let deadline = Instant::now() + STARTUP_TIMEOUT;
loop {
if probe_instance(instance).await {
return Ok(());
}
if let Some(status) = child.try_wait()? {
bail!("A3S Web exited before its control endpoint was ready ({status})");
}
if Instant::now() >= deadline {
let _ = child.kill();
let _ = child.wait();
bail!("A3S Web control endpoint did not become ready in time");
}
sleep(POLL_INTERVAL).await;
}
}
#[cfg(unix)]
fn configure_detached(command: &mut Command) {
use std::os::unix::process::CommandExt;
command.process_group(0);
}
#[cfg(windows)]
fn configure_detached(command: &mut Command) {
use std::os::windows::process::CommandExt;
const DETACHED_PROCESS: u32 = 0x0000_0008;
const CREATE_NEW_PROCESS_GROUP: u32 = 0x0000_0200;
const CREATE_NO_WINDOW: u32 = 0x0800_0000;
command.creation_flags(DETACHED_PROCESS | CREATE_NEW_PROCESS_GROUP | CREATE_NO_WINDOW);
}
#[cfg(not(any(unix, windows)))]
fn configure_detached(_command: &mut Command) {}
#[cfg(test)]
mod tests {
use super::*;
#[test]
fn foreground_arguments_remove_only_the_background_flag() {
let args = vec![
"--host".to_string(),
"127.0.0.1".to_string(),
"-d".to_string(),
"--port".to_string(),
"29653".to_string(),
];
assert_eq!(
foreground_args(&args).cloned().collect::<Vec<_>>(),
["--host", "127.0.0.1", "--port", "29653"]
);
}
}