zng_ext_setup/
service.rs

1#![allow(clippy::result_large_err)]
2
3use core::fmt;
4use std::{any::Any, collections::VecDeque, pin::Pin, sync::Arc};
5
6use zng_app::{event::app_local, update::UPDATES};
7use zng_clone_move::clmv;
8use zng_ext_config::RawConfigValue;
9use zng_task::{Progress, parking_lot::Mutex};
10use zng_txt::Txt;
11use zng_var::{IntoVar, ResponderVar, ResponseVar, Var, VarEq, VarValue, const_var, response_var, var};
12
13use crate::task::{SetupTask, SetupTaskError, SetupTaskType, TaskTypeId};
14
15/// Setup service.
16///
17/// This service runs [install] and [uninstall] operations sequentially. Operations
18/// start once the current app update finishes and there are no other running operations.
19///
20/// [install]:  Self::install
21/// [uninstall]:  Self::uninstall
22pub struct SETUP;
23
24impl SETUP {
25    /// Register a custom task type.
26    pub fn register_task_type<T: SetupTask>(&self) {
27        self.register_task_type_impl(SetupTaskType::new::<T>());
28    }
29    fn register_task_type_impl(&self, t: SetupTaskType) {
30        UPDATES.once_update("register_task_type", move || {
31            let mut sv = SETUP_SV.write();
32            let id = (t.task_type_id)();
33            if let Some(e) = sv.task_types.iter_mut().find(|t| (t.task_type_id)() == id) {
34                *e = t;
35            } else {
36                sv.task_types.push(t);
37            }
38        });
39    }
40
41    /// Enqueue a new install operation.
42    ///
43    /// This will [prepare] and [commit] an install.
44    ///
45    /// Returns a response var that updates once with the result of the operation. If
46    /// successful an [`UninstallConfig`], this data can be (de)serialized and used with
47    /// to [`uninstall`].
48    ///
49    /// [prepare]: Self::prepare_install
50    /// [commit]: Self::commit_install
51    /// [`uninstall`]: Self::uninstall
52    pub fn install(&self, config: InstallConfig, update: Option<UninstallConfig>) -> ResponseVar<Result<UninstallConfig, SetupError>> {
53        let (r, rsp) = response_var();
54        UPDATES.once_update("install", move || {
55            SETUP_SV.write().run(async move { install(config, update).await }, r);
56        });
57        rsp
58    }
59
60    /// Enqueue a new prepare install operation.
61    ///
62    /// This will run all expensive install operations that can run without affecting the system or previous installs.
63    /// One reason to run this step separate is to begin an update installation while the application is still running.
64    ///
65    /// If `update` is set the tasks will use it to find and patch/replace a previous install.
66    ///
67    /// Returns a response var that updates once with the result of the operation. If
68    /// successful a [`PreparedInstallConfig`], this data can be (de)serialized and used with
69    /// to [`commit_install`] or [`cancel_prepared`].
70    ///
71    /// [`commit_install`]: Self::commit_install
72    /// [`cancel_prepared`]: Self::cancel_prepared
73    pub fn prepare_install(
74        &self,
75        config: InstallConfig,
76        update: Option<UninstallConfig>,
77    ) -> ResponseVar<Result<PreparedInstallConfig, SetupError>> {
78        let (r, rsp) = response_var();
79        UPDATES.once_update("prepare_install", move || {
80            SETUP_SV.write().run(async move { prepare_install(config, update).await }, r);
81        });
82        rsp
83    }
84
85    /// Enqueue a prepared install cancellation operation.
86    ///
87    /// Note that [`prepare_install`] will automatically cancel if requested. This method cancels
88    /// a prepared install that already completed.
89    ///
90    /// [`prepare_install`]: Self::prepare_install
91    pub fn cancel_prepared(&self, config: PreparedInstallConfig) -> ResponseVar<Result<(), SetupError>> {
92        let (r, rsp) = response_var();
93        UPDATES.once_update("cancel_prepared", move || {
94            SETUP_SV.write().run(async move { cancel_prepared(config).await }, r);
95        });
96        rsp
97    }
98
99    /// Enqueue a new commit prepared install operation.
100    ///
101    /// This operation cannot be canceled once it starts, if cancel is requested while enqueued
102    /// the [`cancel_prepared`] operation will run instead.
103    ///
104    /// Returns a response var that updates once with the result of the operation. If
105    /// successful an [`UninstallConfig`], this data can be (de)serialized and used with
106    /// to [`uninstall`].
107    ///
108    /// [`cancel_prepared`]: Self::cancel_prepared
109    /// [`uninstall`]: Self::uninstall
110    pub fn commit_install(&self, config: PreparedInstallConfig) -> ResponseVar<Result<UninstallConfig, SetupError>> {
111        let (r, rsp) = response_var();
112        UPDATES.once_update("commit_install", move || {
113            SETUP_SV.write().run(async move { commit_install(config).await }, r);
114        });
115        rsp
116    }
117
118    /// Enqueue a new uninstall operation.
119    ///
120    /// This will quickly [validate] the installation and uninstall. During uninstall
121    /// the operation cannot be canceled.
122    ///
123    /// Returns a response var that updates once with the result of the operation.
124    ///
125    /// [validate]: Self::validate_uninstall
126    pub fn uninstall(&self, config: UninstallConfig) -> ResponseVar<Result<(), SetupError>> {
127        let (r, rsp) = response_var();
128        UPDATES.once_update("uninstall", move || {
129            SETUP_SV.write().run(async move { uninstall(config).await }, r);
130        });
131        rsp
132    }
133
134    /// Enqueue a new validate uninstall config operation.
135    ///
136    /// This will verify that the uninstall config can still be used to [`uninstall`].
137    ///
138    /// Returns a response var that updates once with the result of the operation. If
139    /// successful the config data is returned, potentially corrected if recoverable issues where found.
140    ///
141    /// [`uninstall`]: Self::uninstall
142    pub fn validate_uninstall(&self, config: UninstallConfig) -> ResponseVar<Result<UninstallConfig, SetupError>> {
143        let (r, rsp) = response_var();
144        UPDATES.once_update("validate_uninstall", move || {
145            SETUP_SV.write().run(async move { validate_uninstall(config).await }, r);
146        });
147        rsp
148    }
149
150    /// Status of running operation.
151    pub fn status(&self) -> Var<SetupStatus> {
152        SETUP_SV.read().status.read_only()
153    }
154}
155
156type SetupOp = Pin<Box<dyn Future<Output = ()> + Send>>;
157
158struct Setup {
159    task_types: Vec<SetupTaskType>,
160    queue: Mutex<VecDeque<SetupOp>>, // Mutex for +Sync only
161    status: Var<SetupStatus>,
162    cancel: Var<bool>,
163}
164app_local! {
165    static SETUP_SV: Setup = Setup {
166        task_types: vec![
167            SetupTaskType::new::<crate::task::ExtractTar>(),
168            #[cfg(any(windows, target_os = "linux"))]
169            SetupTaskType::new::<crate::task::CreateShortcut>(),
170            #[cfg(windows)]
171            SetupTaskType::new::<crate::task::RegisterUninstaller>(),
172        ],
173        queue: Mutex::default(),
174        status: var(SetupStatus::Idle),
175        cancel: var(false),
176    };
177}
178
179/// Represents a list of tasks for a [`SETUP.install`] operation.
180///
181/// [`SETUP.install`]: SETUP::install
182#[derive(Default)]
183pub struct InstallConfig {
184    cfg: Vec<(SetupTaskType, Box<dyn Any + Send>)>,
185    tasks: Vec<(TaskTypeId, Txt)>,
186}
187impl InstallConfig {
188    /// New empty.
189    pub fn new() -> Self {
190        Self::default()
191    }
192
193    /// Push a task to run.
194    ///
195    /// The `name` is used to identify the task instance in progress status.
196    pub fn push<T: SetupTask>(&mut self, name: impl Into<Txt>, config: T::InstallConfig) {
197        let t = SetupTaskType::new::<T>();
198        self.tasks.push(((t.task_type_id)(), name.into()));
199        self.cfg.push((t, Box::new(config)))
200    }
201
202    /// Task types and names in order they will execute.
203    pub fn tasks(&self) -> &[(TaskTypeId, Txt)] {
204        &self.tasks
205    }
206
207    /// Inspect the task config.
208    pub fn config<T: SetupTask>(&self, index: usize) -> Option<&T::InstallConfig> {
209        self.cfg.get(index)?.1.downcast_ref()
210    }
211}
212
213/// Represents data generated  by [`SETUP.prepare_install`] that can be used to run a [`SETUP.commit_install`] operation.
214///
215/// [`SETUP.prepare_install`]: SETUP::prepare_install
216/// [`SETUP.commit_install`]: SETUP::commit_install
217#[derive(Clone, PartialEq, Debug, serde::Serialize, serde::Deserialize)]
218pub struct PreparedInstallConfig {
219    tasks: Vec<(TaskTypeId, Txt)>,
220    cfg: Vec<RawConfigValue>,
221}
222impl PreparedInstallConfig {
223    /// Task types and names in order they will execute.
224    pub fn tasks(&self) -> &[(TaskTypeId, Txt)] {
225        &self.tasks
226    }
227}
228
229/// Represents data generated by [`SETUP.install`] that can be used to run a [`SETUP.uninstall`] operation.
230///
231/// [`SETUP.install`]: SETUP::install
232/// [`SETUP.uninstall`]: SETUP::uninstall
233#[derive(Clone, PartialEq, Debug, serde::Serialize, serde::Deserialize)]
234pub struct UninstallConfig {
235    tasks: Vec<(TaskTypeId, Txt)>,
236    cfg: Vec<RawConfigValue>,
237}
238impl UninstallConfig {
239    /// Task types and names in order they will execute.
240    pub fn tasks(&self) -> &[(TaskTypeId, Txt)] {
241        &self.tasks
242    }
243}
244
245#[cfg(feature = "save")]
246macro_rules! impl_save {
247    ($($Config:ident),+) => {$(
248
249#[cfg(feature = "save")]
250impl $Config {
251    /// Serialize and write config to file.
252    ///
253    /// The format is a ZStandard compressed JSON.
254    pub fn save_blocking(&self, file: &std::path::Path) -> std::io::Result<()> {
255        save(self, file)
256    }
257
258    /// Serialize and write config to file.
259    ///
260    /// The format is a ZStandard compressed JSON.
261    pub async fn save(self, file: std::path::PathBuf) -> std::io::Result<()> {
262        zng_task::wait(move || save(&self, &file)).await
263    }
264
265    /// Read and deserialize config from file.
266    ///
267    /// The format must be a ZStandard compressed JSON.
268    pub fn load_blocking(file: &std::path::Path) -> std::io::Result<Self> {
269        load(file)
270    }
271
272    /// Read and deserialize config from file.
273    ///
274    /// The format must be a ZStandard compressed JSON.
275    pub async fn load(file: std::path::PathBuf) -> std::io::Result<Self> {
276        zng_task::wait(move || load(&file)).await
277    }
278}
279    )+};
280}
281#[cfg(feature = "save")]
282impl_save! { UninstallConfig, PreparedInstallConfig }
283
284#[cfg(feature = "save")]
285fn save(config: &impl serde::Serialize, file: &std::path::Path) -> std::io::Result<()> {
286    let file = std::fs::File::create(file)?;
287    let mut zstd = zstd::Encoder::new(file, 22)?;
288    serde_json::to_writer(&mut zstd, config)?;
289    zstd.finish()?;
290    Ok(())
291}
292
293#[cfg(feature = "save")]
294fn load<T: serde::de::DeserializeOwned>(file: &std::path::Path) -> std::io::Result<T> {
295    let file = std::fs::File::open(file)?;
296    // optimal BufReader created internally by decoder
297    let zstd = zstd::Decoder::new(file)?;
298    let cfg = serde_json::from_reader(zstd)?;
299    Ok(cfg)
300}
301
302/// Represents a [`SETUP`] operation error.
303#[derive(Clone, PartialEq, Debug)]
304#[non_exhaustive]
305pub struct SetupError {
306    /// Error associated with operation itself that affects all tasks.
307    ///
308    /// This is often [`SetupTaskError::CorruptedTaskData`] detected
309    pub op_error: Option<SetupTaskError>,
310
311    /// Errors associated with a task in the operation.
312    ///
313    /// Each task is identified by index on the operation, type and name.
314    pub task_errors: Vec<((usize, TaskTypeId, Txt), SetupTaskError)>,
315
316    /// Operation state after not completing successfully.
317    pub state: SetupErrorState,
318}
319impl SetupError {
320    /// No actual error, canceled by request.
321    pub fn canceled() -> Self {
322        Self {
323            op_error: None,
324            task_errors: vec![],
325            state: SetupErrorState::Canceled,
326        }
327    }
328
329    /// One or more tasks failed.
330    ///
331    /// Each task is identified by index on the operation, type and name.
332    ///
333    /// If the tasks managed to reverse all changes before committing the `state` must be `Canceled`.
334    pub fn task_errors(errors: Vec<((usize, TaskTypeId, Txt), SetupTaskError)>, state: SetupErrorState) -> Self {
335        Self {
336            op_error: None,
337            task_errors: errors,
338            state,
339        }
340    }
341
342    /// Operation failed with error associated with operation itself that affects all tasks.
343    ///
344    /// If the error is detected before any task runs and any non-destructive change was made the `state` must be `Canceled`.
345    pub fn op_error(error: SetupTaskError, state: SetupErrorState) -> Self {
346        Self {
347            op_error: Some(error),
348            task_errors: vec![],
349            state,
350        }
351    }
352
353    /// Operation cannot start due to corrupted config data.
354    pub fn corrupted_op_config(config_name: &'static str) -> Self {
355        #[derive(Debug)]
356        struct CorruptedOpConfig(&'static str);
357        impl fmt::Display for CorruptedOpConfig {
358            fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
359                write!(f, "{} is corrupted", self.0)
360            }
361        }
362        impl std::error::Error for CorruptedOpConfig {}
363        Self::op_error(
364            SetupTaskError::CorruptedTaskData(Arc::new(CorruptedOpConfig(config_name))),
365            SetupErrorState::Canceled,
366        )
367    }
368}
369
370/// Represents state of a setup operation that ended in a [`SetupError`].
371#[derive(Clone, PartialEq, Debug)]
372pub enum SetupErrorState {
373    /// Canceled successfully, all changes where reverted.
374    ///
375    /// The operation is canceled by request or by error before it starts committing irreversible changes,
376    /// either way the system and any previous installation is not affected when ended in this state.
377    Canceled,
378
379    /// Install operation failed before it started making irreversible changes and failed to cleanup
380    /// temporary changes.
381    ///
382    /// When an install fails during the preparing phase it automatically attempts to *cancel*, this is
383    /// the error when that cancel fails. If the install was an update the previous version will still be valid.
384    PartialPrepareInstall,
385
386    /// Install operation failed while making irreversible changes to the system.
387    ///
388    /// Operation attempts to complete as much of the install as possible, the associated `data`
389    /// can be used to uninstall the committed changes. Well designed tasks will cleanup all temporary
390    /// *prepared* data on error and generate uninstall data that cleanups even partial written files, but
391    /// there is no guarantee that this data will fully uninstall every change.
392    ///
393    /// If the operation was replacing a previous install (update or repair) the `data` will also
394    /// uninstall the previous installation.
395    PartialInstall {
396        /// Data that can uninstall all the successfully committed changes made during the failed install.
397        data: UninstallConfig,
398        /// Tasks that failed without generating uninstall data.
399        ///
400        /// If this is not empty uninstalling `data` will not fully cleanup the broken install.
401        ///
402        /// Tasks are identified by index on the operation, type and name.
403        no_data: Vec<(usize, TaskTypeId, Txt)>,
404    },
405    /// Uninstall operation failed while making irreversible changes to the system.
406    ///
407    /// Operation attempts to complete as much of the uninstall as possible, so all tasks without error
408    /// have completed successfully.
409    ///
410    ///
411    PartialUninstall,
412}
413
414/// Represents status of [`SETUP`].
415#[derive(Debug, PartialEq, Clone)]
416#[non_exhaustive]
417pub enum SetupStatus {
418    /// No setup operation is running.
419    Idle,
420    /// Prepare install is running or complete.
421    PrepareInstall(SetupOpStatus),
422    /// Commit install is running or complete.
423    CommitInstall(SetupOpStatus),
424    /// Validate uninstall is running or complete.
425    ValidateUninstall(SetupOpStatus),
426    /// Uninstall is running or complete.
427    Uninstall(SetupOpStatus),
428}
429impl SetupStatus {
430    /// If is `Idle` or op is complete.
431    pub fn is_idle(&self) -> bool {
432        match self {
433            SetupStatus::Idle => true,
434            SetupStatus::PrepareInstall(s)
435            | SetupStatus::CommitInstall(s)
436            | SetupStatus::ValidateUninstall(s)
437            | SetupStatus::Uninstall(s) => s.is_complete(),
438        }
439    }
440
441    /// Get operation status.
442    pub fn op_status(&self) -> Option<&SetupOpStatus> {
443        match self {
444            SetupStatus::Idle => None,
445            SetupStatus::PrepareInstall(s)
446            | SetupStatus::CommitInstall(s)
447            | SetupStatus::ValidateUninstall(s)
448            | SetupStatus::Uninstall(s) => Some(s),
449        }
450    }
451}
452
453/// Represents status of a [`SETUP`] install or uninstall operation.
454#[derive(Debug, PartialEq, Clone)]
455#[non_exhaustive]
456pub struct SetupOpStatus {
457    /// Is cancelling.
458    pub cancel: bool,
459    /// Current task.
460    pub task: (TaskTypeId, Txt),
461    /// Task index of len.
462    pub progress: (usize, usize),
463    /// Progress report from task.
464    pub task_progress: VarEq<Progress>,
465
466    /// Errors.
467    ///
468    /// The task is identified by index, type and name.
469    pub errors: Vec<((usize, TaskTypeId, Txt), SetupTaskError)>,
470}
471impl SetupOpStatus {
472    /// New custom  status.
473    pub fn new(
474        cancel: bool,
475        task: (TaskTypeId, Txt),
476        progress: (usize, usize),
477        task_progress: impl IntoVar<Progress>,
478        errors: Vec<((usize, TaskTypeId, Txt), SetupTaskError)>,
479    ) -> Self {
480        Self {
481            cancel,
482            task,
483            progress,
484            task_progress: VarEq(task_progress.into_var()),
485            errors,
486        }
487    }
488
489    /// If `progress` is last task and `task_progress` is complete.
490    ///
491    /// Note that tasks report completion on error.
492    pub fn is_complete(&self) -> bool {
493        self.progress.0 == self.progress.1.saturating_sub(1) && self.task_progress.with(|p| p.is_complete())
494    }
495
496    /// If `cancel` and `is_complete`.
497    ///
498    /// If this is `true` and `errors` is not empty the tasks managed to cleanup and the install is not corrupted.
499    pub fn is_canceled(&self) -> bool {
500        self.cancel && self.is_complete()
501    }
502
503    /// If `is_complete`, not `cancel` and has `errors`.
504    ///
505    /// If this is `true` the tasks did not manage to cleanup and the install is in a corrupted state.
506    pub fn is_corrupted(&self) -> bool {
507        !self.errors.is_empty() && !self.cancel && self.is_complete()
508    }
509}
510
511impl Setup {
512    fn run<R: VarValue>(
513        &mut self,
514        op: impl Future<Output = Result<R, SetupError>> + Send + 'static,
515        r: ResponderVar<Result<R, SetupError>>,
516    ) {
517        self.run_impl(Box::pin(async move {
518            let res = op.await;
519            r.respond(res);
520        }));
521    }
522    fn run_impl(&mut self, op: SetupOp) {
523        let q = self.queue.get_mut();
524        q.push_back(op);
525        if q.len() == 1 {
526            zng_task::spawn(async {
527                fn next_op() -> Option<SetupOp> {
528                    let mut sv = SETUP_SV.write();
529                    let op = sv.queue.get_mut().pop_front();
530                    if op.is_some() {
531                        // ensure op will not retain errors from previous op
532                        sv.status.set(SetupStatus::Idle);
533                    }
534                    op
535                }
536                while let Some(op) = next_op() {
537                    op.await;
538                }
539            });
540        }
541    }
542
543    fn task_type(&self, id: &TaskTypeId) -> Result<SetupTaskType, SetupTaskError> {
544        for t in &self.task_types {
545            if &(t.task_type_id)() == id {
546                return Ok(t.clone());
547            }
548        }
549        Err(SetupTaskError::UnknownType(id.clone()))
550    }
551}
552
553async fn install(config: InstallConfig, update: Option<UninstallConfig>) -> Result<UninstallConfig, SetupError> {
554    let config = prepare_install(config, update).await?;
555    if SETUP_SV.read().cancel.get() {
556        cancel_prepared(config).await?;
557        Err(SetupError::canceled())
558    } else {
559        commit_install(config).await
560    }
561}
562
563async fn prepare_install(config: InstallConfig, update: Option<UninstallConfig>) -> Result<PreparedInstallConfig, SetupError> {
564    let (status, cancel) = {
565        let sv = SETUP_SV.read();
566        (sv.status.clone(), sv.cancel.clone())
567    };
568
569    if let Some(u) = &update
570        && u.tasks.len() != u.cfg.len()
571    {
572        return Err(SetupError::corrupted_op_config("UninstallConfig"));
573    }
574
575    let tasks_len = config.tasks.len();
576    if tasks_len != config.cfg.len() {
577        return Err(SetupError::corrupted_op_config("InstallConfig"));
578    }
579    let mut prepared_cfg: Vec<RawConfigValue> = Vec::with_capacity(tasks_len);
580    let mut error = None;
581    for (i, (id, (task_ty, cfg))) in config.tasks.iter().zip(config.cfg).enumerate() {
582        let task_progress = var(Progress::indeterminate());
583        // notify new task started
584        let task_progress_s = task_progress.read_only();
585        status.modify(clmv!(id, |a| {
586            match a.value_mut() {
587                SetupStatus::PrepareInstall(s) => {
588                    s.task = id;
589                    s.progress.0 = i;
590                    s.task_progress = VarEq(task_progress_s);
591                }
592                _ => {
593                    **a = SetupStatus::PrepareInstall(SetupOpStatus {
594                        cancel: false,
595                        task: id,
596                        progress: (i, tasks_len),
597                        task_progress: VarEq(task_progress_s),
598                        errors: vec![],
599                    });
600                }
601            }
602        }));
603
604        // find previous install
605        let mut uninstall_data = None;
606        if let Some(u) = &update {
607            // uninstall is reversed
608            if let Some(i) = u.cfg.len().checked_sub(i + 1)
609                && id == &u.tasks[i]
610            {
611                uninstall_data = Some(u.cfg[i].clone());
612            }
613
614            if uninstall_data.is_none() {
615                #[derive(Debug)]
616                struct TaskTypeMismatch;
617                impl fmt::Display for TaskTypeMismatch {
618                    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
619                        write!(f, "expected different task type")
620                    }
621                }
622                impl std::error::Error for TaskTypeMismatch {}
623                error = Some((
624                    (i, id.0.clone(), id.1.clone()),
625                    SetupTaskError::CorruptedTaskData(Arc::new(TaskTypeMismatch)),
626                ));
627                break;
628            }
629        }
630
631        // run task
632        let r = (task_ty.prepare_install)(cfg, uninstall_data, task_progress.clone(), cancel.read_only()).await;
633        task_progress.set(Progress::complete());
634        match r {
635            Ok(r) => prepared_cfg.push(r),
636            Err(e) => {
637                // go to cancel due to error
638                error = Some(((i, id.0.clone(), id.1.clone()), e));
639                break;
640            }
641        }
642        if cancel.get() {
643            break;
644        }
645    }
646
647    let prepared_cfg = PreparedInstallConfig {
648        tasks: config.tasks,
649        cfg: prepared_cfg,
650    };
651
652    if let Some(e) = error {
653        status.modify(clmv!(e, |a| {
654            // add error, cancel_prepared will preserve it as it updates status
655            if let SetupStatus::PrepareInstall(s) = a.value_mut() {
656                s.errors.push(e);
657                // immediately indicate cancelling too, to avoid notifying completion
658                s.task_progress = VarEq(const_var(Progress::indeterminate()));
659                s.cancel = true;
660            }
661        }));
662
663        // cancel due to error
664        if let Err(mut ce) = cancel_prepared(prepared_cfg).await {
665            // did not cleanup prepared either.
666            ce.state = SetupErrorState::PartialPrepareInstall;
667            ce.task_errors.insert(0, e);
668            Err(ce)
669        } else {
670            Err(SetupError::task_errors(vec![e], SetupErrorState::Canceled))
671        }
672    } else if cancel.get() {
673        // cancel due to request
674        cancel_prepared(prepared_cfg).await?;
675        Err(SetupError::canceled())
676    } else {
677        // ensure general status updates to complete
678        status.modify(clmv!(|a| {
679            if let SetupStatus::PrepareInstall(s) = a.value_mut() {
680                s.task_progress = VarEq(const_var(Progress::complete()));
681            }
682        }));
683        Ok(prepared_cfg)
684    }
685}
686
687async fn cancel_prepared(config: PreparedInstallConfig) -> Result<(), SetupError> {
688    let status = SETUP_SV.read().status.clone();
689
690    let tasks_len = config.cfg.len();
691    if tasks_len > config.tasks.len() {
692        return Err(SetupError::corrupted_op_config("PreparedInstallConfig"));
693    }
694
695    let mut errors = vec![];
696
697    for (i, (id, cfg)) in config.tasks.into_iter().zip(config.cfg).enumerate() {
698        let task_progress = var(Progress::indeterminate());
699        // notify new task started
700        let task_progress_s = task_progress.read_only();
701        status.modify(clmv!(id, |a| {
702            match a.value_mut() {
703                SetupStatus::PrepareInstall(s) => {
704                    s.cancel = true;
705                    s.task = id;
706                    s.progress = (i, tasks_len);
707                    s.task_progress = VarEq(task_progress_s);
708                }
709                _ => {
710                    **a = SetupStatus::PrepareInstall(SetupOpStatus {
711                        cancel: true,
712                        task: id,
713                        progress: (i, tasks_len),
714                        task_progress: VarEq(task_progress_s),
715                        errors: vec![],
716                    });
717                }
718            }
719        }));
720
721        // run task
722        let task_ty = SETUP_SV.read().task_type(&id.0);
723        let error = match task_ty {
724            Ok(task_ty) => {
725                let r = (task_ty.cancel_install)(cfg, task_progress.clone()).await;
726                r.err()
727            }
728            Err(e) => Some(e),
729        };
730
731        if let Some(e) = error {
732            let e = ((i, id.0, id.1), e);
733            // notify error
734            status.modify(clmv!(e, |a| {
735                if let SetupStatus::PrepareInstall(s) = a.value_mut() {
736                    s.errors.push(e);
737                    s.cancel = false;
738                    s.task_progress = VarEq(const_var(Progress::complete()));
739                }
740            }));
741            errors.push(e);
742
743            // continues trying to cancel other tasks for best effort cleanup
744        } else {
745            task_progress.set(Progress::complete());
746        }
747    }
748
749    if errors.is_empty() {
750        Ok(())
751    } else {
752        Err(SetupError::task_errors(errors, SetupErrorState::PartialPrepareInstall))
753    }
754}
755
756async fn commit_install(config: PreparedInstallConfig) -> Result<UninstallConfig, SetupError> {
757    let status = SETUP_SV.read().status.clone();
758
759    let tasks_len = config.tasks.len();
760    if tasks_len != config.cfg.len() {
761        return Err(SetupError::corrupted_op_config("PreparedInstallConfig"));
762    }
763
764    let mut errors = vec![];
765    let mut uninstall_cfg = vec![];
766    let mut err_no_clean = vec![];
767
768    for (i, (id, cfg)) in config.tasks.iter().zip(config.cfg).enumerate() {
769        let task_progress = var(Progress::indeterminate());
770        // notify new task started
771        let task_progress_s = task_progress.read_only();
772        status.modify(clmv!(id, |a| {
773            match a.value_mut() {
774                SetupStatus::CommitInstall(s) => {
775                    s.task = id;
776                    s.progress.0 = i;
777                    s.task_progress = VarEq(task_progress_s);
778                }
779                _ => {
780                    **a = SetupStatus::CommitInstall(SetupOpStatus {
781                        cancel: false,
782                        task: id,
783                        progress: (i, tasks_len),
784                        task_progress: VarEq(task_progress_s),
785                        errors: vec![],
786                    })
787                }
788            }
789        }));
790
791        // run task
792        let task_ty = SETUP_SV.read().task_type(&id.0);
793        let mut error = None;
794        match task_ty {
795            Ok(task_ty) => match (task_ty.install)(cfg, task_progress.clone()).await {
796                Ok(c) => {
797                    uninstall_cfg.push(c);
798                }
799                Err(e) => {
800                    error = Some(e.error);
801                    if let Some(d) = e.clean_data {
802                        uninstall_cfg.push(d);
803                    } else {
804                        err_no_clean.push((i, id.0.clone(), id.1.clone()));
805                    }
806                }
807            },
808            Err(e) => error = Some(e),
809        };
810
811        if let Some(e) = error {
812            let e = ((i, id.0.clone(), id.1.clone()), e);
813            // notify error
814            status.modify(clmv!(e, |a| {
815                if let SetupStatus::CommitInstall(s) = a.value_mut() {
816                    s.errors.push(e);
817                    s.cancel = false;
818                    s.task_progress = VarEq(const_var(Progress::complete()));
819                }
820            }));
821            errors.push(e);
822
823            // continues trying to commit other tasks, since cannot recover at this
824            // point might as well try to deliver a partial install
825        } else {
826            task_progress.set(Progress::complete());
827        }
828    }
829
830    let mut tasks = config.tasks;
831    tasks.reverse();
832    uninstall_cfg.reverse();
833    let data = UninstallConfig { tasks, cfg: uninstall_cfg };
834
835    if errors.is_empty() {
836        // ensure general status updates to complete
837        status.modify(clmv!(|a| {
838            if let SetupStatus::CommitInstall(s) = a.value_mut() {
839                s.task_progress = VarEq(const_var(Progress::complete()));
840            }
841        }));
842
843        Ok(data)
844    } else {
845        Err(SetupError::task_errors(
846            errors,
847            SetupErrorState::PartialInstall {
848                data,
849                no_data: err_no_clean,
850            },
851        ))
852    }
853}
854
855async fn uninstall(config: UninstallConfig) -> Result<(), SetupError> {
856    let config = validate_uninstall(config).await?;
857
858    let status = SETUP_SV.read().status.clone();
859
860    let tasks_len = config.tasks.len();
861    if tasks_len != config.cfg.len() {
862        return Err(SetupError::corrupted_op_config("UninstallConfig"));
863    }
864
865    let mut errors = vec![];
866
867    for (i, (id, cfg)) in config.tasks.into_iter().zip(config.cfg).enumerate() {
868        let task_progress = var(Progress::indeterminate());
869        // notify new task started
870        let task_progress_s = task_progress.read_only();
871        status.modify(clmv!(id, |a| {
872            match a.value_mut() {
873                SetupStatus::Uninstall(s) => {
874                    s.task = id;
875                    s.progress.0 = i;
876                    s.task_progress = VarEq(task_progress_s);
877                }
878                _ => {
879                    **a = SetupStatus::Uninstall(SetupOpStatus {
880                        cancel: false,
881                        task: id,
882                        progress: (i, tasks_len),
883                        task_progress: VarEq(task_progress_s),
884                        errors: vec![],
885                    })
886                }
887            }
888        }));
889
890        // run task
891        let task_ty = SETUP_SV.read().task_type(&id.0);
892        let error = match task_ty {
893            Ok(task_ty) => (task_ty.uninstall)(cfg, task_progress.clone()).await.err(),
894            Err(e) => Some(e),
895        };
896
897        if let Some(e) = error {
898            let e = ((i, id.0.clone(), id.1.clone()), e);
899            // notify error
900            status.modify(clmv!(e, |a| {
901                if let SetupStatus::Uninstall(s) = a.value_mut() {
902                    s.errors.push(e);
903                    s.cancel = false;
904                    s.task_progress = VarEq(const_var(Progress::complete()));
905                }
906            }));
907            errors.push(e);
908
909            // continues trying to commit other tasks, since cannot recover at this
910            // point might as well try to deliver a partial install
911        } else {
912            task_progress.set(Progress::complete());
913        }
914    }
915    if errors.is_empty() {
916        // ensure general status updates to complete
917        status.modify(clmv!(|a| {
918            if let SetupStatus::Uninstall(s) = a.value_mut() {
919                s.task_progress = VarEq(const_var(Progress::complete()));
920            }
921        }));
922        Ok(())
923    } else {
924        Err(SetupError::task_errors(errors, SetupErrorState::PartialUninstall))
925    }
926}
927
928async fn validate_uninstall(config: UninstallConfig) -> Result<UninstallConfig, SetupError> {
929    let (status, cancel) = {
930        let sv = SETUP_SV.read();
931        (sv.status.clone(), sv.cancel.clone())
932    };
933
934    let tasks_len = config.tasks.len();
935    if tasks_len != config.cfg.len() {
936        return Err(SetupError::corrupted_op_config("UninstallConfig"));
937    }
938
939    let UninstallConfig { tasks, mut cfg } = config;
940    let empty_cfg = RawConfigValue::serialize(()).unwrap();
941
942    let mut errors = vec![];
943
944    for (i, (id, cfg)) in tasks.iter().zip(cfg.iter_mut()).enumerate() {
945        let task_progress = var(Progress::indeterminate());
946        // notify new task started
947        let task_progress_s = task_progress.read_only();
948        status.modify(clmv!(id, |a| {
949            match a.value_mut() {
950                SetupStatus::ValidateUninstall(s) => {
951                    s.task = id;
952                    s.progress.0 = i;
953                    s.task_progress = VarEq(task_progress_s);
954                }
955                _ => {
956                    **a = SetupStatus::ValidateUninstall(SetupOpStatus {
957                        cancel: false,
958                        task: id,
959                        progress: (i, tasks_len),
960                        task_progress: VarEq(task_progress_s),
961                        errors: vec![],
962                    })
963                }
964            }
965        }));
966
967        // run task
968        let task_ty = SETUP_SV.read().task_type(&id.0);
969        let error = match task_ty {
970            Ok(task_ty) => {
971                match (task_ty.validate_uninstall)(std::mem::replace(cfg, empty_cfg.clone()), task_progress.clone(), cancel.clone()).await {
972                    Ok(c) => {
973                        *cfg = c;
974                        None
975                    }
976                    Err(e) => Some(e),
977                }
978            }
979            Err(e) => Some(e),
980        };
981
982        if let Some(e) = error {
983            let e = ((i, id.0.clone(), id.1.clone()), e);
984            // notify error
985            status.modify(clmv!(e, |a| {
986                if let SetupStatus::ValidateUninstall(s) = a.value_mut() {
987                    s.errors.push(e);
988                    s.cancel = false;
989                    s.task_progress = VarEq(const_var(Progress::complete()));
990                }
991            }));
992            errors.push(e);
993
994            // continues trying to commit other tasks, since cannot recover at this
995            // point might as well try to deliver a partial install
996        } else {
997            task_progress.set(Progress::complete());
998        }
999
1000        if cancel.get() {
1001            break;
1002        }
1003    }
1004
1005    let canceled = cancel.get();
1006    if errors.is_empty() && !canceled {
1007        // ensure general status updates to complete
1008        status.modify(clmv!(|a| {
1009            if let SetupStatus::ValidateUninstall(s) = a.value_mut() {
1010                s.task_progress = VarEq(const_var(Progress::complete()));
1011            }
1012        }));
1013        Ok(UninstallConfig { tasks, cfg })
1014    } else {
1015        Err(SetupError::task_errors(errors, SetupErrorState::Canceled))
1016    }
1017}
zng_ext_setup/service.rs

zng_ext_setup/
service.rs
