mirror of
https://github.com/Start9Labs/start-os.git
synced 2026-03-30 04:01:58 +00:00
feature: Adding in the stopping state (#2677)
* feature: Adding in the stopping state * chore: Deal with timeout in the sigterm for main * chore: Update the timeout * Update web/projects/ui/src/app/pages/apps-routes/app-list/app-list-pkg/app-list-pkg.component.ts Co-authored-by: Aiden McClelland <3732071+dr-bonez@users.noreply.github.com> * Update web/projects/ui/src/app/pages/apps-routes/app-show/components/app-show-status/app-show-status.component.ts Co-authored-by: Aiden McClelland <3732071+dr-bonez@users.noreply.github.com> --------- Co-authored-by: Aiden McClelland <3732071+dr-bonez@users.noreply.github.com>
This commit is contained in:
Jade
@@ -10,6 +10,7 @@ use models::{HealthCheckId, PackageId, ProcedureName};
|
||||
use persistent_container::PersistentContainer;
|
||||
use rpc_toolkit::{from_fn_async, CallRemoteHandler, Empty, HandlerArgs, HandlerFor};
|
||||
use serde::{Deserialize, Serialize};
|
||||
use service_actor::ServiceActor;
|
||||
use start_stop::StartStop;
|
||||
use tokio::sync::Notify;
|
||||
use ts_rs::TS;
|
||||
@@ -45,6 +46,7 @@ mod dependencies;
|
||||
pub mod persistent_container;
|
||||
mod properties;
|
||||
mod rpc;
|
||||
mod service_actor;
|
||||
pub mod service_effect_handler;
|
||||
pub mod service_map;
|
||||
mod start_stop;
|
||||
@@ -482,124 +484,6 @@ impl ServiceActorSeed {
|
||||
});
|
||||
}
|
||||
}
|
||||
#[derive(Clone)]
|
||||
struct ServiceActor(Arc<ServiceActorSeed>);
|
||||
|
||||
impl Actor for ServiceActor {
|
||||
fn init(&mut self, jobs: &BackgroundJobQueue) {
|
||||
let seed = self.0.clone();
|
||||
jobs.add_job(async move {
|
||||
let id = seed.id.clone();
|
||||
let mut current = seed.persistent_container.state.subscribe();
|
||||
|
||||
loop {
|
||||
let kinds = current.borrow().kinds();
|
||||
|
||||
if let Err(e) = async {
|
||||
let main_status = match (
|
||||
kinds.transition_state,
|
||||
kinds.desired_state,
|
||||
kinds.running_status,
|
||||
) {
|
||||
(Some(TransitionKind::Restarting), StartStop::Stop, Some(_)) => {
|
||||
seed.persistent_container.stop().await?;
|
||||
MainStatus::Restarting
|
||||
}
|
||||
(Some(TransitionKind::Restarting), StartStop::Start, _) => {
|
||||
seed.persistent_container.start().await?;
|
||||
MainStatus::Restarting
|
||||
}
|
||||
(Some(TransitionKind::Restarting), _, _) => MainStatus::Restarting,
|
||||
(Some(TransitionKind::Restoring), _, _) => MainStatus::Restoring,
|
||||
(Some(TransitionKind::BackingUp), StartStop::Stop, Some(status)) => {
|
||||
seed.persistent_container.stop().await?;
|
||||
MainStatus::BackingUp {
|
||||
started: Some(status.started),
|
||||
health: status.health.clone(),
|
||||
}
|
||||
}
|
||||
(Some(TransitionKind::BackingUp), StartStop::Start, _) => {
|
||||
seed.persistent_container.start().await?;
|
||||
MainStatus::BackingUp {
|
||||
started: None,
|
||||
health: OrdMap::new(),
|
||||
}
|
||||
}
|
||||
(Some(TransitionKind::BackingUp), _, _) => MainStatus::BackingUp {
|
||||
started: None,
|
||||
health: OrdMap::new(),
|
||||
},
|
||||
(None, StartStop::Stop, None) => MainStatus::Stopped,
|
||||
(None, StartStop::Stop, Some(_)) => MainStatus::Stopping {
|
||||
timeout: seed.persistent_container.stop().await?.into(),
|
||||
},
|
||||
(None, StartStop::Start, Some(status)) => MainStatus::Running {
|
||||
started: status.started,
|
||||
health: status.health.clone(),
|
||||
},
|
||||
(None, StartStop::Start, None) => {
|
||||
seed.persistent_container.start().await?;
|
||||
MainStatus::Starting
|
||||
}
|
||||
};
|
||||
seed.ctx
|
||||
.db
|
||||
.mutate(|d| {
|
||||
if let Some(i) = d.as_public_mut().as_package_data_mut().as_idx_mut(&id)
|
||||
{
|
||||
let previous = i.as_status().as_main().de()?;
|
||||
let previous_health = previous.health();
|
||||
let previous_started = previous.started();
|
||||
let mut main_status = main_status;
|
||||
match &mut main_status {
|
||||
&mut MainStatus::Running { ref mut health, .. }
|
||||
| &mut MainStatus::BackingUp { ref mut health, .. } => {
|
||||
*health = previous_health.unwrap_or(health).clone();
|
||||
}
|
||||
_ => (),
|
||||
};
|
||||
match &mut main_status {
|
||||
MainStatus::Running {
|
||||
ref mut started, ..
|
||||
} => {
|
||||
*started = previous_started.unwrap_or(*started);
|
||||
}
|
||||
MainStatus::BackingUp {
|
||||
ref mut started, ..
|
||||
} => {
|
||||
*started = previous_started.map(Some).unwrap_or(*started);
|
||||
}
|
||||
_ => (),
|
||||
};
|
||||
i.as_status_mut().as_main_mut().ser(&main_status)?;
|
||||
}
|
||||
Ok(())
|
||||
})
|
||||
.await?;
|
||||
|
||||
Ok::<_, Error>(())
|
||||
}
|
||||
.await
|
||||
{
|
||||
tracing::error!("error synchronizing state of service: {e}");
|
||||
tracing::debug!("{e:?}");
|
||||
|
||||
seed.synchronized.notify_waiters();
|
||||
|
||||
tracing::error!("Retrying in {}s...", SYNC_RETRY_COOLDOWN_SECONDS);
|
||||
tokio::time::sleep(Duration::from_secs(SYNC_RETRY_COOLDOWN_SECONDS)).await;
|
||||
continue;
|
||||
}
|
||||
|
||||
seed.synchronized.notify_waiters();
|
||||
|
||||
tokio::select! {
|
||||
_ = current.changed() => (),
|
||||
}
|
||||
}
|
||||
})
|
||||
}
|
||||
}
|
||||
|
||||
#[derive(Deserialize, Serialize, Parser, TS)]
|
||||
pub struct ConnectParams {
|
||||
|
||||
@@ -441,11 +441,11 @@ impl PersistentContainer {
|
||||
}
|
||||
|
||||
#[instrument(skip_all)]
|
||||
pub async fn stop(&self) -> Result<Duration, Error> {
|
||||
pub async fn stop(&self) -> Result<(), Error> {
|
||||
let timeout: Option<crate::util::serde::Duration> = self
|
||||
.execute(Guid::new(), ProcedureName::StopMain, Value::Null, None)
|
||||
.await?;
|
||||
Ok(timeout.map(|a| *a).unwrap_or(Duration::from_secs(30)))
|
||||
Ok(())
|
||||
}
|
||||
|
||||
#[instrument(skip_all)]
|
||||
|
||||
159
core/startos/src/service/service_actor.rs
Normal file
159
core/startos/src/service/service_actor.rs
Normal file
@@ -0,0 +1,159 @@
|
||||
use std::sync::Arc;
|
||||
use std::time::Duration;
|
||||
|
||||
use imbl::OrdMap;
|
||||
use models::PackageId;
|
||||
|
||||
use super::start_stop::StartStop;
|
||||
|
||||
use crate::prelude::*;
|
||||
use crate::service::transition::TransitionKind;
|
||||
use crate::service::SYNC_RETRY_COOLDOWN_SECONDS;
|
||||
use crate::status::MainStatus;
|
||||
use crate::util::actor::background::BackgroundJobQueue;
|
||||
use crate::util::actor::Actor;
|
||||
|
||||
use super::ServiceActorSeed;
|
||||
|
||||
#[derive(Clone)]
|
||||
pub(super) struct ServiceActor(pub(super) Arc<ServiceActorSeed>);
|
||||
|
||||
enum ServiceActorLoopNext {
|
||||
Wait,
|
||||
DontWait,
|
||||
}
|
||||
|
||||
impl Actor for ServiceActor {
|
||||
fn init(&mut self, jobs: &BackgroundJobQueue) {
|
||||
let seed = self.0.clone();
|
||||
jobs.add_job(async move {
|
||||
let id = seed.id.clone();
|
||||
let mut current = seed.persistent_container.state.subscribe();
|
||||
|
||||
loop {
|
||||
match service_actor_loop(¤t, &seed, &id).await {
|
||||
ServiceActorLoopNext::Wait => tokio::select! {
|
||||
_ = current.changed() => (),
|
||||
},
|
||||
ServiceActorLoopNext::DontWait => (),
|
||||
}
|
||||
}
|
||||
})
|
||||
}
|
||||
}
|
||||
|
||||
async fn service_actor_loop(
|
||||
current: &tokio::sync::watch::Receiver<super::persistent_container::ServiceState>,
|
||||
seed: &Arc<ServiceActorSeed>,
|
||||
id: &PackageId,
|
||||
) -> ServiceActorLoopNext {
|
||||
let kinds = current.borrow().kinds();
|
||||
if let Err(e) = async {
|
||||
let main_status = match (
|
||||
kinds.transition_state,
|
||||
kinds.desired_state,
|
||||
kinds.running_status,
|
||||
) {
|
||||
(Some(TransitionKind::Restarting), StartStop::Stop, Some(_)) => {
|
||||
seed.persistent_container.stop().await?;
|
||||
MainStatus::Restarting
|
||||
}
|
||||
(Some(TransitionKind::Restarting), StartStop::Start, _) => {
|
||||
seed.persistent_container.start().await?;
|
||||
MainStatus::Restarting
|
||||
}
|
||||
(Some(TransitionKind::Restarting), _, _) => MainStatus::Restarting,
|
||||
(Some(TransitionKind::Restoring), _, _) => MainStatus::Restoring,
|
||||
(Some(TransitionKind::BackingUp), StartStop::Stop, Some(status)) => {
|
||||
seed.persistent_container.stop().await?;
|
||||
MainStatus::BackingUp {
|
||||
started: Some(status.started),
|
||||
health: status.health.clone(),
|
||||
}
|
||||
}
|
||||
(Some(TransitionKind::BackingUp), StartStop::Start, _) => {
|
||||
seed.persistent_container.start().await?;
|
||||
MainStatus::BackingUp {
|
||||
started: None,
|
||||
health: OrdMap::new(),
|
||||
}
|
||||
}
|
||||
(Some(TransitionKind::BackingUp), _, _) => MainStatus::BackingUp {
|
||||
started: None,
|
||||
health: OrdMap::new(),
|
||||
},
|
||||
(None, StartStop::Stop, None) => MainStatus::Stopped,
|
||||
(None, StartStop::Stop, Some(_)) => {
|
||||
let task_seed = seed.clone();
|
||||
seed.ctx
|
||||
.db
|
||||
.mutate(|d| {
|
||||
if let Some(i) = d.as_public_mut().as_package_data_mut().as_idx_mut(&id) {
|
||||
i.as_status_mut().as_main_mut().ser(&MainStatus::Stopping)?;
|
||||
}
|
||||
Ok(())
|
||||
})
|
||||
.await?;
|
||||
task_seed.persistent_container.stop().await?;
|
||||
MainStatus::Stopped
|
||||
}
|
||||
(None, StartStop::Start, Some(status)) => MainStatus::Running {
|
||||
started: status.started,
|
||||
health: status.health.clone(),
|
||||
},
|
||||
(None, StartStop::Start, None) => {
|
||||
seed.persistent_container.start().await?;
|
||||
MainStatus::Starting
|
||||
}
|
||||
};
|
||||
seed.ctx
|
||||
.db
|
||||
.mutate(|d| {
|
||||
if let Some(i) = d.as_public_mut().as_package_data_mut().as_idx_mut(&id) {
|
||||
let previous = i.as_status().as_main().de()?;
|
||||
let previous_health = previous.health();
|
||||
let previous_started = previous.started();
|
||||
let mut main_status = main_status;
|
||||
match &mut main_status {
|
||||
&mut MainStatus::Running { ref mut health, .. }
|
||||
| &mut MainStatus::BackingUp { ref mut health, .. } => {
|
||||
*health = previous_health.unwrap_or(health).clone();
|
||||
}
|
||||
_ => (),
|
||||
};
|
||||
match &mut main_status {
|
||||
MainStatus::Running {
|
||||
ref mut started, ..
|
||||
} => {
|
||||
*started = previous_started.unwrap_or(*started);
|
||||
}
|
||||
MainStatus::BackingUp {
|
||||
ref mut started, ..
|
||||
} => {
|
||||
*started = previous_started.map(Some).unwrap_or(*started);
|
||||
}
|
||||
_ => (),
|
||||
};
|
||||
i.as_status_mut().as_main_mut().ser(&main_status)?;
|
||||
}
|
||||
Ok(())
|
||||
})
|
||||
.await?;
|
||||
|
||||
Ok::<_, Error>(())
|
||||
}
|
||||
.await
|
||||
{
|
||||
tracing::error!("error synchronizing state of service: {e}");
|
||||
tracing::debug!("{e:?}");
|
||||
|
||||
seed.synchronized.notify_waiters();
|
||||
|
||||
tracing::error!("Retrying in {}s...", SYNC_RETRY_COOLDOWN_SECONDS);
|
||||
tokio::time::sleep(Duration::from_secs(SYNC_RETRY_COOLDOWN_SECONDS)).await;
|
||||
return ServiceActorLoopNext::DontWait;
|
||||
}
|
||||
seed.synchronized.notify_waiters();
|
||||
|
||||
ServiceActorLoopNext::Wait
|
||||
}
|
||||
@@ -1,4 +1,4 @@
|
||||
use std::collections::BTreeMap;
|
||||
use std::{collections::BTreeMap, sync::Arc};
|
||||
|
||||
use chrono::{DateTime, Utc};
|
||||
use imbl::OrdMap;
|
||||
@@ -6,8 +6,8 @@ use serde::{Deserialize, Serialize};
|
||||
use ts_rs::TS;
|
||||
|
||||
use self::health_check::HealthCheckId;
|
||||
use crate::prelude::*;
|
||||
use crate::status::health_check::HealthCheckResult;
|
||||
use crate::{prelude::*, util::GeneralGuard};
|
||||
|
||||
pub mod health_check;
|
||||
#[derive(Clone, Debug, Deserialize, Serialize, HasModel, TS)]
|
||||
@@ -26,10 +26,7 @@ pub enum MainStatus {
|
||||
Stopped,
|
||||
Restarting,
|
||||
Restoring,
|
||||
#[serde(rename_all = "camelCase")]
|
||||
Stopping {
|
||||
timeout: crate::util::serde::Duration,
|
||||
},
|
||||
Stopping,
|
||||
Starting,
|
||||
#[serde(rename_all = "camelCase")]
|
||||
Running {
|
||||
|
||||
Reference in New Issue
Block a user