@@ -1,0 +1,473 @@
//! Windows Service (SCM) integration — the peer of the systemd/launchd
//! code on Unix. One file holds everything Windows-specific: installing /
//! controlling the service via the SCM, and the in-process service worker
//! that the SCM launches on boot.
//!
//! ## How it runs
//!
//! `runner service install` registers a service whose image path is
//! `anvil.exe runner service run --anvil-service-worker --service-name <i> …`.
//! On boot the SCM launches that command; `main()` spots the
//! [`SERVICE_WORKER_FLAG`] before building the async runtime and calls
//! [`run_dispatcher`], which hands control to the SCM dispatcher. The
//! dispatcher invokes [`service_main`] on a background thread, where we
//! register a control handler (translating `SERVICE_CONTROL_STOP` into the
//! same graceful-drain `Notify` the rest of the runner uses), report
//! `Running`, then run the normal poll loop until stop.
use std::ffi::{OsStr, OsString};
use std::path::{Path, PathBuf};
use std::sync::OnceLock;
use std::time::Duration;
use windows_service::service::{
ServiceAccess, ServiceAction, ServiceActionType, ServiceControl, ServiceControlAccept,
ServiceErrorControl, ServiceExitCode, ServiceFailureActions, ServiceFailureResetPeriod,
ServiceInfo, ServiceStartType, ServiceState, ServiceStatus, ServiceType,
};
use windows_service::service_control_handler::{self, ServiceControlHandlerResult};
use windows_service::service_dispatcher;
use windows_service::service_manager::{ServiceManager, ServiceManagerAccess};
use crate::output;
use crate::runner::service_mode::{InstallRecord, Scope};
/// Hidden flag baked into the service image path so the worker invocation
/// is unambiguously distinguishable from a normal `runner service` call.
pub const SERVICE_WORKER_FLAG: &str = "--anvil-service-worker";
/// SCM service name for an instance. Mirrors `unit_or_label_for` in the
/// command layer so both agree on the name without sharing state.
pub fn service_name_for(instance: &str) -> String {
format!("anvil-runner-{instance}")
}
/// Where the service worker writes stdout/stderr (services have no
/// console). `runner logs` tails this file.
pub fn service_log_path(instance: &str) -> PathBuf {
dirs::home_dir()
.unwrap_or_else(|| PathBuf::from("."))
.join(".anvil-runner")
.join(format!("{instance}.log"))
}
type DynErr = Box<dyn std::error::Error + Send + Sync>;
// ─────────────────────────── install / uninstall / control ──────────────
/// Create the SCM service for `instance`. Requires an elevated shell.
/// Returns a sentinel path (there is no on-disk unit file) for the install
/// record.
#[allow(clippy::too_many_arguments)]
pub fn install(
exe: &Path,
config_file: &Path,
scope: Scope,
user_account: Option<&str>,
parallel: u32,
instance: &str,
service_name: &str,
) -> Result<PathBuf, DynErr> {
if !crate::platform::is_elevated() {
return Err(
"installing a Windows service requires an elevated (Administrator) terminal".into(),
);
}
let manager = ServiceManager::local_computer(
None::<&str>,
ServiceManagerAccess::CONNECT | ServiceManagerAccess::CREATE_SERVICE,
)?;
// Pre-create state/log dirs so the (possibly non-admin) service account
// doesn't have to.
let pid_file = crate::runner::pid_file::instance_path(instance);
let log_path = service_log_path(instance);
if let Some(p) = log_path.parent() {
let _ = std::fs::create_dir_all(p);
}
if let Some(p) = pid_file.parent() {
let _ = std::fs::create_dir_all(p);
}
let launch_arguments: Vec<OsString> = vec![
OsString::from("runner"),
OsString::from("service"),
OsString::from("run"),
OsString::from(SERVICE_WORKER_FLAG),
OsString::from("--service-name"),
OsString::from(instance),
OsString::from("--config"),
config_file.as_os_str().to_os_string(),
OsString::from("--pid-file"),
pid_file.as_os_str().to_os_string(),
OsString::from("--parallel"),
OsString::from(parallel.to_string()),
OsString::from("--shutdown-timeout"),
OsString::from("60"),
];
// Account mapping: System scope → LocalSystem (account None). User scope
// with an explicit account → that account (prompts for a password). User
// scope without an account → LocalSystem too (machine-wide SCM has no
// per-user analogue; documented behaviour).
let (account_name, account_password) = match (scope, user_account) {
(Scope::User, Some(acct)) => {
let password = dialoguer::Password::new()
.with_prompt(format!("Password for service account '{acct}'"))
.interact()
.map_err(|e| format!("could not read password: {e}"))?;
(Some(OsString::from(acct)), Some(OsString::from(password)))
}
_ => (None, None),
};
let info = ServiceInfo {
name: OsString::from(service_name),
display_name: OsString::from(format!("Anvil CI Runner ({instance})")),
service_type: ServiceType::OWN_PROCESS,
start_type: ServiceStartType::AutoStart,
error_control: ServiceErrorControl::Normal,
executable_path: exe.to_path_buf(),
launch_arguments,
dependencies: vec![],
account_name,
account_password,
};
let service = manager.create_service(
&info,
ServiceAccess::CHANGE_CONFIG
| ServiceAccess::START
| ServiceAccess::STOP
| ServiceAccess::QUERY_STATUS
| ServiceAccess::DELETE,
)?;
let _ = service.set_description(format!(
"Anvil CI runner instance '{instance}'. Polls the Anvil server for CI jobs and executes them."
));
// Auto-restart on failure: restart after 5s, again after 5s, then 30s,
// resetting the failure counter after a day of clean running. Mirrors
// systemd's Restart=always / RestartSec=5.
let failure_actions = ServiceFailureActions {
reset_period: ServiceFailureResetPeriod::After(Duration::from_secs(86_400)),
reboot_msg: None,
command: None,
actions: Some(vec![
ServiceAction {
action_type: ServiceActionType::Restart,
delay: Duration::from_secs(5),
},
ServiceAction {
action_type: ServiceActionType::Restart,
delay: Duration::from_secs(5),
},
ServiceAction {
action_type: ServiceActionType::Restart,
delay: Duration::from_secs(30),
},
]),
};
if let Err(e) = service.update_failure_actions(failure_actions) {
output::warn(&format!(
"could not set restart-on-failure policy (service will still run): {e}"
));
}
// No on-disk unit; record a sentinel so status/doctor have something
// human-readable to show.
Ok(PathBuf::from(format!("SCM:{service_name}")))
}
/// Stop (if running) and delete the SCM service.
pub fn uninstall(record: &InstallRecord) -> Result<(), DynErr> {
let manager = ServiceManager::local_computer(None::<&str>, ServiceManagerAccess::CONNECT)?;
let service = manager.open_service(
&record.unit_or_label,
ServiceAccess::QUERY_STATUS | ServiceAccess::STOP | ServiceAccess::DELETE,
)?;
if let Ok(status) = service.query_status() {
if status.current_state != ServiceState::Stopped {
let _ = service.stop();
wait_for_state(&service, ServiceState::Stopped, Duration::from_secs(30));
}
}
service.delete()?;
Ok(())
}
/// `svc-start` / `svc-stop` / `svc-restart` / `svc-status`.
pub fn control(action: &str, record: &InstallRecord) -> Result<(), DynErr> {
let manager = ServiceManager::local_computer(None::<&str>, ServiceManagerAccess::CONNECT)?;
let access = ServiceAccess::QUERY_STATUS | ServiceAccess::START | ServiceAccess::STOP;
let service = manager.open_service(&record.unit_or_label, access)?;
match action {
"start" => {
service.start::<&OsStr>(&[])?;
output::success("Service started");
}
"stop" => {
service.stop()?;
output::success("Service stopped");
}
"restart" => {
let _ = service.stop();
wait_for_state(&service, ServiceState::Stopped, Duration::from_secs(30));
service.start::<&OsStr>(&[])?;
output::success("Service restarted");
}
"status" => {
let status = service.query_status()?;
output::detail("State", &format!("{:?}", status.current_state));
if let Some(pid) = status.process_id {
output::detail("PID", &pid.to_string());
}
}
other => return Err(format!("unknown service action: {other}").into()),
}
Ok(())
}
/// Poll a service until it reaches `target` or the deadline elapses.
fn wait_for_state(
service: &windows_service::service::Service,
target: ServiceState,
timeout: Duration,
) {
let deadline = std::time::Instant::now() + timeout;
while std::time::Instant::now() < deadline {
match service.query_status() {
Ok(s) if s.current_state == target => return,
_ => std::thread::sleep(Duration::from_millis(200)),
}
}
}
// ─────────────────────────── service worker (SCM-launched) ──────────────
#[derive(Clone)]
struct ServiceRunConfig {
instance: String,
config_path: Option<String>,
pid_file: Option<String>,
parallel: Option<u32>,
shutdown_timeout: u64,
log_path: PathBuf,
}
static RUN_CONFIG: OnceLock<ServiceRunConfig> = OnceLock::new();
/// True when this process was launched by the SCM as the service worker.
pub fn is_service_invocation() -> bool {
std::env::args().any(|a| a == SERVICE_WORKER_FLAG)
}
/// Parse the worker flags from the process command line (the SCM passes
/// them as part of the service image path, so they land in `env::args`).
fn parse_run_config() -> ServiceRunConfig {
let mut instance = crate::runner::service_mode::DEFAULT_INSTANCE.to_string();
let mut config_path: Option<String> = None;
let mut pid_file: Option<String> = None;
let mut parallel: Option<u32> = None;
let mut shutdown_timeout: u64 = crate::runner::loop_runner::DEFAULT_SHUTDOWN_TIMEOUT_SECS;
let args: Vec<String> = std::env::args().collect();
let mut i = 0;
while i < args.len() {
match args[i].as_str() {
"--service-name" => {
if let Some(v) = args.get(i + 1) {
instance = v.clone();
i += 1;
}
}
"--config" => {
config_path = args.get(i + 1).cloned();
i += 1;
}
"--pid-file" => {
pid_file = args.get(i + 1).cloned();
i += 1;
}
"--parallel" => {
parallel = args.get(i + 1).and_then(|v| v.parse().ok());
i += 1;
}
"--shutdown-timeout" => {
if let Some(v) = args.get(i + 1).and_then(|v| v.parse().ok()) {
shutdown_timeout = v;
i += 1;
}
}
_ => {}
}
i += 1;
}
let log_path = service_log_path(&instance);
ServiceRunConfig {
instance,
config_path,
pid_file,
parallel,
shutdown_timeout,
log_path,
}
}
windows_service::define_windows_service!(ffi_service_main, service_main);
/// Hand off to the SCM dispatcher. Blocks until the service stops. Returns
/// an error if not actually launched by the SCM (e.g. run by hand).
pub fn run_dispatcher() -> Result<(), DynErr> {
let cfg = parse_run_config();
let service_name = service_name_for(&cfg.instance);
let _ = RUN_CONFIG.set(cfg);
service_dispatcher::start(service_name, ffi_service_main)?;
Ok(())
}
fn service_main(_arguments: Vec<OsString>) {
if let Err(e) = run_service_worker() {
// Last-resort: the control handler may not be registered yet, so we
// can only log. stdout/stderr are redirected to the log file by the
// time most errors occur.
eprintln!("anvil runner service worker error: {e}");
}
}
fn run_service_worker() -> Result<(), DynErr> {
let cfg = RUN_CONFIG
.get()
.cloned()
.ok_or("service worker started without a parsed run config")?;
// Redirect stdout/stderr to the log file before anything prints, so the
// runner's eprintln!-based logging is captured for `runner logs`.
let _ = redirect_stdio_to_log(&cfg.log_path);
// The SCM stop control feeds the same graceful-drain notify the rest of
// the runner waits on (see shutdown.rs). The status handle is shared
// into the handler (set just after `register` returns) so STOP can
// report `StopPending` with a wait hint — otherwise the SCM may treat a
// long in-flight-job drain as a hang.
let shutdown = crate::runner::shutdown::external_stop();
let handler_shutdown = shutdown.clone();
let status_shared: std::sync::Arc<
std::sync::OnceLock<windows_service::service_control_handler::ServiceStatusHandle>,
> = std::sync::Arc::new(std::sync::OnceLock::new());
let status_for_handler = status_shared.clone();
let event_handler = move |control| -> ServiceControlHandlerResult {
match control {
ServiceControl::Stop => {
if let Some(handle) = status_for_handler.get() {
let _ = handle.set_service_status(ServiceStatus {
service_type: ServiceType::OWN_PROCESS,
current_state: ServiceState::StopPending,
controls_accepted: ServiceControlAccept::empty(),
exit_code: ServiceExitCode::Win32(0),
checkpoint: 0,
wait_hint: Duration::from_secs(90),
process_id: None,
});
}
handler_shutdown.notify_one();
ServiceControlHandlerResult::NoError
}
ServiceControl::Interrogate => ServiceControlHandlerResult::NoError,
_ => ServiceControlHandlerResult::NotImplemented,
}
};
let status_handle =
service_control_handler::register(service_name_for(&cfg.instance), event_handler)?;
let _ = status_shared.set(status_handle);
status_handle.set_service_status(ServiceStatus {
service_type: ServiceType::OWN_PROCESS,
current_state: ServiceState::Running,
controls_accepted: ServiceControlAccept::STOP,
exit_code: ServiceExitCode::Win32(0),
checkpoint: 0,
wait_hint: Duration::default(),
process_id: None,
})?;
// The runner loop is async; build a runtime here (the SCM thread is not
// the tokio main thread).
let runtime = tokio::runtime::Builder::new_multi_thread()
.enable_all()
.build()?;
let result: Result<(), DynErr> = runtime.block_on(async move {
let mut config = crate::runner::RunnerConfig::load(cfg.config_path.as_deref())?;
if let Some(p) = cfg.parallel {
config.parallel = p;
}
let pid_path = cfg
.pid_file
.clone()
.map(PathBuf::from)
.unwrap_or_else(|| crate::runner::pid_file::instance_path(&cfg.instance));
crate::runner::loop_runner::start(config, pid_path, cfg.shutdown_timeout).await
});
let exit_code = if result.is_ok() {
ServiceExitCode::Win32(0)
} else {
ServiceExitCode::ServiceSpecific(1)
};
let _ = status_handle.set_service_status(ServiceStatus {
service_type: ServiceType::OWN_PROCESS,
current_state: ServiceState::Stopped,
controls_accepted: ServiceControlAccept::empty(),
exit_code,
checkpoint: 0,
wait_hint: Duration::default(),
process_id: None,
});
result
}
/// Point the process's STDOUT/STDERR handles at the log file (append). Done
/// before the first `print!`/`eprintln!` so Rust's lazily-cached std
/// handles pick up the redirected target.
fn redirect_stdio_to_log(path: &Path) -> std::io::Result<()> {
use std::os::windows::ffi::OsStrExt;
use windows_sys::Win32::Foundation::{GENERIC_WRITE, INVALID_HANDLE_VALUE};
use windows_sys::Win32::Storage::FileSystem::{
CreateFileW, FILE_ATTRIBUTE_NORMAL, FILE_SHARE_READ, FILE_SHARE_WRITE, OPEN_ALWAYS,
};
use windows_sys::Win32::System::Console::{SetStdHandle, STD_ERROR_HANDLE, STD_OUTPUT_HANDLE};
if let Some(parent) = path.parent() {
let _ = std::fs::create_dir_all(parent);
}
let wide: Vec<u16> = path
.as_os_str()
.encode_wide()
.chain(std::iter::once(0))
.collect();
unsafe {
let handle = CreateFileW(
wide.as_ptr(),
GENERIC_WRITE,
FILE_SHARE_READ | FILE_SHARE_WRITE,
std::ptr::null(),
OPEN_ALWAYS,
FILE_ATTRIBUTE_NORMAL,
std::ptr::null_mut(),
);
if handle == INVALID_HANDLE_VALUE {
return Err(std::io::Error::last_os_error());
}
SetStdHandle(STD_OUTPUT_HANDLE, handle);
SetStdHandle(STD_ERROR_HANDLE, handle);
}
Ok(())
}