Feature/backup+restore (#2613)

* feat: Implementation on the backup for the service.

* wip: Getting the flow of backup/restore

* feat: Recover

* Feature: Commit the full pass on the backup restore.

* use special type for backup instead of special id (#2614)

* fix: Allow compat docker style to run again

* fix: Backup for the js side

* chore: Update some of the callbacks

---------

Co-authored-by: Aiden McClelland <3732071+dr-bonez@users.noreply.github.com>

core/startos/src/service/mod.rs

@@ -10,7 +10,7 @@ use persistent_container::PersistentContainer;
 use rpc_toolkit::{from_fn_async, CallRemoteHandler, Empty, HandlerArgs, HandlerFor};
 use serde::{Deserialize, Serialize};
 use start_stop::StartStop;
-use tokio::sync::Notify;
+use tokio::{fs::File, sync::Notify};
 use ts_rs::TS;
 
 use crate::context::{CliContext, RpcContext};
@@ -296,13 +296,20 @@ impl Service {
     }
 
     pub async fn restore(
-        _ctx: RpcContext,
-        _s9pk: S9pk,
-        _guard: impl GenericMountGuard,
-        _progress: Option<InstallProgressHandles>,
+        ctx: RpcContext,
+        s9pk: S9pk,
+        backup_source: impl GenericMountGuard,
+        progress: Option<InstallProgressHandles>,
     ) -> Result<Self, Error> {
-        // TODO
-        Err(Error::new(eyre!("not yet implemented"), ErrorKind::Unknown))
+        let service = Service::install(ctx.clone(), s9pk, None, progress).await?;
+
+        service
+            .actor
+            .send(transition::restore::Restore {
+                path: backup_source.path().to_path_buf(),
+            })
+            .await?;
+        Ok(service)
     }
 
     pub async fn shutdown(self) -> Result<(), Error> {
@@ -348,9 +355,23 @@ impl Service {
         Ok(())
     }
 
-    pub async fn backup(&self, _guard: impl GenericMountGuard) -> Result<BackupReturn, Error> {
-        // TODO
-        Err(Error::new(eyre!("not yet implemented"), ErrorKind::Unknown))
+    #[instrument(skip_all)]
+    pub async fn backup(&self, guard: impl GenericMountGuard) -> Result<(), Error> {
+        let id = &self.seed.id;
+        let mut file = File::create(guard.path().join(id).with_extension("s9pk")).await?;
+        self.seed
+            .persistent_container
+            .s9pk
+            .clone()
+            .serialize(&mut file, true)
+            .await?;
+        drop(file);
+        self.actor
+            .send(transition::backup::Backup {
+                path: guard.path().to_path_buf(),
+            })
+            .await?;
+        Ok(())
     }
 
     pub fn container_id(&self) -> Result<ContainerId, Error> {
@@ -425,6 +446,7 @@ impl Actor for ServiceActor {
                         kinds.running_status,
                     ) {
                         (Some(TransitionKind::Restarting), _, _) => MainStatus::Restarting,
+                        (Some(TransitionKind::Restoring), _, _) => MainStatus::Restoring,
                         (Some(TransitionKind::BackingUp), _, Some(status)) => {
                             MainStatus::BackingUp {
                                 started: Some(status.started),

core/startos/src/service/persistent_container.rs

@@ -1,5 +1,5 @@
 use std::collections::BTreeMap;
-use std::path::Path;
+use std::path::{Path, PathBuf};
 use std::sync::{Arc, Weak};
 use std::time::Duration;
 
@@ -222,6 +222,40 @@ impl PersistentContainer {
         })
     }
 
+    #[instrument(skip_all)]
+    pub async fn mount_backup(
+        &self,
+        backup_path: impl AsRef<Path>,
+        mount_type: MountType,
+    ) -> Result<MountGuard, Error> {
+        let backup_path: PathBuf = backup_path.as_ref().to_path_buf();
+        let mountpoint = self
+            .lxc_container
+            .get()
+            .ok_or_else(|| {
+                Error::new(
+                    eyre!("PersistentContainer has been destroyed"),
+                    ErrorKind::Incoherent,
+                )
+            })?
+            .rootfs_dir()
+            .join("media/startos/backup");
+        tokio::fs::create_dir_all(&mountpoint).await?;
+        Command::new("chown")
+            .arg("100000:100000")
+            .arg(mountpoint.as_os_str())
+            .invoke(ErrorKind::Filesystem)
+            .await?;
+        let bind = Bind::new(&backup_path);
+        let mount_guard = MountGuard::mount(&bind, &mountpoint, mount_type).await;
+        Command::new("chown")
+            .arg("100000:100000")
+            .arg(backup_path.as_os_str())
+            .invoke(ErrorKind::Filesystem)
+            .await?;
+        mount_guard
+    }
+
     #[instrument(skip_all)]
     pub async fn init(&self, seed: Weak<ServiceActorSeed>) -> Result<(), Error> {
         let socket_server_context = EffectContext::new(seed);

core/startos/src/service/service_effect_handler.rs

@@ -1070,11 +1070,6 @@ pub async fn create_overlayed_image(
             .rootfs_dir();
         let mountpoint = rootfs_dir.join("media/startos/overlays").join(&*guid);
         tokio::fs::create_dir_all(&mountpoint).await?;
-        Command::new("chown")
-            .arg("100000:100000")
-            .arg(&mountpoint)
-            .invoke(ErrorKind::Filesystem)
-            .await?;
         let container_mountpoint = Path::new("/").join(
             mountpoint
                 .strip_prefix(rootfs_dir)
@@ -1083,9 +1078,14 @@ pub async fn create_overlayed_image(
         tracing::info!("Mounting overlay {guid} for {image_id}");
         let guard = OverlayGuard::mount(
             &IdMapped::new(LoopDev::from(&**image), 0, 100000, 65536),
-            mountpoint,
+            &mountpoint,
         )
         .await?;
+        Command::new("chown")
+            .arg("100000:100000")
+            .arg(&mountpoint)
+            .invoke(ErrorKind::Filesystem)
+            .await?;
         tracing::info!("Mounted overlay {guid} for {image_id}");
         ctx.persistent_container
             .overlays

core/startos/src/service/start_stop.rs

@@ -15,6 +15,7 @@ impl From<MainStatus> for StartStop {
     fn from(value: MainStatus) -> Self {
         match value {
             MainStatus::Stopped => StartStop::Stop,
+            MainStatus::Restoring => StartStop::Stop,
             MainStatus::Restarting => StartStop::Start,
             MainStatus::Stopping { .. } => StartStop::Stop,
             MainStatus::Starting => StartStop::Start,

core/startos/src/service/transition/backup.rs

@@ -1 +1,94 @@
+use std::path::PathBuf;
 
+use futures::FutureExt;
+use models::ProcedureName;
+
+use super::TempDesiredRestore;
+use crate::disk::mount::filesystem::ReadWrite;
+use crate::prelude::*;
+use crate::service::config::GetConfig;
+use crate::service::dependencies::DependencyConfig;
+use crate::service::transition::{TransitionKind, TransitionState};
+use crate::service::ServiceActor;
+use crate::util::actor::background::BackgroundJobQueue;
+use crate::util::actor::{ConflictBuilder, Handler};
+use crate::util::future::RemoteCancellable;
+
+pub(in crate::service) struct Backup {
+    pub path: PathBuf,
+}
+impl Handler<Backup> for ServiceActor {
+    type Response = Result<(), Error>;
+    fn conflicts_with(_: &Backup) -> ConflictBuilder<Self> {
+        ConflictBuilder::everything()
+            .except::<GetConfig>()
+            .except::<DependencyConfig>()
+    }
+    async fn handle(&mut self, backup: Backup, jobs: &BackgroundJobQueue) -> Self::Response {
+        // So Need a handle to just a single field in the state
+        let temp: TempDesiredRestore = TempDesiredRestore::new(&self.0.persistent_container.state);
+        let mut current = self.0.persistent_container.state.subscribe();
+        let path = backup.path.clone();
+        let seed = self.0.clone();
+
+        let state = self.0.persistent_container.state.clone();
+        let transition = RemoteCancellable::new(
+            async move {
+                temp.stop();
+                current
+                    .wait_for(|s| s.running_status.is_none())
+                    .await
+                    .with_kind(ErrorKind::Unknown)?;
+
+                let backup_guard = seed
+                    .persistent_container
+                    .mount_backup(path, ReadWrite)
+                    .await?;
+                seed.persistent_container
+                    .execute(ProcedureName::CreateBackup, Value::Null, None)
+                    .await?;
+                backup_guard.unmount(true).await?;
+
+                if temp.restore().is_start() {
+                    current
+                        .wait_for(|s| s.running_status.is_some())
+                        .await
+                        .with_kind(ErrorKind::Unknown)?;
+                }
+                drop(temp);
+                state.send_modify(|s| {
+                    s.transition_state.take();
+                });
+                Ok::<_, Error>(())
+            }
+            .map(|x| {
+                if let Err(err) = dbg!(x) {
+                    tracing::debug!("{:?}", err);
+                    tracing::warn!("{}", err);
+                }
+            }),
+        );
+        let cancel_handle = transition.cancellation_handle();
+        let transition = transition.shared();
+        let job_transition = transition.clone();
+        jobs.add_job(job_transition.map(|_| ()));
+
+        let mut old = None;
+        self.0.persistent_container.state.send_modify(|s| {
+            old = std::mem::replace(
+                &mut s.transition_state,
+                Some(TransitionState {
+                    kind: TransitionKind::BackingUp,
+                    cancel_handle,
+                }),
+            )
+        });
+        if let Some(t) = old {
+            t.abort().await;
+        }
+        match transition.await {
+            None => Err(Error::new(eyre!("Backup canceled"), ErrorKind::Unknown)),
+            Some(x) => Ok(x),
+        }
+    }
+}

core/startos/src/service/transition/mod.rs

@@ -10,11 +10,13 @@ use crate::util::future::{CancellationHandle, RemoteCancellable};
 
 pub mod backup;
 pub mod restart;
+pub mod restore;
 
 #[derive(Clone, Copy, Debug, PartialEq, Eq, PartialOrd, Ord)]
 pub enum TransitionKind {
     BackingUp,
     Restarting,
+    Restoring,
 }
 
 /// Used only in the manager/mod and is used to keep track of the state of the manager during the
@@ -59,21 +61,23 @@ impl Drop for TransitionState {
 }
 
 #[derive(Debug, Clone)]
-pub struct TempDesiredState(pub(super) Arc<watch::Sender<ServiceState>>);
-impl TempDesiredState {
+pub struct TempDesiredRestore(pub(super) Arc<watch::Sender<ServiceState>>, StartStop);
+impl TempDesiredRestore {
     pub fn new(state: &Arc<watch::Sender<ServiceState>>) -> Self {
-        Self(state.clone())
+        Self(state.clone(), state.borrow().desired_state)
     }
     pub fn stop(&self) {
         self.0
             .send_modify(|s| s.temp_desired_state = Some(StartStop::Stop));
     }
-    pub fn start(&self) {
+    pub fn restore(&self) -> StartStop {
+        let restore_state = self.1;
         self.0
-            .send_modify(|s| s.temp_desired_state = Some(StartStop::Start));
+            .send_modify(|s| s.temp_desired_state = Some(restore_state));
+        restore_state
     }
 }
-impl Drop for TempDesiredState {
+impl Drop for TempDesiredRestore {
     fn drop(&mut self) {
         self.0.send_modify(|s| s.temp_desired_state = None);
     }

core/startos/src/service/transition/restart.rs

@@ -1,6 +1,6 @@
 use futures::FutureExt;
 
-use super::TempDesiredState;
+use super::TempDesiredRestore;
 use crate::prelude::*;
 use crate::service::config::GetConfig;
 use crate::service::dependencies::DependencyConfig;
@@ -20,18 +20,39 @@ impl Handler<Restart> for ServiceActor {
     }
     async fn handle(&mut self, _: Restart, jobs: &BackgroundJobQueue) -> Self::Response {
         // So Need a handle to just a single field in the state
-        let temp = TempDesiredState::new(&self.0.persistent_container.state);
+        let temp = TempDesiredRestore::new(&self.0.persistent_container.state);
         let mut current = self.0.persistent_container.state.subscribe();
-        let transition = RemoteCancellable::new(async move {
-            temp.stop();
-            current.wait_for(|s| s.running_status.is_none()).await;
-            temp.start();
-            current.wait_for(|s| s.running_status.is_some()).await;
-            drop(temp);
-        });
+        let state = self.0.persistent_container.state.clone();
+        let transition = RemoteCancellable::new(
+            async move {
+                temp.stop();
+                current
+                    .wait_for(|s| s.running_status.is_none())
+                    .await
+                    .with_kind(ErrorKind::Unknown)?;
+                if temp.restore().is_start() {
+                    current
+                        .wait_for(|s| s.running_status.is_some())
+                        .await
+                        .with_kind(ErrorKind::Unknown)?;
+                }
+                drop(temp);
+                state.send_modify(|s| {
+                    s.transition_state.take();
+                });
+                Ok::<_, Error>(())
+            }
+            .map(|x| {
+                if let Err(err) = x {
+                    tracing::debug!("{:?}", err);
+                    tracing::warn!("{}", err);
+                }
+            }),
+        );
         let cancel_handle = transition.cancellation_handle();
-        jobs.add_job(transition.map(|_| ()));
-        let notified = self.0.synchronized.notified();
+        let transition = transition.shared();
+        let job_transition = transition.clone();
+        jobs.add_job(job_transition.map(|_| ()));
 
         let mut old = None;
         self.0.persistent_container.state.send_modify(|s| {
@@ -46,10 +67,13 @@ impl Handler<Restart> for ServiceActor {
         if let Some(t) = old {
             t.abort().await;
         }
-        notified.await
+        if transition.await.is_none() {
+            tracing::warn!("Service {} has been cancelled", &self.0.id);
+        }
     }
 }
 impl Service {
+    #[instrument(skip_all)]
     pub async fn restart(&self) -> Result<(), Error> {
         self.actor.send(Restart).await
     }

core/startos/src/service/transition/restore.rs

@@ -0,0 +1,74 @@
+use std::path::PathBuf;
+
+use futures::FutureExt;
+use models::ProcedureName;
+
+use crate::disk::mount::filesystem::ReadOnly;
+use crate::prelude::*;
+use crate::service::transition::{TransitionKind, TransitionState};
+use crate::service::ServiceActor;
+use crate::util::actor::background::BackgroundJobQueue;
+use crate::util::actor::{ConflictBuilder, Handler};
+use crate::util::future::RemoteCancellable;
+
+pub(in crate::service) struct Restore {
+    pub path: PathBuf,
+}
+impl Handler<Restore> for ServiceActor {
+    type Response = Result<(), Error>;
+    fn conflicts_with(_: &Restore) -> ConflictBuilder<Self> {
+        ConflictBuilder::everything()
+    }
+    async fn handle(&mut self, restore: Restore, jobs: &BackgroundJobQueue) -> Self::Response {
+        // So Need a handle to just a single field in the state
+        let path = restore.path.clone();
+        let seed = self.0.clone();
+
+        let state = self.0.persistent_container.state.clone();
+        let transition = RemoteCancellable::new(
+            async move {
+                let backup_guard = seed
+                    .persistent_container
+                    .mount_backup(path, ReadOnly)
+                    .await?;
+                seed.persistent_container
+                    .execute(ProcedureName::RestoreBackup, Value::Null, None)
+                    .await?;
+                backup_guard.unmount(true).await?;
+
+                state.send_modify(|s| {
+                    s.transition_state.take();
+                });
+                Ok::<_, Error>(())
+            }
+            .map(|x| {
+                if let Err(err) = dbg!(x) {
+                    tracing::debug!("{:?}", err);
+                    tracing::warn!("{}", err);
+                }
+            }),
+        );
+        let cancel_handle = transition.cancellation_handle();
+        let transition = transition.shared();
+        let job_transition = transition.clone();
+        jobs.add_job(job_transition.map(|_| ()));
+
+        let mut old = None;
+        self.0.persistent_container.state.send_modify(|s| {
+            old = std::mem::replace(
+                &mut s.transition_state,
+                Some(TransitionState {
+                    kind: TransitionKind::Restoring,
+                    cancel_handle,
+                }),
+            )
+        });
+        if let Some(t) = old {
+            t.abort().await;
+        }
+        match transition.await {
+            None => Err(Error::new(eyre!("Restoring canceled"), ErrorKind::Unknown)),
+            Some(x) => Ok(x),
+        }
+    }
+}