feat: added checking for bundle updates on game launch (#778)

packages/oneclient_app/src/hooks/actions.rs

@@ -3,8 +3,9 @@
 //! Always `spawn_forever` never `spawn` Freya's `spawn` cancels the task when
 //! the calling component unmounts this work is app-scoped not component-scoped
 
+use std::collections::HashMap;
 use std::path::PathBuf;
-use std::sync::Arc;
+use std::sync::{Arc, Mutex};
 use std::time::{Duration, Instant};
 
 use freya::prelude::spawn_forever;
@@ -34,6 +35,28 @@ const LAUNCH_HOLD: Duration = Duration::from_secs(2);
 /// Mirrors `oneclient_net`'s reachability probe not reqwest's per-request
 /// timeouts which are minutes long This sits between Play and the game
 const UPDATE_CHECK_BUDGET: Duration = Duration::from_secs(8);
+const BUNDLE_CHECK_INTERVAL: Duration = Duration::from_secs(15 * 60);
+const BUNDLE_SYNC_BUDGET: Duration = Duration::from_secs(30);
+/// The downloads need one too the net stack alone retries for minutes and the
+/// launch button is already claimed by then
+const BUNDLE_APPLY_BUDGET: Duration = Duration::from_secs(60);
+
+#[derive(Debug, Default)]
+struct BundleCheckLog {
+    checked_at: HashMap<ClusterId, Instant>,
+}
+
+impl BundleCheckLog {
+    fn due(&self, cluster_id: ClusterId, now: Instant) -> bool {
+        self.checked_at
+            .get(&cluster_id)
+            .is_none_or(|last| now.duration_since(*last) >= BUNDLE_CHECK_INTERVAL)
+    }
+
+    fn record(&mut self, cluster_id: ClusterId, now: Instant) {
+        self.checked_at.insert(cluster_id, now);
+    }
+}
 
 #[derive(Debug, Clone, PartialEq)]
 enum LaunchUpdatePlan {
@@ -77,6 +100,7 @@ pub struct Actions {
     station: RadioStation<AppState, AppChannel>,
     pump: mpsc::UnboundedSender<PumpSignal>,
     events: oneclient_events::EventBus,
+    bundle_checks: Arc<Mutex<BundleCheckLog>>,
 }
 
 impl Actions {
@@ -101,6 +125,7 @@ impl Actions {
             station,
             pump,
             events,
+            bundle_checks: Arc::new(Mutex::new(BundleCheckLog::default())),
         }
     }
 
@@ -108,6 +133,21 @@ impl Actions {
         let _ = self.pump.send(signal);
     }
 
+    fn bundle_check_due(&self, cluster_id: ClusterId) -> bool {
+        self.bundle_checks
+            .lock()
+            .unwrap()
+            .due(cluster_id, Instant::now())
+    }
+
+    fn record_bundle_checks(&self, cluster_ids: impl IntoIterator<Item = ClusterId>) {
+        let now = Instant::now();
+        let mut log = self.bundle_checks.lock().unwrap();
+        for cluster_id in cluster_ids {
+            log.record(cluster_id, now);
+        }
+    }
+
     fn with_engine(&self, mutate: impl FnOnce(&mut AppState)) {
         {
             let mut guard = self.station.clone().write_channel(AppChannel::Notifications);
@@ -1133,6 +1173,8 @@ impl Actions {
                 }
             };
 
+            actions.record_bundle_checks([cluster_id]);
+
             super::invalidate_cluster_queries().await;
             if let Some(spec) =
                 crate::install::cluster_update_notification(cluster_id, &result, &state.services)
@@ -1226,9 +1268,13 @@ impl Actions {
 
         spawn_forever(async move {
 
-            if let Err(err) = state.bundles.sync(&state.services.content()).await {
-                tracing::error!("bundle catalog sync failed: {err:#}");
-            }
+            let synced = match state.bundles.sync(&state.services.content()).await {
+                Ok(_) => true,
+                Err(err) => {
+                    tracing::error!("bundle catalog sync failed: {err:#}");
+                    false
+                }
+            };
             if let Err(err) = oneclient_core::clusters::apply_remote_migrations(&state).await {
                 tracing::error!("cluster migrations failed: {err:#}");
             }
@@ -1250,6 +1296,12 @@ impl Actions {
             .await;
             let session_id = session.detach();
 
+            if synced
+                && let Ok(clusters) = state.clusters.list().await
+            {
+                actions.record_bundle_checks(clusters.iter().map(|cluster| cluster.id));
+            }
+
             actions
                 .station
                 .clone()
@@ -1266,6 +1318,115 @@ impl Actions {
         });
     }
 
+    async fn resolve_bundle_updates_before_launch(
+        &self,
+        state: &Arc<oneclient_core::LauncherState>,
+        cluster_id: ClusterId,
+    ) {
+        if !self.bundle_check_due(cluster_id) {
+            tracing::debug!(cluster_id, "bundle update check skipped, checked recently");
+            return;
+        }
+
+        let content = state.services.content();
+
+        if let Ok(false) = oneclient_core::cluster_has_bundle_content(cluster_id, &content).await {
+            tracing::debug!(cluster_id, "bundle update check skipped, no bundle content");
+            return;
+        }
+
+        let mode = self.cluster_update_mode(state, cluster_id).await;
+
+        self.station
+            .clone()
+            .write_channel(AppChannel::Launcher)
+            .launcher
+            .syncing_bundles = true;
+
+        let synced = match tokio::time::timeout(BUNDLE_SYNC_BUDGET, state.bundles.sync(&content))
+            .await
+        {
+            Ok(Ok(_)) => true,
+            Ok(Err(err)) => {
+                tracing::warn!(
+                    cluster_id,
+                    error = %err,
+                    "bundle catalog sync failed, launching against the cached catalog"
+                );
+                false
+            }
+            Err(_elapsed) => {
+                tracing::debug!(
+                    cluster_id,
+                    "bundle catalog sync exceeded its launch budget"
+                );
+                false
+            }
+        };
+
+        self.station
+            .clone()
+            .write_channel(AppChannel::Launcher)
+            .launcher
+            .syncing_bundles = false;
+
+        // No prompt for bundles the package manager's markers carry them instead
+        if !matches!(mode, PackageUpdateMode::Automatic) {
+            tracing::debug!(cluster_id, "bundle updates left for the user to apply");
+            if synced {
+                self.record_bundle_checks([cluster_id]);
+            }
+            return;
+        }
+
+        let applied = match tokio::time::timeout(
+            BUNDLE_APPLY_BUDGET,
+            oneclient_core::apply_bundle_updates(cluster_id, state.bundles.as_ref(), &content),
+        )
+        .await
+        {
+            Ok(applied) => applied,
+            Err(_elapsed) => {
+                tracing::warn!(
+                    cluster_id,
+                    "bundle updates exceeded their launch budget, launching anyway"
+                );
+                return;
+            }
+        };
+
+        let result = match applied {
+            Ok(result) => result,
+            Err(err) => {
+                tracing::warn!(
+                    cluster_id,
+                    error = %err,
+                    "bundle update failed, launching anyway"
+                );
+                return;
+            }
+        };
+
+        if synced {
+            self.record_bundle_checks([cluster_id]);
+        }
+
+        if result.updates_applied.is_empty()
+            && result.additions_applied.is_empty()
+            && result.removals_applied.is_empty()
+        {
+            return;
+        }
+
+        super::invalidate_cluster_queries().await;
+
+        if let Some(spec) =
+            crate::install::cluster_update_notification(cluster_id, &result, &state.services).await
+        {
+            self.push_notification(spec);
+        }
+    }
+
     /// The check runs and the cache is written whatever the mode says keeping
     /// the package manager's "Update available" markers honest
     ///
@@ -1584,6 +1745,9 @@ async fn launch(actions: &Actions, cluster_id: ClusterId) {
 
     // Before the game process never after Minecraft reads its mods once at
     // startup
+    actions
+        .resolve_bundle_updates_before_launch(&state, cluster_id)
+        .await;
     actions
         .resolve_package_updates_before_launch(&state, cluster_id)
         .await;
@@ -1743,6 +1907,41 @@ mod tests {
         );
     }
 
+    #[test]
+    fn a_cluster_that_was_never_checked_is_due() {
+        assert!(BundleCheckLog::default().due(1, Instant::now()));
+    }
+
+    #[test]
+    fn a_relaunch_inside_the_window_reuses_the_last_check() {
+        let mut log = BundleCheckLog::default();
+        let now = Instant::now();
+        log.record(1, now);
+
+        assert!(!log.due(1, now));
+        assert!(!log.due(1, now + BUNDLE_CHECK_INTERVAL / 2));
+    }
+
+    #[test]
+    fn the_window_reopens_once_the_interval_has_passed() {
+        let mut log = BundleCheckLog::default();
+        let now = Instant::now();
+        log.record(1, now);
+
+        assert!(log.due(1, now + BUNDLE_CHECK_INTERVAL));
+        assert!(log.due(1, now + BUNDLE_CHECK_INTERVAL * 2));
+    }
+
+    #[test]
+    fn clusters_are_throttled_independently() {
+        let mut log = BundleCheckLog::default();
+        let now = Instant::now();
+        log.record(1, now);
+
+        assert!(!log.due(1, now));
+        assert!(log.due(2, now));
+    }
+
     #[test]
     fn nothing_pending_never_opens_a_modal() {
         for mode in PackageUpdateMode::ALL.iter().copied() {

packages/oneclient_content/src/bundles/mod.rs

@@ -35,5 +35,6 @@ pub use types::{
 };
 pub use updates::{
     apply_bundle_updates, apply_bundle_updates_with, check_bundle_updates,
+    cluster_has_bundle_content,
     get_bundles_with_update_status,
 };

packages/oneclient_content/src/bundles/updates.rs

@@ -54,6 +54,22 @@ pub async fn check_bundle_updates(
     check_bundle_updates_inner(cluster_id, bundles, ctx, &overrides).await
 }
 
+/// Local only nothing tracked and nothing opted into is a cluster no bundle can
+/// update so the catalog never has to be fetched for it
+#[tracing::instrument(level = "debug", skip(ctx))]
+pub async fn cluster_has_bundle_content(cluster_id: i64, ctx: &ContentCtx) -> ContentResult<bool> {
+    if !bundle_dao::list_bundle_tracked(&ctx.db, cluster_id)
+        .await?
+        .is_empty()
+    {
+        return Ok(true);
+    }
+
+    Ok(!bundle_dao::list_overrides(&ctx.db, cluster_id)
+        .await?
+        .is_empty())
+}
+
 #[tracing::instrument(level = "debug", skip(bundles, ctx, overrides))]
 async fn check_bundle_updates_inner(
     cluster_id: i64,

packages/oneclient_core/src/lib.rs

@@ -23,7 +23,8 @@ pub mod verify;
 pub mod versions;
 
 pub use oneclient_content::bundles::{
-    apply_bundle_updates, apply_bundle_updates_with, check_bundle_updates, effective_enabled,
+    apply_bundle_updates, apply_bundle_updates_with, check_bundle_updates,
+    cluster_has_bundle_content, effective_enabled,
     install_bundle,
     install_cluster_bundles,
     install_package_from_bundle, is_bundle_syncing, list_cluster_bundle_overrides,