dbus: rauc: remove tacd-based update polling

Signed-off-by: Leonard Göhrs <l.goehrs@pengutronix.de>

Author: hnez

src/broker/topic.rs

@@ -336,18 +336,6 @@ impl<E: Serialize + DeserializeOwned + Clone + PartialEq> Topic<E> {
 
         self.modify(|prev| if prev != msg { msg } else { None });
     }
-
-    /// Wait until the topic is set to the specified value
-    pub async fn wait_for(self: &Arc<Self>, val: E) {
-        let (mut stream, sub) = self.clone().subscribe_unbounded();
-
-        // Unwrap here to keep the interface simple. The stream could only yield
-        // None if the sender side is dropped, which will not happen as we hold
-        // an Arc to self which contains the senders vec.
-        while stream.next().await.unwrap() != val {}
-
-        sub.unsubscribe()
-    }
 }
 
 impl<E: Serialize + DeserializeOwned + Clone + Not + Not<Output = E>> Topic<E> {

src/dbus/rauc.rs

@@ -15,15 +15,12 @@
 // with this program; if not, write to the Free Software Foundation, Inc.,
 // 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301 USA.
 
-use std::cmp::Ordering;
 use std::collections::HashMap;
-use std::time::{Duration, Instant};
 
 use anyhow::Result;
 use async_std::channel::Receiver;
 use async_std::stream::StreamExt;
 use async_std::sync::Arc;
-use async_std::task::{sleep, spawn, JoinHandle};
 use log::warn;
 use serde::{Deserialize, Serialize};
 
@@ -48,38 +45,6 @@ use installer::InstallerProxy;
 
 #[cfg(feature = "demo_mode")]
 mod imports {
-    use std::collections::HashMap;
-
-    pub(super) struct InstallerProxy<'a> {
-        _dummy: &'a (),
-    }
-
-    impl<'a> InstallerProxy<'a> {
-        pub async fn new<C>(_conn: C) -> Option<InstallerProxy<'a>> {
-            Some(Self { _dummy: &() })
-        }
-
-        pub async fn inspect_bundle(
-            &self,
-            _source: &str,
-            _args: HashMap<&str, zbus::zvariant::Value<'_>>,
-        ) -> zbus::Result<HashMap<String, zbus::zvariant::OwnedValue>> {
-            let update: HashMap<String, String> = [
-                (
-                    "compatible".into(),
-                    "Linux Automation GmbH - LXA TAC".into(),
-                ),
-                ("version".into(), "24.04-20240415070800".into()),
-            ]
-            .into();
-
-            let info: HashMap<String, zbus::zvariant::OwnedValue> =
-                [("update".into(), update.into())].into();
-
-            Ok(info)
-        }
-    }
-
     pub(super) const CHANNELS_DIR: &str = "demo_files/usr/share/tacd/update_channels";
 }
 
@@ -91,10 +56,6 @@ mod imports {
     pub(super) const CHANNELS_DIR: &str = "/usr/share/tacd/update_channels";
 }
 
-const RELOAD_RATE_LIMIT: Duration = Duration::from_secs(10 * 60);
-const RETRY_INTERVAL_MIN: Duration = Duration::from_secs(60);
-const RETRY_INTERVAL_MAX: Duration = Duration::from_secs(60 * 60);
-
 use imports::*;
 
 #[derive(Serialize, Deserialize, Clone)]
@@ -158,20 +119,10 @@ pub struct Rauc {
     pub channels: Arc<Topic<Channels>>,
     pub reload: Arc<Topic<bool>>,
     pub should_reboot: Arc<Topic<bool>>,
+    #[allow(dead_code)]
     pub enable_polling: Arc<Topic<bool>>,
 }
 
-fn compare_versions(v1: &str, v2: &str) -> Option<Ordering> {
-    // Version strings look something like this: "4.0-0-20230428214619"
-    // Use string sorting on the date part to determine which bundle is newer.
-    let date_1 = v1.rsplit_once('-').map(|(_, d)| d);
-    let date_2 = v2.rsplit_once('-').map(|(_, d)| d);
-
-    // Return Sone if either version could not be split or a Some with the
-    // ordering between the dates.
-    date_1.zip(date_2).map(|(d1, d2)| d1.cmp(d2))
-}
-
 #[cfg(not(feature = "demo_mode"))]
 fn would_reboot_into_other_slot(slot_status: &SlotStatus, primary: Option<String>) -> Result<bool> {
     let rootfs_0 = slot_status.get("rootfs_0");
@@ -231,97 +182,15 @@ fn would_reboot_into_other_slot(slot_status: &SlotStatus, primary: Option<String
     }
 }
 
-async fn channel_polling_task(
-    conn: Arc<Connection>,
-    enable_polling: Arc<Topic<bool>>,
-    channels: Arc<Topic<Channels>>,
-    slot_status: Arc<Topic<Arc<SlotStatus>>>,
-    name: String,
-) {
-    let proxy = InstallerProxy::new(&conn).await.unwrap();
-
-    let mut retry_interval = RETRY_INTERVAL_MIN;
-
-    while let Some(mut channel) = channels
-        .try_get()
-        .and_then(|chs| chs.into_vec().into_iter().find(|ch| ch.name == name))
-    {
-        // Make sure update polling is enabled before doing anything,
-        // as contacting the update server requires user consent.
-        enable_polling.wait_for(true).await;
-
-        let polling_interval = channel.polling_interval;
-        let slot_status = slot_status.try_get();
-
-        if let Err(e) = channel.poll(&proxy, slot_status.as_deref()).await {
-            warn!(
-                "Failed to fetch update for update channel \"{}\": {}. Retrying in {}s.",
-                channel.name,
-                e,
-                retry_interval.as_secs()
-            );
-
-            if retry_interval < RETRY_INTERVAL_MAX {
-                sleep(retry_interval).await;
-
-                // Perform a (limited) exponential backoff on the retry interval to recover
-                // fast from short-term issues while also preventing the update server from
-                // being DDOSed by excessive retries.
-                retry_interval *= 2;
-
-                continue;
-            }
-        }
-
-        retry_interval = RETRY_INTERVAL_MIN;
-
-        channels.modify(|chs| {
-            let mut chs = chs?;
-            let channel_prev = chs.iter_mut().find(|ch| ch.name == name)?;
-
-            // Check if the bundle we polled is the same as before and we don't need
-            // to send a message to the subscribers.
-            if *channel_prev == channel {
-                return None;
-            }
-
-            // Update the channel description with the newly polled bundle info
-            *channel_prev = channel;
-
-            Some(chs)
-        });
-
-        match polling_interval {
-            Some(pi) => sleep(pi).await,
-            None => break,
-        }
-    }
-}
-
 async fn channel_list_update_task(
-    conn: Arc<Connection>,
     mut reload_stream: Receiver<bool>,
-    enable_polling: Arc<Topic<bool>>,
     channels: Arc<Topic<Channels>>,
-    slot_status: Arc<Topic<Arc<SlotStatus>>>,
 ) -> Result<()> {
-    let mut previous: Option<Instant> = None;
-    let mut polling_tasks: Vec<JoinHandle<_>> = Vec::new();
-
     while let Some(reload) = reload_stream.next().await {
         if !reload {
             continue;
         }
 
-        // Polling for updates is a somewhat expensive operation.
-        // Make sure it can not be abused to DOS the tacd.
-        if previous
-            .map(|p| p.elapsed() < RELOAD_RATE_LIMIT)
-            .unwrap_or(false)
-        {
-            continue;
-        }
-
         // Read the list of available update channels
         let new_channels = match Channels::from_directory(CHANNELS_DIR) {
             Ok(chs) => chs,
@@ -331,29 +200,7 @@ async fn channel_list_update_task(
             }
         };
 
-        // Stop the currently running polling tasks
-        for task in polling_tasks.drain(..) {
-            task.cancel().await;
-        }
-
-        let names: Vec<String> = new_channels.iter().map(|c| c.name.clone()).collect();
-
         channels.set(new_channels);
-
-        // Spawn new polling tasks. They will poll once immediately.
-        for name in names.into_iter() {
-            let polling_task = spawn(channel_polling_task(
-                conn.clone(),
-                enable_polling.clone(),
-                channels.clone(),
-                slot_status.clone(),
-                name,
-            ));
-
-            polling_tasks.push(polling_task);
-        }
-
-        previous = Some(Instant::now());
     }
 
     Ok(())
@@ -398,13 +245,7 @@ impl Rauc {
         let (reload_stream, _) = inst.reload.clone().subscribe_unbounded();
         wtb.spawn_task(
             "rauc-channel-list-update",
-            channel_list_update_task(
-                Arc::new(Connection),
-                reload_stream,
-                inst.enable_polling.clone(),
-                inst.channels.clone(),
-                inst.slot_status.clone(),
-            ),
+            channel_list_update_task(reload_stream, inst.channels.clone()),
         )?;
 
         Ok(inst)
@@ -422,7 +263,6 @@ impl Rauc {
         let operation = inst.operation.clone();
         let slot_status = inst.slot_status.clone();
         let primary = inst.primary.clone();
-        let channels = inst.channels.clone();
         let should_reboot = inst.should_reboot.clone();
 
         wtb.spawn_task("rauc-slot-status-update", async move {
@@ -485,23 +325,6 @@ impl Rauc {
                         })
                         .collect();
 
-                    // Update the `newer_than_installed` field for the upstream bundles inside
-                    // of the update channels.
-                    channels.modify(|prev| {
-                        let prev = prev?;
-
-                        let mut new = prev.clone();
-
-                        for ch in new.iter_mut() {
-                            if let Some(bundle) = ch.bundle.as_mut() {
-                                bundle.update_install(&slots);
-                            }
-                        }
-
-                        // Only send out messages if anything changed
-                        (new != prev).then_some(new)
-                    });
-
                     // Provide a simple yes/no "should reboot into other slot?" information
                     // based on the bundle versions in the booted slot and the other slot.
                     match would_reboot_into_other_slot(&slots, new_primary) {
@@ -608,13 +431,7 @@ impl Rauc {
         let (reload_stream, _) = inst.reload.clone().subscribe_unbounded();
         wtb.spawn_task(
             "rauc-channel-list-update",
-            channel_list_update_task(
-                conn.clone(),
-                reload_stream,
-                inst.enable_polling.clone(),
-                inst.channels.clone(),
-                inst.slot_status.clone(),
-            ),
+            channel_list_update_task(reload_stream, inst.channels.clone()),
         )?;
 
         Ok(inst)