single-user service manager
Something went wrong. Try again.
1234567891011121314151617181920212223242526272829303132333435363738394041424344454647484950515253545556575859606162636465666768697071727374757677787980818283848586878889909192939495969798991001011021031041051061071081091101111121131141151161171181191201211221231241251261271281291301311321331341351361371381391401411421431441451461471481491501511521531541551561571581591601611621631641651661671681691701711721731741751761771781791801811821831841851861871881891901911921931941951961971981992002012022032042052062072082092102112122132142152162172182192202212222232242252262272282292302312322332342352362372382392402412422432442452462472482492502512522532542552562572582592602612622632642652662672682692702712722732742752762772782792802812822832842852862872882892902912922932942952962972982993003013023033043053063073083093103113123133143153163173183193203213223233243253263273283293303313323333343353363373383393403413423433443453463473483493503513523533543553563573583593603613623633643653663673683693703713723733743753763773783793803813823833843853863873883893903913923933943953963973983994004014024034044054064074084094104114124134144154164174184194204214224234244254264274284294304314324334344354364374384394404414424434444454464474484494504514524534544554564574584594604614624634644654664674684694704714724734744754764774784794804814824834844854864874884894904914924934944954964974984995005015025035045055065075085095105115125135145155165175185195205215225235245255265275285295305315325335345355365375385395405415425435445455465475485495505515525535545555565575585595605615625635645655665675685695705715725735745755765775785795805815825835845855865875885895905915925935945955965975985996006016026036046056066076086096106116126136146156166176186196206216226236246256266276286296306316326336346356366376386396406416426436446456466476486496506516526536546556566576586596606616626636646656666676686696706716726736746756766776786796806816826836846856866876886896906916926936946956966976986997007017027037047057067077087097107117127137147157167177187197207217227237247257267277287297307317327337347357367377387397407417427437447457467477487497507517527537547557567577587597607617627637647657667677687697707717727737747757767777787797807817827837847857867877887897907917927937947957967977987998008018028038048058068078088098108118128138148158168178188198208218228238248258268278288298308318328338348358368378388398408418428438448458468478488498508518528538548558568578588598608618628638648658668678688698708718728738748758768778788798808818828838848858868878888898908918928938948958968978988999009019029039049059069079089099109119129139149159169179189199209219229239249259269279289299309319329339349359369379389399409419429439449459469479489499509519529539549559569579589599609619629639649659669679689699709719729739749759769779789799809819829839849859869879889899909919929939949959969979989991000100110021003100410051006100710081009101010111012101310141015101610171018101910201021102210231024102510261027102810291030103110321033103410351036103710381039104010411042104310441045104610471048104910501051105210531054105510561057105810591060106110621063106410651066106710681069107010711072107310741075107610771078107910801081108210831084108510861087108810891090109110921093109410951096109710981099110011011102110311041105110611071108110911101111111211131114111511161117111811191120112111221123112411251126112711281129113011311132113311341135113611371138113911401141114211431144114511461147114811491150115111521153115411551156115711581159#[allow(clippy::missing_panics_doc, clippy::must_use_candidate)]pub mod apogee_core_v1 { hyprwire::include_protocol!("apogee_core_v1"); pub use server::*; pub use spec::*;}
mod condition;mod config;mod ecs;mod environment;mod logging;mod notify;mod unit;
use anyhow::Context as _;use apogee_core_v1::{apogee_manager_v1, apogee_reload_v1};use bevy_ecs::message;use bevy_ecs::prelude::*;use bevy_ecs::system::SystemParam;use calloop::generic;use clap::Parser;use hyprwire::server;use std::collections::{BTreeMap, HashSet};use std::os::fd::AsRawFd;use std::{fs, io, path, process, sync};use unit::{service, task};
#[derive(SystemSet, Debug, Clone, Copy, PartialEq, Eq, Hash)]enum ReconcileSet { Input, Removal, Lifecycle, TaskLifecycle, Sync,}
#[derive(Parser)]struct Args { #[arg(long)] config_path: path::PathBuf, #[arg(long, default_value_t = log::LevelFilter::Info)] log_level: log::LevelFilter,}
struct Apogee { world: World,}
impl hyprwire::Handler<apogee_core_v1::ApogeeCoreV1> for Apogee { fn protocol_filter(&mut self, client: server::ServerClient) -> bool { client.creds().uid.is_root() || client.creds().uid == rustix::process::getuid() }
fn bind(&mut self, object: apogee_manager_v1::ApogeeManagerV1) { for name in self.world.resource::<ecs::ServiceRegistry>().by_name.keys() { object.send_service_added(name); } for name in self.world.resource::<ecs::TaskRegistry>().by_name.keys() { object.send_task_added(name); } for name in self .world .resource::<ecs::ConditionRegistry>() .by_name .keys() { object.send_condition_added(name); } object.send_done();
self.world.resource_mut::<ecs::Managers>().0.push(object); }}
impl hyprwire::Dispatch<apogee_manager_v1::ApogeeManagerV1> for Apogee { #[allow(clippy::panic)] fn event( &mut self, _object: &apogee_manager_v1::ApogeeManagerV1, event: <apogee_manager_v1::ApogeeManagerV1 as hyprwire::Object>::Event<'_>, ) { match event { apogee_manager_v1::Event::Reload { apogee_reload_v1 } => { self.world .trigger(ecs::ReloadRequested(Some(apogee_reload_v1))); } apogee_manager_v1::Event::GetService { apogee_service_v1, name, } => { let service = self .world .resource::<ecs::ServiceRegistry>() .by_name .get(name.as_str()) .copied(); if let Some(service) = service { self.world .entity_mut(service) .get_mut::<service::Objects>() .unwrap_or_else(|| { panic!("service {service:?} is missing its object bindings") }) .0 .push(apogee_service_v1.clone()); self.world .resource_mut::<ecs::ServiceRegistry>() .by_object .insert(apogee_service_v1, service); } else { apogee_service_v1.send_destroyed(); } } apogee_manager_v1::Event::GetCondition { apogee_condition_v1, name, } => { let condition = self .world .resource::<ecs::ConditionRegistry>() .by_name .get(name.as_str()) .copied(); if let Some(condition) = condition { self.world .entity_mut(condition) .get_mut::<condition::Objects>() .unwrap_or_else(|| { panic!("condition {condition:?} is missing its object bindings") }) .0 .push(apogee_condition_v1.clone()); self.world .resource_mut::<ecs::ConditionRegistry>() .by_object .insert(apogee_condition_v1, condition); } else { apogee_condition_v1.send_destroyed(); } } apogee_manager_v1::Event::GetTask { apogee_task_v1, name, } => { let task = self .world .resource::<ecs::TaskRegistry>() .by_name .get(name.as_str()) .copied(); if let Some(task) = task { self.world .entity_mut(task) .get_mut::<task::Objects>() .unwrap_or_else(|| { panic!("condition {task:?} is missing its object bindings") }) .0 .push(apogee_task_v1.clone()); self.world .resource_mut::<ecs::TaskRegistry>() .by_object .insert(apogee_task_v1, task); } else { apogee_task_v1.send_destroyed(); } } apogee_manager_v1::Event::GetEnvironment { apogee_environment_v1, } => { let environment = &self.world.resource::<environment::Environment>().0; for (name, value) in environment { let name = name.to_string_lossy(); let value = value.to_string_lossy(); apogee_environment_v1.send_getenv(name.as_ref(), value.as_ref()); } apogee_environment_v1.send_can_modify(1); apogee_environment_v1.send_done(); } } }}
fn spawn_process( loop_handle: &ecs::LoopHandle, global_environment: &environment::Environment, service_environment: &BTreeMap<String, String>, service: Entity, service_name: sync::Arc<str>, command: &str, args: &[String], notify: &mut dyn notify::Notify,) -> anyhow::Result<unit::Process> { let mut cmd = process::Command::new(command); cmd.args(args) .env_clear() .envs(&global_environment.0) .envs(service_environment) .stdout(process::Stdio::piped()) .stderr(process::Stdio::piped()); let notify_token = notify.configure_command(loop_handle, service, &mut cmd)?;
let mut child = match cmd.spawn() { Ok(child) => child, Err(error) => { if let Some(token) = notify_token { loop_handle.0.remove(token); } return Err(error).with_context(|| format!("failed to spawn {command}")); } }; let id = unit::ProcessId(child.id());
let stdout = child.stdout.take().context("child stdout was not piped")?; let flags = rustix::fs::fcntl_getfl(&stdout)?; rustix::fs::fcntl_setfl(&stdout, flags | rustix::fs::OFlags::NONBLOCK)?;
let stderr = child.stderr.take().context("child stderr was not piped")?; let flags = rustix::fs::fcntl_getfl(&stderr)?; rustix::fs::fcntl_setfl(&stderr, flags | rustix::fs::OFlags::NONBLOCK)?;
let stdout_token = register_service_output( loop_handle, stdout, sync::Arc::clone(&service_name), id, log::Level::Info, apogee_core_v1::LogStream::Stdout, )?; let stderr_token = match register_service_output( loop_handle, stderr, sync::Arc::clone(&service_name), id, log::Level::Warn, apogee_core_v1::LogStream::Stderr, ) { Ok(token) => token, Err(error) => { loop_handle.0.remove(stdout_token); return Err(error.context("failed to register child stderr")); } };
let setup = || -> anyhow::Result<calloop::RegistrationToken> { let pidfd = rustix::process::pidfd_open( rustix::process::Pid::from_child(&child), rustix::process::PidfdFlags::empty(), ) .context("failed to open pidfd")?;
loop_handle .0 .insert_source( generic::Generic::new(pidfd, calloop::Interest::READ, calloop::Mode::Level), move |_, _, apogee| { if apogee.world.get::<task::Task>(service).is_some() { apogee.world.trigger(task::Exited { task: service, process: id, reason: unit::StopReason::Exited, }); } else { apogee.world.trigger(service::Exited { service, process: id, reason: unit::StopReason::Exited, }); } apogee.world.flush(); Ok(calloop::PostAction::Remove) }, ) .map_err(|error| error.error) .context("failed to register pidfd") };
match setup() { Ok(token) => Ok(unit::Process { id, child, token, notify_token, }), Err(error) => { if let Some(token) = notify_token { loop_handle.0.remove(token); } loop_handle.0.remove(stdout_token); loop_handle.0.remove(stderr_token); if let Err(kill_error) = child.kill() { log::debug!("process may already have exited after setup failed: {kill_error}"); } if let Err(wait_error) = child.wait() { log::warn!("failed to reap process after setup failed: {wait_error}"); } Err(error) } }}
fn register_service_output( loop_handle: &ecs::LoopHandle, reader: impl std::os::fd::AsFd + io::Read + 'static, service_name: sync::Arc<str>, process: unit::ProcessId, level: log::Level, stream: apogee_core_v1::LogStream,) -> anyhow::Result<calloop::RegistrationToken> { let mut buffer = Vec::new(); let mut logger = logging::ServiceLogger::new(&service_name, process.0) .map_err(|error| anyhow::anyhow!(error))?; loop_handle .0 .insert_source( generic::Generic::new(reader, calloop::Interest::READ, calloop::Mode::Level), move |_, reader, state| { let reader = unsafe { reader.get_mut() }; let mut chunk = [0; 4096]; loop { match reader.read(&mut chunk) { Ok(0) => { if !buffer.is_empty() { logger.emit(level, message(&buffer)); } return Ok(calloop::PostAction::Remove); } Ok(length) => { buffer.extend_from_slice(&chunk[..length]); while let Some(position) = buffer.iter().position(|byte| *byte == b'\n') { let line = buffer.drain(..=position).collect::<Vec<_>>(); let line: sync::Arc<str> = message(&line).into(); logger.emit(level, &line); state.world.write_message(unit::ServiceOutput { stream, process, line, }); } } Err(error) if error.kind() == io::ErrorKind::WouldBlock => { return Ok(calloop::PostAction::Continue); } Err(error) => { log::warn!( "{}[{}]: failed reading service output: {error}", service_name, process.0 ); return Ok(calloop::PostAction::Remove); } } } }, ) .map_err(|error| anyhow::anyhow!(error.error)) .context("failed to register child stdout/stderr")}
fn message(bytes: &[u8]) -> String { String::from_utf8_lossy(bytes) .trim_end_matches(['\r', '\n']) .to_owned()}
fn handle_restart_requested( restart: On<ecs::RestartRequested>, mut commands: Commands, mut services: Query< ( &unit::Name, &unit::Command, &unit::Args, &unit::Conditions, &service::Ready, &unit::Running, &unit::Environment, &apogee_core_v1::ServiceState, &mut service::RestartCount, &config::Notify, ), (With<service::Service>, Without<unit::Process>), >, fulfilled: Query<(), With<condition::Fulfilled>>, loop_handle: NonSend<ecs::LoopHandle>, global_env: Res<environment::Environment>, mut systemd_notify: ResMut<notify::Systemd>,) { let Ok(( name, command, args, conditions, ready, running, environment, state, mut restart_count, notify, )) = services.get_mut(restart.0) else { return; };
if !matches!(*state, apogee_core_v1::ServiceState::Restarting) { return; }
if !conditions .0 .iter() .all(|condition| fulfilled.contains(*condition)) { commands .entity(restart.0) .insert(apogee_core_v1::ServiceState::Waiting); return; }
log::info!( "restarting service {:?}: {} ({} {:?})", restart.0, name.0, command.0, args.0 );
let notifier: &mut dyn notify::Notify = match notify { config::Notify::None => &mut notify::Dummy, config::Notify::Systemd => &mut *systemd_notify, config::Notify::S6 => &mut notify::S6, }; match spawn_process( &loop_handle, &global_env, &environment.0, restart.0, sync::Arc::clone(&name.0), &command.0, &args.0, notifier, ) { Ok(process) => { restart_count.0 = restart_count.0.saturating_add(1); log::info!("Starting {}[{}]", name.0, process.id.0); commands.entity(restart.0).insert(( process, unit::Timestamp(unit::unix_timestamp()), apogee_core_v1::ServiceState::Running, )); if *notify == config::Notify::None { commands.entity(ready.0).insert(condition::Fulfilled); } commands.entity(running.0).insert(condition::Fulfilled); } Err(error) => { log::error!("failed to restart {}: {error:#}", name.0); commands .entity(restart.0) .insert(apogee_core_v1::ServiceState::Failed); } }}
#[derive(SystemParam)]struct UnitParams<'w, 's> { commands: Query<'w, 's, &'static unit::Command>, descriptions: Query<'w, 's, &'static unit::Description>, args: Query<'w, 's, &'static unit::Args>, environments: Query<'w, 's, &'static unit::Environment>, conditions: Query<'w, 's, &'static unit::Conditions>, running: Query<'w, 's, &'static unit::Running>,}
#[derive(SystemParam)]struct ServiceParams<'w, 's> { ready: Query<'w, 's, &'static service::Ready>, restart_delays: Query<'w, 's, &'static service::RestartDelay>, restart_policies: Query<'w, 's, &'static apogee_core_v1::RestartPolicy>, notifies: Query<'w, 's, &'static config::Notify>, existing: Query<'w, 's, (Entity, &'static unit::Name), With<service::Service>>,}
#[derive(SystemParam)]struct TaskParams<'w, 's> { successful: Query<'w, 's, &'static task::Successful>, existing: Query<'w, 's, (Entity, &'static unit::Name), With<task::Task>>,}
fn handle_reload( reload: On<ecs::ReloadRequested>, mut commands: Commands, unit: UnitParams, service: ServiceParams, task: TaskParams, config: Res<config::Config>, mut services: ResMut<ecs::ServiceRegistry>, mut conditions: ResMut<ecs::ConditionRegistry>, mut tasks: ResMut<ecs::TaskRegistry>, managers: Res<ecs::Managers>,) { let UnitParams { commands: unit_commands, descriptions: unit_descriptions, args: unit_args, environments: unit_environments, conditions: unit_conditions, running: unit_running, } = unit; let ServiceParams { ready: service_ready, restart_delays: service_restart_delays, restart_policies: service_restart_policies, notifies: service_notifies, existing: existing_services, } = service; let TaskParams { successful: task_successful, existing: existing_tasks, } = task;
match config.load() { Ok(config) => { for configured_condition in config.conditions { condition::resolve( &mut commands, &mut conditions, &managers, configured_condition.name, configured_condition.immutable, ); }
let desired: HashSet<_> = config .services .iter() .map(|service| service.name.as_ref()) .collect(); for (entity, name) in existing_services.iter() { if !desired.contains(&*name.0) { commands.entity(entity).remove::<service::Service>(); } }
for service in config.services { let resolved_conditions = service .conditions .iter() .map(|condition| { condition::resolve( &mut commands, &mut conditions, &managers, sync::Arc::clone(&condition.name), condition.immutable, ) }) .collect::<Vec<_>>(); let ready = condition::resolve( &mut commands, &mut conditions, &managers, service.ready.name, service.ready.immutable, ); let running = condition::resolve( &mut commands, &mut conditions, &managers, service.running.name, service.running.immutable, );
if let Some(entity) = services.by_name.get(&service.name) { if unit_descriptions .get(*entity) .is_ok_and(|current| current.0 != service.description) { commands .entity(*entity) .insert(unit::Description(service.description)); } if unit_commands .get(*entity) .is_ok_and(|current| current.0.as_str() != service.command) { commands .entity(*entity) .insert(unit::Command(service.command)); } if unit_args .get(*entity) .is_ok_and(|current| current.0 != service.args) { commands.entity(*entity).insert(unit::Args(service.args)); } if unit_environments .get(*entity) .is_ok_and(|current| current.0 != service.env) { commands .entity(*entity) .insert(unit::Environment(service.env)); } if unit_conditions .get(*entity) .is_ok_and(|current| current.0 != resolved_conditions) { commands .entity(*entity) .insert(unit::Conditions(resolved_conditions)); } if service_ready .get(*entity) .is_ok_and(|current| current.0 != ready) { commands.entity(*entity).insert(service::Ready(ready)); } if unit_running .get(*entity) .is_ok_and(|current| current.0 != running) { commands.entity(*entity).insert(unit::Running(running)); } if service_restart_delays .get(*entity) .is_ok_and(|current| current.0 != service.restart.delay) { commands .entity(*entity) .insert(service::RestartDelay(service.restart.delay)); } if service_restart_policies .get(*entity) .is_ok_and(|current| *current != service.restart.on) { commands.entity(*entity).insert(service.restart.on); } if service_notifies.get(*entity) != Ok(&service.notify) { commands.entity(*entity).insert(service.notify); } } else { service::spawn( &mut commands, &mut services, &managers, service.name, service.description, service.command, service.args, resolved_conditions, ready, running, service.env, service.restart.delay, service.restart.on, service.notify, ); } }
let desired: HashSet<_> = config.tasks.iter().map(|task| task.name.as_ref()).collect(); for (entity, name) in existing_tasks.iter() { if !desired.contains(&*name.0) { commands.entity(entity).remove::<task::Task>(); } }
for task in config.tasks { let resolved_conditions = task .conditions .iter() .map(|condition| { condition::resolve( &mut commands, &mut conditions, &managers, sync::Arc::clone(&condition.name), condition.immutable, ) }) .collect::<Vec<_>>(); let successful = condition::resolve( &mut commands, &mut conditions, &managers, task.successful.name, task.successful.immutable, ); let running = condition::resolve( &mut commands, &mut conditions, &managers, task.running.name, task.running.immutable, );
if let Some(entity) = tasks.by_name.get(&task.name) { if unit_descriptions .get(*entity) .is_ok_and(|current| current.0 != task.description) { commands .entity(*entity) .insert(unit::Description(task.description)); } if unit_commands .get(*entity) .is_ok_and(|current| current.0.as_str() != task.command) { commands.entity(*entity).insert(unit::Command(task.command)); } if unit_args .get(*entity) .is_ok_and(|current| current.0 != task.args) { commands.entity(*entity).insert(unit::Args(task.args)); } if unit_environments .get(*entity) .is_ok_and(|current| current.0 != task.env) { commands.entity(*entity).insert(unit::Environment(task.env)); } if unit_conditions .get(*entity) .is_ok_and(|current| current.0 != resolved_conditions) { commands .entity(*entity) .insert(unit::Conditions(resolved_conditions)); } if task_successful .get(*entity) .is_ok_and(|current| current.0 != successful) { commands .entity(*entity) .insert(task::Successful(successful)); } if unit_running .get(*entity) .is_ok_and(|current| current.0 != running) { commands.entity(*entity).insert(unit::Running(running)); } } else { task::spawn( &mut commands, &mut tasks, &managers, task.name, task.description, task.command, task.args, resolved_conditions, successful, running, task.env, ); } }
if let Some(object) = reload.0.as_ref() { object.send_finished(); } } Err(e) => { log::error!("failed to load service configuration: {e:#}"); if let Some(object) = reload.0.as_ref() { object.send_failed(apogee_core_v1::ReloadError::Io, e.to_string()); } } }}
fn service_modified( mut commands: Commands, mut services: Query< ( Entity, &unit::Name, &unit::Command, &unit::Args, &unit::Conditions, Ref<service::Service>, Option<&mut unit::Process>, &service::Ready, &unit::Running, &unit::Environment, &config::Notify, ), Or<( Changed<unit::Command>, Changed<unit::Args>, Changed<unit::Environment>, Changed<unit::Conditions>, Changed<service::Ready>, Changed<unit::Running>, Changed<config::Notify>, )>, >, fulfilled: Query<(), With<condition::Fulfilled>>, loop_handle: NonSend<ecs::LoopHandle>, global_env: Res<environment::Environment>, mut systemd_notify: ResMut<notify::Systemd>,) { for ( entity, name, command, args, conditions, service, process, ready, running, environment, notify, ) in services.iter_mut() { if service.is_added() { continue; }
log::info!( "restarting service {entity:?}: {} ({} {:?})", name.0, command.0, args.0 );
if let Some(mut process) = process { process.remove_sources(&loop_handle); log::info!("Stopping {}[{}], sending SIGKILL ...", name.0, process.id.0); if let Err(error) = process.child.kill() { log::debug!("old process may already have exited: {error}"); }
let event = service::Exited { service: entity, process: process.id, reason: unit::StopReason::Restarted, }; commands.queue(move |world: &mut World| { world.trigger(event); world.flush(); }); }
if conditions .0 .iter() .all(|condition| fulfilled.contains(*condition)) { let notifier: &mut dyn notify::Notify = match notify { config::Notify::None => &mut notify::Dummy, config::Notify::Systemd => &mut *systemd_notify, config::Notify::S6 => &mut notify::S6, }; match spawn_process( &loop_handle, &global_env, &environment.0, entity, sync::Arc::clone(&name.0), &command.0, &args.0, notifier, ) { Ok(process) => { commands.entity(entity).insert(( process, unit::Timestamp(unit::unix_timestamp()), apogee_core_v1::ServiceState::Running, ));
log::info!("restarted service: {}", name.0); if *notify == config::Notify::None { commands.entity(ready.0).insert(condition::Fulfilled); } commands.entity(running.0).insert(condition::Fulfilled); } Err(error) => { log::error!("failed to restart {}: {error:#}", name.0); commands .entity(entity) .insert(apogee_core_v1::ServiceState::Failed); } } } else { commands .entity(entity) .insert(apogee_core_v1::ServiceState::Waiting); } }}
fn start_services( mut commands: Commands, added_services: Query<(), Added<service::Service>>, added_conditions: Query<(), Added<condition::Fulfilled>>, services: Query< ( Entity, &unit::Name, &unit::Command, &unit::Args, &unit::Conditions, &service::Ready, &unit::Running, &unit::Environment, &config::Notify, ), (With<service::Service>, Without<unit::Process>), >, fulfilled: Query<(), With<condition::Fulfilled>>, loop_handle: NonSend<ecs::LoopHandle>, loop_signal: Res<ecs::LoopSignal>, global_env: Res<environment::Environment>, mut systemd_notify: ResMut<notify::Systemd>,) { if added_services.is_empty() && added_conditions.is_empty() { return; }
let any_started = services .iter() .filter(|(_, _, _, _, conditions, _, _, _, _)| { conditions .0 .iter() .all(|condition| fulfilled.contains(*condition)) }) .fold( false, |any_started, (entity, name, command, args, _, ready, running, environment, notify)| { let notifier: &mut dyn notify::Notify = match notify { config::Notify::None => &mut notify::Dummy, config::Notify::Systemd => &mut *systemd_notify, config::Notify::S6 => &mut notify::S6, }; (match spawn_process( &loop_handle, &global_env, &environment.0, entity, sync::Arc::clone(&name.0), &command.0, &args.0, notifier, ) { Ok(process) => { log::info!("Starting {}[{}]", name.0, process.id.0); commands.entity(entity).insert(( process, unit::Timestamp(unit::unix_timestamp()), apogee_core_v1::ServiceState::Running, )); if *notify == config::Notify::None { commands.entity(ready.0).insert(condition::Fulfilled); } commands.entity(running.0).insert(condition::Fulfilled); true } Err(error) => { log::error!("failed to start {}: {error:#}", name.0); commands .entity(entity) .insert(apogee_core_v1::ServiceState::Failed); false } }) || any_started }, );
if any_started { loop_signal.0.wakeup(); }}
fn start_tasks( mut commands: Commands, added_tasks: Query<(), Added<task::Task>>, added_conditions: Query<(), Added<condition::Fulfilled>>, tasks: Query< ( Entity, &unit::Name, &unit::Command, &unit::Args, &unit::Conditions, &unit::Running, &unit::Environment, &apogee_core_v1::TaskState, ), (With<task::Task>, Without<unit::Process>), >, fulfilled: Query<(), With<condition::Fulfilled>>, loop_handle: NonSend<ecs::LoopHandle>, loop_signal: Res<ecs::LoopSignal>, global_env: Res<environment::Environment>,) { if added_tasks.is_empty() && added_conditions.is_empty() { return; }
let any_started = tasks .iter() .filter(|(_, _, _, _, conditions, _, _, state)| { matches!(state, apogee_core_v1::TaskState::Waiting) && conditions .0 .iter() .all(|condition| fulfilled.contains(*condition)) }) .fold( false, |any_started, (entity, name, command, args, _, running, environment, _)| { (match crate::spawn_process( &loop_handle, &global_env, &environment.0, entity, sync::Arc::clone(&name.0), &command.0, &args.0, &mut notify::Dummy, ) { Ok(process) => { log::info!("Starting task {}[{}]", name.0, process.id.0); commands.entity(entity).insert(( process, unit::Timestamp(unit::unix_timestamp()), apogee_core_v1::TaskState::Running, )); commands.entity(running.0).insert(condition::Fulfilled); true } Err(error) => { log::error!("failed to start task {}: {error:#}", name.0); commands .entity(entity) .insert(apogee_core_v1::TaskState::Failed); false } }) || any_started }, );
if any_started { loop_signal.0.wakeup(); }}
fn sync_managers(mut managers: ResMut<ecs::Managers>) { managers.0.retain(|manager| manager.try_client().is_some()); for manager in &managers.0 { manager.send_done(); }}
fn main() -> anyhow::Result<()> { let args = Args::parse();
logging::init_syslog(syslog::Facility::LOG_DAEMON, args.log_level)?;
let mut event_loop = calloop::EventLoop::try_new()?;
let mut world = World::new(); message::MessageRegistry::register_message::<unit::ServiceOutput>(&mut world); let mut schedule = Schedule::new(ecs::Reconcile); schedule.add_systems(( (message::message_update_system, unit::store_service_output) .chain() .in_set(ReconcileSet::Input), (service::removed, task::removed) .after(ReconcileSet::Input) .in_set(ReconcileSet::Removal), (start_services, service_modified) .after(ReconcileSet::Removal) .in_set(ReconcileSet::Lifecycle), start_tasks .after(ReconcileSet::Lifecycle) .in_set(ReconcileSet::TaskLifecycle), (unit::sync_objects, condition::sync_objects) .after(ReconcileSet::TaskLifecycle) .in_set(ReconcileSet::Sync), sync_managers.after(ReconcileSet::Sync), )); world.init_resource::<ecs::ServiceRegistry>(); world.init_resource::<ecs::ConditionRegistry>(); world.init_resource::<ecs::TaskRegistry>(); world.init_resource::<ecs::Managers>(); world.init_resource::<notify::Systemd>(); world.init_resource::<environment::Environment>(); world.insert_resource(config::Config(args.config_path)); world.insert_non_send(ecs::LoopHandle(event_loop.handle())); world.insert_resource(ecs::LoopSignal(event_loop.get_signal()));
world.add_observer(handle_reload); world.add_observer(service::handle_exited); world.add_observer(task::handle_exited); world.add_observer(handle_restart_requested); world.add_observer(condition::handle_start); world.add_observer(condition::handle_stop); world.add_observer(service::handle_start); world.add_observer(service::handle_stop); world.add_observer(service::handle_restart); world.add_observer(task::handle_start); world.add_observer(task::handle_stop); world.add_observer(task::handle_restart); world.add_observer(unit::handle_new_log_object); world.flush();
world.trigger(ecs::ReloadRequested(None));
schedule.run(&mut world);
let mut apogee = Apogee { world };
let runtime_dir = dirs::runtime_dir() .unwrap_or_else(|| path::PathBuf::from("run")) .join("apogee"); fs::create_dir_all(&runtime_dir)?;
let socket_path = runtime_dir.join("apogee.sock"); let mut socket = server::Server::bind(&socket_path)?;
socket.add_implementation::<apogee_core_v1::ApogeeCoreV1, _>(&mut apogee, 1);
let fd_wrapper = unsafe { generic::FdWrapper::new(socket.extract_loop_fd().as_raw_fd()) }; let source = generic::Generic::new( fd_wrapper, calloop::Interest { readable: true, writable: false, }, calloop::Mode::Level, ); event_loop .handle() .insert_source(source, move |_, _, state| { _ = socket.dispatch_events(state, false); Ok(calloop::PostAction::Continue) }) .map_err(|err| anyhow::anyhow!("failed to insert VT watcher: {err:?}"))?;
event_loop.run(None, &mut apogee, |state| { schedule.run(&mut state.world); })?;
Ok(())}
hyprwire::delegate_noop!(Apogee: apogee_reload_v1::ApogeeReloadV1);