1use std::ffi::OsString;
5use std::sync::atomic::{AtomicBool, Ordering};
6use std::sync::mpsc::{self, Receiver, RecvTimeoutError, Sender};
7use std::sync::{Arc, Mutex, MutexGuard, Weak};
8use std::time::{Duration, Instant};
9
10use corundum_framework_platform::package;
11use corundum_framework_update::{PackageState, PackageStatus};
12use semver::Version;
13use velopack::UpdateInfo;
14
15use crate::backend::{Applied, Backend, Velopack};
16use crate::error::Error;
17use crate::start::{self, PackageSpec};
18
19#[derive(Clone, Debug)]
21pub enum Source {
22 Github { repo: String, prerelease: bool },
25 Http(String),
27}
28
29#[derive(Clone, Debug)]
31pub enum After {
32 RestartApp,
34 RestartWith(Vec<OsString>),
36 Exit,
39}
40
41#[derive(Clone)]
45pub struct Updater {
46 shared: Arc<Shared>,
47}
48
49type Listener = Arc<dyn Fn() + Send + Sync>;
50
51struct Shared {
52 spec: &'static PackageSpec,
53 backend: Arc<dyn Backend>,
54 current: Version,
55 state: Mutex<State>,
56 listeners: Mutex<Vec<Listener>>,
57 worker: Mutex<Option<Sender<Request>>>,
60 applying: AtomicBool,
63}
64
65#[cfg(feature = "test-support")]
67struct Inert;
68
69#[cfg(feature = "test-support")]
70impl Backend for Inert {
71 fn current(&self) -> Version {
72 Version::new(1, 0, 0)
73 }
74
75 fn check(&self) -> Result<Option<UpdateInfo>, Error> {
76 Ok(None)
77 }
78
79 fn download(&self, _info: &UpdateInfo, _progress: &dyn Fn(u8)) -> Result<(), Error> {
80 Ok(())
81 }
82
83 fn pending(&self) -> Option<Version> {
84 None
85 }
86
87 fn apply(&self, _applied: Applied, _after: &After) -> Result<(), Error> {
88 Ok(())
89 }
90}
91
92struct State {
93 state: PackageState,
94 downloaded: Option<UpdateInfo>,
96}
97
98enum Request {
99 Check,
100 Every(Duration),
101}
102
103impl Updater {
104 pub fn new(spec: &'static PackageSpec, source: Source) -> Result<Updater, Error> {
106 if package::current().is_none() {
109 return Err(Error::NotPackaged);
110 }
111 Ok(Updater::with_backend(
112 spec,
113 Arc::new(Velopack::new(&source)?),
114 ))
115 }
116
117 pub(crate) fn with_backend(spec: &'static PackageSpec, backend: Arc<dyn Backend>) -> Updater {
118 Updater {
119 shared: Arc::new(Shared {
120 spec,
121 current: backend.current(),
122 backend,
123 state: Mutex::new(State {
124 state: PackageState::Idle,
125 downloaded: None,
126 }),
127 listeners: Mutex::new(Vec::new()),
128 worker: Mutex::new(None),
129 applying: AtomicBool::new(false),
130 }),
131 }
132 }
133
134 #[cfg(feature = "test-support")]
138 #[doc(hidden)]
139 pub fn for_tests(state: PackageState) -> Updater {
140 static SPEC: PackageSpec = PackageSpec {
141 app_name: "test",
142 on_installed: None,
143 on_uninstalling: None,
144 before_apply: None,
145 cli: None,
146 };
147 let updater = Updater::with_backend(&SPEC, Arc::new(Inert));
148 let downloaded = match &state {
149 PackageState::Ready { version } => Some(UpdateInfo {
150 TargetFullRelease: velopack::VelopackAsset {
151 Version: version.to_string(),
152 ..Default::default()
153 },
154 ..Default::default()
155 }),
156 _ => None,
157 };
158 *lock(&updater.shared.state) = State { state, downloaded };
159 updater
160 }
161
162 pub fn on_change(&self, f: impl Fn() + Send + Sync + 'static) {
165 lock(&self.shared.listeners).push(Arc::new(f));
166 }
167
168 pub fn status(&self) -> PackageStatus {
170 PackageStatus {
171 current: Some(self.shared.current.clone()),
172 state: lock(&self.shared.state).state.clone(),
173 }
174 }
175
176 pub fn check_now(&self) {
178 self.send(Request::Check);
179 }
180
181 pub fn run_in_background(&self, every: Duration) {
184 self.send(Request::Every(every));
185 }
186
187 pub fn apply(&self, after: After) -> Result<(), Error> {
204 let applying = self.shared.begin_apply()?;
205 let info = lock(&self.shared.state)
208 .downloaded
209 .clone()
210 .ok_or_else(|| Error::Velopack("nothing downloaded".into()))?;
211 self.shared.prepare()?;
212 self.shared
213 .backend
214 .apply(Applied::Downloaded(info), &after)?;
215 applying.keep();
216 Ok(())
217 }
218
219 pub fn apply_pending_at_start(&self) -> Result<(), Error> {
230 if self.apply_pending(start::restarted())? {
231 std::process::exit(0);
232 }
233 Ok(())
234 }
235
236 fn apply_pending(&self, restarted: bool) -> Result<bool, Error> {
239 if restarted {
240 return Ok(false);
241 }
242 match self.shared.backend.pending() {
243 Some(pending) if pending > self.shared.current => {
244 let applying = self.shared.begin_apply()?;
245 self.shared.prepare()?;
246 self.shared
247 .backend
248 .apply(Applied::Pending, &After::RestartApp)?;
249 applying.keep();
250 Ok(true)
251 }
252 _ => Ok(false),
253 }
254 }
255
256 fn send(&self, request: Request) {
258 let mut worker = lock(&self.shared.worker);
259 let sender = worker.get_or_insert_with(|| {
260 let (tx, rx) = mpsc::channel();
261 let shared = Arc::downgrade(&self.shared);
262 let backend = Arc::clone(&self.shared.backend);
263 std::thread::Builder::new()
264 .name("package-updater".into())
265 .spawn(move || work(shared, backend, rx))
266 .expect("could not start the updater's thread");
267 tx
268 });
269 let _ = sender.send(request);
271 }
272}
273
274struct ApplyGuard<'a>(Option<&'a AtomicBool>);
277
278impl ApplyGuard<'_> {
279 fn keep(mut self) {
280 self.0 = None;
281 }
282}
283
284impl Drop for ApplyGuard<'_> {
285 fn drop(&mut self) {
286 if let Some(applying) = self.0 {
287 applying.store(false, Ordering::SeqCst);
288 }
289 }
290}
291
292impl Shared {
293 fn begin_apply(&self) -> Result<ApplyGuard<'_>, Error> {
295 self.applying
296 .compare_exchange(false, true, Ordering::SeqCst, Ordering::SeqCst)
297 .map_err(|_| Error::Applying)?;
298 Ok(ApplyGuard(Some(&self.applying)))
299 }
300
301 fn prepare(&self) -> Result<(), Error> {
303 let Some(before_apply) = self.spec.before_apply else {
304 return Ok(());
305 };
306 before_apply().map_err(|e| {
307 self.set(PackageState::Failed(e.clone()));
308 Error::Prepare(e)
309 })
310 }
311
312 fn set(&self, new: PackageState) {
314 {
315 let mut state = lock(&self.state);
316 if state.state == new {
317 return;
318 }
319 state.state = new;
320 }
321 self.notify();
322 }
323
324 fn notify(&self) {
325 let listeners = lock(&self.listeners).clone();
327 for listener in listeners {
328 listener();
329 }
330 }
331
332 fn check(&self, backend: &dyn Backend) {
338 let ready = {
339 let state = lock(&self.state);
340 match (&state.state, &state.downloaded) {
341 (PackageState::Failed(_), Some(info)) => {
342 Version::parse(&info.TargetFullRelease.Version).ok()
343 }
344 _ => None,
345 }
346 };
347 if let Some(version) = ready {
348 self.set(PackageState::Ready { version });
349 }
350 let quiet = lock(&self.state).downloaded.is_some();
351 let fail = |msg: String| {
352 if quiet {
353 tracing::warn!("update check failed, keeping the downloaded update: {msg}");
354 } else {
355 self.set(PackageState::Failed(msg));
356 }
357 };
358 if !quiet {
359 self.set(PackageState::Checking);
360 }
361 let info = match backend.check() {
362 Ok(Some(info)) => info,
363 Ok(None) => {
364 if !quiet {
365 self.set(PackageState::UpToDate);
366 }
367 return;
368 }
369 Err(e) => return fail(e.to_string()),
370 };
371 let version = match Version::parse(&info.TargetFullRelease.Version) {
372 Ok(version) => version,
373 Err(e) => {
374 return fail(format!(
375 "the update's version {:?} is not a version: {e}",
376 info.TargetFullRelease.Version
377 ));
378 }
379 };
380 if !quiet {
381 self.set(PackageState::Downloading { percent: 0 });
382 }
383 let progress = |percent: u8| {
384 if !quiet {
385 self.set(PackageState::Downloading { percent });
386 }
387 };
388 match backend.download(&info, &progress) {
389 Ok(()) => {
390 {
391 let mut state = lock(&self.state);
392 state.state = PackageState::Ready { version };
393 state.downloaded = Some(info);
394 }
395 self.notify();
396 }
397 Err(e) => fail(e.to_string()),
398 }
399 }
400}
401
402fn work(shared: Weak<Shared>, backend: Arc<dyn Backend>, requests: Receiver<Request>) {
405 let mut every: Option<Duration> = None;
406 let mut next: Option<Instant> = None;
407 loop {
408 let request = match next {
409 Some(at) => match requests.recv_timeout(at.saturating_duration_since(Instant::now())) {
410 Ok(request) => request,
411 Err(RecvTimeoutError::Timeout) => Request::Check,
412 Err(RecvTimeoutError::Disconnected) => return,
413 },
414 None => match requests.recv() {
415 Ok(request) => request,
416 Err(_) => return,
417 },
418 };
419 if let Request::Every(interval) = request {
420 every = Some(interval);
421 }
422 let Some(shared) = shared.upgrade() else {
423 return;
424 };
425 shared.check(&*backend);
426 drop(shared);
427 next = every.map(|every| Instant::now() + every);
428 }
429}
430
431fn lock<T>(mutex: &Mutex<T>) -> MutexGuard<'_, T> {
434 mutex.lock().unwrap_or_else(|e| e.into_inner())
435}
436
437#[cfg(test)]
438mod tests {
439 use std::cell::RefCell;
440 use std::collections::VecDeque;
441 use std::sync::atomic::{AtomicUsize, Ordering};
442 use std::sync::{Arc, Mutex};
443 use std::time::{Duration, Instant};
444
445 use corundum_framework_update::{PackageState, PackageStatus};
446 use semver::Version;
447 use velopack::{UpdateInfo, VelopackAsset};
448
449 use super::*;
450 use crate::backend::{Applied, Backend};
451 use crate::error::Error;
452 use crate::start::PackageSpec;
453
454 thread_local! {
455 static APPLY_LOG: RefCell<Vec<String>> = const { RefCell::new(Vec::new()) };
458 }
459
460 fn log(entry: impl Into<String>) {
461 APPLY_LOG.with(|l| l.borrow_mut().push(entry.into()));
462 }
463
464 fn apply_log() -> Vec<String> {
465 APPLY_LOG.with(|l| l.borrow().clone())
466 }
467
468 fn v(s: &str) -> Version {
469 Version::parse(s).unwrap()
470 }
471
472 fn info(version: &str) -> UpdateInfo {
473 UpdateInfo {
474 TargetFullRelease: VelopackAsset {
475 Version: version.to_owned(),
476 FileName: format!("app-{version}-full.nupkg"),
477 ..Default::default()
478 },
479 ..Default::default()
480 }
481 }
482
483 fn spec(before_apply: Option<fn() -> Result<(), String>>) -> &'static PackageSpec {
484 Box::leak(Box::new(PackageSpec {
485 app_name: "updater-test",
486 on_installed: None,
487 on_uninstalling: None,
488 before_apply,
489 cli: None,
490 }))
491 }
492
493 fn stopping() -> Result<(), String> {
494 log("before_apply");
495 Ok(())
496 }
497
498 fn failing_to_stop() -> Result<(), String> {
499 log("before_apply");
500 Err("stop failed".to_owned())
501 }
502
503 struct Fake {
506 current: Version,
507 checks: Mutex<VecDeque<Result<Option<UpdateInfo>, Error>>>,
508 checked: AtomicUsize,
509 pending: Option<Version>,
510 }
511
512 impl Fake {
513 fn new(
514 current: &str,
515 checks: Vec<Result<Option<UpdateInfo>, Error>>,
516 pending: Option<&str>,
517 ) -> Arc<Fake> {
518 Arc::new(Fake {
519 current: v(current),
520 checks: Mutex::new(checks.into()),
521 checked: AtomicUsize::new(0),
522 pending: pending.map(v),
523 })
524 }
525 }
526
527 impl Backend for Fake {
528 fn current(&self) -> Version {
529 self.current.clone()
530 }
531
532 fn check(&self) -> Result<Option<UpdateInfo>, Error> {
533 self.checked.fetch_add(1, Ordering::SeqCst);
534 self.checks.lock().unwrap().pop_front().unwrap_or(Ok(None))
535 }
536
537 fn download(&self, _info: &UpdateInfo, progress: &dyn Fn(u8)) -> Result<(), Error> {
538 for percent in [0, 50, 100] {
539 progress(percent);
540 }
541 Ok(())
542 }
543
544 fn pending(&self) -> Option<Version> {
545 self.pending.clone()
546 }
547
548 fn apply(&self, applied: Applied, after: &After) -> Result<(), Error> {
549 let what = match applied {
550 Applied::Downloaded(info) => info.TargetFullRelease.Version,
551 Applied::Pending => "pending".to_owned(),
552 };
553 log(format!("apply {what} {after:?}"));
554 Ok(())
555 }
556 }
557
558 fn updater(spec: &'static PackageSpec, fake: Arc<Fake>) -> Updater {
559 Updater::with_backend(spec, fake)
560 }
561
562 fn wait_for(updater: &Updater, done: impl Fn(&PackageState) -> bool) -> PackageStatus {
564 let deadline = Instant::now() + Duration::from_secs(5);
565 loop {
566 let status = updater.status();
567 if done(&status.state) {
568 return status;
569 }
570 assert!(
571 Instant::now() < deadline,
572 "still {:?} after 5 seconds",
573 status.state
574 );
575 std::thread::sleep(Duration::from_millis(5));
576 }
577 }
578
579 fn settled(state: &PackageState) -> bool {
580 matches!(
581 state,
582 PackageState::UpToDate | PackageState::Ready { .. } | PackageState::Failed(_)
583 )
584 }
585
586 #[test]
587 fn a_found_update_is_downloaded_and_ready() {
588 let updater = updater(
589 spec(None),
590 Fake::new("1.0.0", vec![Ok(Some(info("1.1.0")))], None),
591 );
592 assert_eq!(
593 updater.status(),
594 PackageStatus {
595 current: Some(v("1.0.0")),
596 state: PackageState::Idle
597 }
598 );
599 updater.check_now();
600 let status = wait_for(&updater, settled);
601 assert_eq!(
602 status.state,
603 PackageState::Ready {
604 version: v("1.1.0")
605 }
606 );
607 assert_eq!(status.current, Some(v("1.0.0")));
608 }
609
610 #[test]
611 fn nothing_found_is_up_to_date() {
612 let updater = updater(spec(None), Fake::new("1.0.0", vec![Ok(None)], None));
613 updater.check_now();
614 assert_eq!(wait_for(&updater, settled).state, PackageState::UpToDate);
615 }
616
617 #[test]
618 fn a_check_error_is_failed_and_a_later_check_recovers() {
619 let updater = updater(
620 spec(None),
621 Fake::new(
622 "1.0.0",
623 vec![Err(Error::Network("offline".into())), Ok(None)],
624 None,
625 ),
626 );
627 updater.check_now();
628 let failed = wait_for(&updater, settled).state;
629 assert!(
630 matches!(&failed, PackageState::Failed(msg) if msg.contains("offline")),
631 "{failed:?}"
632 );
633 updater.check_now();
634 assert_eq!(
635 wait_for(&updater, |s| *s == PackageState::UpToDate).state,
636 PackageState::UpToDate
637 );
638 }
639
640 #[test]
641 fn apply_runs_before_apply_first() {
642 let updater = updater(
643 spec(Some(stopping)),
644 Fake::new("1.0.0", vec![Ok(Some(info("1.1.0")))], None),
645 );
646 updater.check_now();
647 wait_for(&updater, settled);
648 updater.apply(After::RestartApp).unwrap();
649 assert_eq!(apply_log(), ["before_apply", "apply 1.1.0 RestartApp"]);
650 }
651
652 #[test]
653 fn a_second_apply_while_one_runs_is_refused() {
654 static UPDATER: std::sync::OnceLock<Updater> = std::sync::OnceLock::new();
655 fn applying_again() -> Result<(), String> {
658 let again = UPDATER.get().unwrap().apply(After::RestartApp);
659 log(format!("again {again:?}"));
660 Ok(())
661 }
662 let updater = updater(
663 spec(Some(applying_again)),
664 Fake::new("1.0.0", vec![Ok(Some(info("1.1.0")))], None),
665 );
666 UPDATER.set(updater.clone()).ok().unwrap();
667 updater.check_now();
668 wait_for(&updater, settled);
669 updater.apply(After::RestartApp).unwrap();
670 assert_eq!(
671 apply_log(),
672 ["again Err(Applying)", "apply 1.1.0 RestartApp"]
673 );
674 assert!(matches!(
676 updater.apply(After::RestartApp),
677 Err(Error::Applying)
678 ));
679 }
680
681 #[test]
682 fn a_failing_before_apply_applies_nothing() {
683 let updater = updater(
684 spec(Some(failing_to_stop)),
685 Fake::new("1.0.0", vec![Ok(Some(info("1.1.0")))], None),
686 );
687 updater.check_now();
688 wait_for(&updater, settled);
689 let err = updater.apply(After::Exit).unwrap_err();
690 assert!(
691 matches!(err, Error::Prepare(ref m) if m == "stop failed"),
692 "{err:?}"
693 );
694 assert_eq!(apply_log(), ["before_apply"]);
695 assert_eq!(
696 updater.status().state,
697 PackageState::Failed("stop failed".to_owned())
698 );
699 }
700
701 #[test]
702 fn apply_without_a_download_is_an_error() {
703 let updater = updater(spec(Some(stopping)), Fake::new("1.0.0", vec![], None));
704 let err = updater.apply(After::RestartApp).unwrap_err();
705 assert!(
706 matches!(err, Error::Velopack(ref m) if m == "nothing downloaded"),
707 "{err:?}"
708 );
709 assert!(apply_log().is_empty());
710 }
711
712 #[test]
713 fn a_pending_update_newer_than_current_is_applied_at_start() {
714 let updater = updater(
715 spec(Some(stopping)),
716 Fake::new("1.0.0", vec![], Some("1.1.0")),
717 );
718 assert!(updater.apply_pending(false).unwrap());
719 assert_eq!(apply_log(), ["before_apply", "apply pending RestartApp"]);
720 }
721
722 #[test]
723 fn a_pending_package_not_newer_is_left_alone() {
724 for pending in [Some("1.0.0"), Some("0.9.0"), None] {
725 let updater = updater(spec(Some(stopping)), Fake::new("1.0.0", vec![], pending));
726 assert!(!updater.apply_pending(false).unwrap());
727 }
728 assert!(apply_log().is_empty());
729 }
730
731 #[test]
732 fn a_start_by_velopack_s_updater_applies_no_pending_package() {
733 let updater = updater(
734 spec(Some(stopping)),
735 Fake::new("1.0.0", vec![], Some("1.1.0")),
736 );
737 assert!(!updater.apply_pending(true).unwrap());
738 assert!(apply_log().is_empty());
739 }
740
741 #[test]
742 fn on_change_is_called_on_every_state_change() {
743 let updater = updater(
744 spec(None),
745 Fake::new("1.0.0", vec![Ok(Some(info("1.1.0")))], None),
746 );
747 let first = Arc::new(AtomicUsize::new(0));
748 let second = Arc::new(AtomicUsize::new(0));
749 for counter in [&first, &second] {
750 let counter = Arc::clone(counter);
751 updater.on_change(move || {
752 counter.fetch_add(1, Ordering::SeqCst);
753 });
754 }
755 updater.check_now();
756 wait_for(&updater, settled);
757 let deadline = Instant::now() + Duration::from_secs(5);
760 while second.load(Ordering::SeqCst) < 5 && Instant::now() < deadline {
761 std::thread::sleep(Duration::from_millis(5));
762 }
763 assert_eq!(first.load(Ordering::SeqCst), 5);
764 assert_eq!(second.load(Ordering::SeqCst), 5);
765 }
766
767 #[test]
768 fn a_failing_check_after_a_download_keeps_it_ready() {
769 let fake = Fake::new(
770 "1.0.0",
771 vec![
772 Ok(Some(info("1.1.0"))),
773 Err(Error::Network("offline".into())),
774 ],
775 None,
776 );
777 let updater = updater(spec(Some(stopping)), Arc::clone(&fake));
778 let changes = Arc::new(AtomicUsize::new(0));
779 let counter = Arc::clone(&changes);
780 updater.on_change(move || {
781 counter.fetch_add(1, Ordering::SeqCst);
782 });
783 updater.check_now();
784 wait_for(&updater, settled);
785 updater.check_now();
788 updater.check_now();
789 let deadline = Instant::now() + Duration::from_secs(5);
790 while fake.checked.load(Ordering::SeqCst) < 3 {
791 assert!(Instant::now() < deadline, "the checks never ran");
792 std::thread::sleep(Duration::from_millis(5));
793 }
794 assert_eq!(
795 updater.status().state,
796 PackageState::Ready {
797 version: v("1.1.0")
798 }
799 );
800 assert_eq!(changes.load(Ordering::SeqCst), 5);
802 updater.apply(After::RestartApp).unwrap();
803 assert_eq!(apply_log(), ["before_apply", "apply 1.1.0 RestartApp"]);
804 }
805
806 #[test]
807 fn apply_is_retried_after_a_failing_before_apply() {
808 static REFUSE: std::sync::atomic::AtomicBool = std::sync::atomic::AtomicBool::new(true);
809 fn refusing_once() -> Result<(), String> {
810 log("before_apply");
811 if REFUSE.swap(false, Ordering::SeqCst) {
812 Err("stop failed".to_owned())
813 } else {
814 Ok(())
815 }
816 }
817 let updater = updater(
818 spec(Some(refusing_once)),
819 Fake::new("1.0.0", vec![Ok(Some(info("1.1.0")))], None),
820 );
821 updater.check_now();
822 wait_for(&updater, settled);
823 let err = updater.apply(After::Exit).unwrap_err();
824 assert!(matches!(err, Error::Prepare(_)), "{err:?}");
825 assert_eq!(
826 updater.status().state,
827 PackageState::Failed("stop failed".to_owned())
828 );
829 updater.apply(After::Exit).unwrap();
830 assert_eq!(
831 apply_log(),
832 ["before_apply", "before_apply", "apply 1.1.0 Exit"]
833 );
834 }
835
836 #[test]
837 fn a_check_after_a_refused_before_apply_is_ready_again() {
838 let fake = Fake::new(
839 "1.0.0",
840 vec![
841 Ok(Some(info("1.1.0"))),
842 Err(Error::Network("offline".into())),
843 ],
844 None,
845 );
846 let updater = updater(spec(Some(failing_to_stop)), Arc::clone(&fake));
847 updater.check_now();
848 wait_for(&updater, settled);
849 updater.apply(After::RestartApp).unwrap_err();
850 assert_eq!(
851 updater.status().state,
852 PackageState::Failed("stop failed".to_owned())
853 );
854 updater.check_now();
856 let status = wait_for(&updater, |s| matches!(s, PackageState::Ready { .. }));
857 assert_eq!(
858 status.state,
859 PackageState::Ready {
860 version: v("1.1.0")
861 }
862 );
863 }
864
865 #[test]
866 fn run_in_background_checks_at_once_and_stops_with_the_last_updater() {
867 let fake = Fake::new("1.0.0", vec![Ok(None)], None);
868 let updater = updater(spec(None), Arc::clone(&fake));
869 updater.run_in_background(Duration::from_secs(3600));
870 assert_eq!(wait_for(&updater, settled).state, PackageState::UpToDate);
871 let clone = updater.clone();
872 drop(updater);
873 drop(clone);
874 let deadline = Instant::now() + Duration::from_secs(5);
877 while Arc::strong_count(&fake) > 1 {
878 assert!(Instant::now() < deadline, "the worker outlived the updater");
879 std::thread::sleep(Duration::from_millis(5));
880 }
881 }
882}