single-user service manager
Something went wrong. Try again.
123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457458459460461462463464465466467468469470471472473474475476477478479480481482483484485486487488489490491492493494495496497use crate::apogee_core_v1::apogee_task_v1;use crate::{apogee_core_v1, condition, ecs, environment, notify};use bevy_ecs::prelude::*;use std::collections::BTreeMap;use std::sync;
#[derive(Component)]pub struct Task;
#[derive(Component, Default)]pub struct Objects(pub Vec<apogee_task_v1::ApogeeTaskV1>);
#[derive(Component)]pub struct Successful(pub Entity);
#[derive(Bundle)]pub struct Bundle { pub marker: Task, pub name: super::Name, pub description: super::Description, pub command: super::Command, pub args: super::Args, pub conditions: super::Conditions, pub successful: Successful, pub running: super::Running, pub environment: super::Environment, pub state: apogee_core_v1::TaskState, pub objects: Objects, pub logs: super::Logs, pub log_objects: super::LogObjects, pub timestamp: super::Timestamp,}
pub fn spawn( commands: &mut Commands, tasks: &mut ecs::TaskRegistry, managers: &ecs::Managers, name: sync::Arc<str>, description: String, command: String, args: Vec<String>, conditions: Vec<Entity>, successful: Entity, running: Entity, environment: BTreeMap<String, String>,) -> Entity { let entity = commands .spawn(Bundle { marker: Task, name: super::Name(sync::Arc::clone(&name)), description: super::Description(description), command: super::Command(command), args: super::Args(args), conditions: super::Conditions(conditions), successful: Successful(successful), running: super::Running(running), environment: super::Environment(environment), state: apogee_core_v1::TaskState::Waiting, objects: Objects::default(), logs: super::Logs::default(), log_objects: super::LogObjects::default(), timestamp: super::Timestamp::default(), }) .id(); tasks.by_name.insert(sync::Arc::clone(&name), entity); managers.task_added(&name); entity}
impl hyprwire::Dispatch<apogee_task_v1::ApogeeTaskV1> for crate::Apogee { fn event( &mut self, object: &apogee_task_v1::ApogeeTaskV1, event: <apogee_task_v1::ApogeeTaskV1 as hyprwire::Object>::Event<'_>, ) { let Some(task) = self .world .resource_mut::<ecs::TaskRegistry>() .by_object .get(object) .copied() else { return; };
match event { apogee_task_v1::Event::Start => { self.world.trigger(Start { entity: task, object: object.clone(), }); } apogee_task_v1::Event::Stop => { self.world.trigger(Stop { entity: task, object: object.clone(), }); } apogee_task_v1::Event::Restart => { self.world.trigger(Restart { entity: task, object: object.clone(), }); } apogee_task_v1::Event::GetLogs { apogee_service_logs_v1, } => { self.world.trigger(super::NewLogObject { entity: task, object: apogee_service_logs_v1, }); } } }}
#[derive(Event)]pub struct Stop { entity: Entity, object: apogee_task_v1::ApogeeTaskV1,}
pub fn handle_stop( stop: On<Stop>, mut commands: Commands, mut tasks: Query< ( &super::Name, &mut apogee_core_v1::TaskState, &super::Running, &Successful, Option<&mut super::Process>, ), With<Task>, >, loop_handle: NonSend<ecs::LoopHandle>,) { let Ok((name, mut state, running, successful, process)) = tasks.get_mut(stop.entity) else { stop.object.send_error( apogee_core_v1::TaskError::Unavailable, "task is no longer available", ); return; }; let Some(mut process) = process else { *state = apogee_core_v1::TaskState::Stopped; commands .entity(successful.0) .remove::<condition::Fulfilled>(); stop.object.send_ok(); return; };
log::info!("Stopping {}[{}], sending SIGKILL ...", name.0, process.id.0); if let Err(kill_error) = process.child.kill() { log::debug!("failed to stop {}: {kill_error}", name.0); stop.object.send_error( apogee_core_v1::TaskError::KillFailed, kill_error.to_string(), ); return; } process.remove_sources(&loop_handle); if let Err(wait_error) = process.child.wait() { log::warn!("failed to reap {} after stopping it: {wait_error}", name.0); stop.object.send_error( apogee_core_v1::TaskError::WaitFailed, wait_error.to_string(), ); return; }
*state = apogee_core_v1::TaskState::Stopped; commands.entity(stop.entity).remove::<super::Process>(); commands.entity(stop.entity).insert(super::Timestamp(0)); commands .entity(successful.0) .remove::<condition::Fulfilled>(); commands.entity(running.0).remove::<condition::Fulfilled>();
stop.object.send_ok();}
#[derive(Event)]pub struct Start { entity: Entity, object: apogee_task_v1::ApogeeTaskV1,}
pub fn handle_start( start: On<Start>, mut commands: Commands, tasks: Query< ( Entity, &super::Name, &super::Command, &super::Args, &super::Running, &super::Environment, &apogee_core_v1::TaskState, &Successful, Option<&super::Process>, ), With<Task>, >, loop_handle: NonSend<ecs::LoopHandle>, global_env: Res<environment::Environment>,) { let Ok((entity, name, command, args, running, environment, state, successful, process)) = tasks.get(start.entity) else { start.object.send_error( apogee_core_v1::TaskError::Unavailable, "service is no longer available", ); return; };
commands .entity(successful.0) .remove::<condition::Fulfilled>();
if process.is_some() || matches!(state, apogee_core_v1::TaskState::Running) { start.object.send_error( apogee_core_v1::TaskError::AlreadyActive, "service is running or has a pending restart", ); return; }
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) => { commands.entity(entity).insert(( process, super::Timestamp(super::unix_timestamp()), apogee_core_v1::TaskState::Running, ));
log::info!("restarted service: {}", name.0); commands.entity(running.0).insert(condition::Fulfilled); start.object.send_ok(); } Err(error) => { log::error!("failed to restart {}: {error:#}", name.0); commands .entity(entity) .insert(apogee_core_v1::TaskState::Failed); start .object .send_error(apogee_core_v1::TaskError::SpawnFailed, error.to_string()); } }}
#[derive(Event)]pub struct Restart { entity: Entity, object: apogee_task_v1::ApogeeTaskV1,}
pub fn handle_restart( restart: On<Restart>, mut commands: Commands, mut tasks: Query< ( Entity, &super::Name, &super::Command, &super::Args, &super::Conditions, &Successful, &super::Running, &super::Environment, &mut apogee_core_v1::TaskState, Option<&mut super::Process>, ), With<Task>, >, fulfilled: Query<(), With<condition::Fulfilled>>, loop_handle: NonSend<ecs::LoopHandle>, global_env: Res<environment::Environment>,) { let Ok(( entity, name, command, args, conditions, ready, running, environment, mut state, process, )) = tasks.get_mut(restart.entity) else { restart.object.send_error( apogee_core_v1::TaskError::Unavailable, "service is no longer available", ); return; };
if !conditions .0 .iter() .all(|condition| fulfilled.contains(*condition)) { restart.object.send_error( apogee_core_v1::TaskError::ConditionsUnsatisfied, "one or more required conditions are unfulfilled", ); return; }
if let Some(mut process) = process { log::info!("Stopping {}[{}], sending SIGKILL ...", name.0, process.id.0); if let Err(error) = process.child.kill() { log::debug!("failed to stop {} for restart: {error}", name.0); restart .object .send_error(apogee_core_v1::TaskError::KillFailed, error.to_string()); return; } process.remove_sources(&loop_handle);
if let Err(error) = process.child.wait() { log::warn!("failed to reap {} for restart: {error}", name.0); restart .object .send_error(apogee_core_v1::TaskError::WaitFailed, error.to_string()); return; } }
commands.entity(ready.0).remove::<condition::Fulfilled>();
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) => { *state = apogee_core_v1::TaskState::Running; log::info!("Starting {}[{}]", name.0, process.id.0); commands .entity(entity) .insert((process, super::Timestamp(super::unix_timestamp()))); commands.entity(running.0).insert(condition::Fulfilled); restart.object.send_ok(); } Err(error) => { *state = apogee_core_v1::TaskState::Failed; commands.entity(entity).remove::<super::Process>(); commands.entity(entity).insert(super::Timestamp(0)); commands.entity(ready.0).remove::<condition::Fulfilled>(); commands.entity(running.0).remove::<condition::Fulfilled>(); log::error!("failed to restart {}: {error:#}", name.0); restart .object .send_error(apogee_core_v1::TaskError::SpawnFailed, error.to_string()); } }}
#[derive(Event)]pub struct Exited { pub task: Entity, pub process: super::ProcessId, pub reason: super::StopReason,}
pub fn handle_exited( process_exited: On<Exited>, mut commands: Commands, mut tasks: Query<( &super::Name, &mut super::Process, &Successful, &super::Running, &mut apogee_core_v1::TaskState, )>,) { let Ok((name, mut process, successful, running, mut state)) = tasks.get_mut(process_exited.task) else { return; };
if process.id != process_exited.process { log::debug!( "ignoring exit notification from an old process of {}", name.0 ); return; }
commands .entity(successful.0) .remove::<condition::Fulfilled>(); commands.entity(running.0).remove::<condition::Fulfilled>(); commands .entity(process_exited.task) .remove::<super::Process>(); commands .entity(process_exited.task) .insert(super::Timestamp(0));
let status = match process.child.wait() { Ok(status) => { if process_exited.reason == super::StopReason::Exited { log::info!("{}[{}] exited with {status}", name.0, process.id.0); } status } Err(error) => { log::warn!("failed waiting for {} to exit: {error}", name.0); if process_exited.reason == super::StopReason::Exited { commands .entity(process_exited.task) .insert(apogee_core_v1::TaskState::Failed); } return; } };
match (process_exited.reason, status.success()) { (super::StopReason::Removed | super::StopReason::Restarted, _) => {} (super::StopReason::Exited, true) => { *state = apogee_core_v1::TaskState::Successful; commands.entity(successful.0).insert(condition::Fulfilled); } (super::StopReason::Exited, false) => { *state = apogee_core_v1::TaskState::Failed; } }}
pub fn removed( mut commands: Commands, mut removed_services: RemovedComponents<Task>, task_state: Query<(&super::Name, &Objects)>, mut processes: Query<&mut super::Process>, mut services: ResMut<ecs::TaskRegistry>, managers: Res<ecs::Managers>,) { for entity in removed_services.read() { if let Ok((name, objects)) = task_state.get(entity) { log::info!("task removed: {}", name.0); services.by_name.remove(&name.0); managers.task_removed(&name.0); for object in objects .0 .iter() .filter(|object| object.try_client().is_some()) { object.send_destroyed(); } }
if let Ok(mut process) = processes.get_mut(entity) { if let Ok((name, _)) = task_state.get(entity) { log::info!("Stopping {}[{}], sending SIGKILL ...", name.0, process.id.0); } if let Err(error) = process.child.kill() { log::warn!("failed to stop service {entity:?}: {error}"); }
let event = Exited { task: entity, process: process.id, reason: super::StopReason::Removed, }; commands.queue(move |world: &mut World| { world.trigger(event); world.flush(); }); }
commands.entity(entity).despawn(); }}