[RSDK-9593] move ota to config monitor, config monitor restart hook

micro-rdk/src/common/config_monitor.rs

@@ -8,13 +8,16 @@ use crate::{
 };
 use async_io::Timer;
 use futures_lite::{Future, FutureExt};
-use std::fmt::Debug;
-use std::pin::Pin;
-use std::time::Duration;
+use std::{fmt::Debug, pin::Pin, time::Duration};
+
+#[cfg(feature = "ota")]
+use crate::common::{exec::Executor, ota};
 
 pub struct ConfigMonitor<'a, Storage> {
     curr_config: Box<RobotConfig>, //config for robot gotten from last robot startup, aka inputted from entry
     storage: Storage,
+    #[cfg(feature = "ota")]
+    executor: Executor,
     restart_hook: Box<dyn Fn() + 'a>,
 }
 
@@ -27,11 +30,14 @@ where
     pub fn new(
         curr_config: Box<RobotConfig>,
         storage: Storage,
+        #[cfg(feature = "ota")] executor: Executor,
         restart_hook: impl Fn() + 'a,
     ) -> Self {
         Self {
             curr_config,
             storage,
+            #[cfg(feature = "ota")]
+            executor,
             restart_hook: Box::new(restart_hook),
         }
     }
@@ -71,16 +77,48 @@ where
                 })
                 .await?;
 
-            if new_config
-                .config
-                .is_some_and(|cfg| cfg != *self.curr_config)
-            {
-                if let Err(e) = self.storage.reset_robot_configuration() {
-                    log::warn!(
-                        "Failed to reset robot config after new config detected: {}",
-                        e
-                    );
-                } else {
+            if let Some(config) = new_config.as_ref().config.as_ref() {
+                let mut reboot = false;
+
+                #[cfg(feature = "ota")]
+                {
+                    if let Some(service) = config
+                        .services
+                        .iter()
+                        .find(|&service| service.model == *ota::OTA_MODEL_TRIPLET)
+                    {
+                        // TODO(RSDK-9676): new OtaService created at every invocation, not ideal
+                        match ota::OtaService::from_config(
+                            service,
+                            self.storage.clone(),
+                            self.executor.clone(),
+                        ) {
+                            Ok(mut ota) => match ota.update().await {
+                                Ok(needs_reboot) => reboot = needs_reboot,
+                                Err(e) => log::error!("failed to complete ota update: {}", e),
+                            },
+                            Err(e) => log::error!(
+                                "failed to create ota service from config:{} - {:?}",
+                                e,
+                                service,
+                            ),
+                        }
+                    }
+                }
+
+                if *config != *self.curr_config {
+                    if let Err(e) = self.storage.reset_robot_configuration() {
+                        log::warn!(
+                            "Failed to reset robot config after new config detected: {}",
+                            e
+                        );
+                    } else {
+                        reboot = true;
+                    }
+                }
+
+                if reboot {
+                    // TODO(RSDK-9464): flush logs to app.viam before restarting
                     self.restart();
                 }
             }

micro-rdk/src/common/conn/viam.rs

@@ -51,7 +51,7 @@ use super::server::{IncomingConnectionManager, WebRtcConfiguration};
 use crate::common::provisioning::server::AsNetwork;
 
 #[cfg(feature = "ota")]
-use crate::common::{credentials_storage::OtaMetadataStorage, ota};
+use crate::common::credentials_storage::OtaMetadataStorage;
 
 pub struct RobotCloudConfig {
     local_fqdn: String,
@@ -364,8 +364,6 @@ where
             http2_server_port: self.http2_server_port,
             wifi_manager: self.wifi_manager.into(),
             app_client_tasks: self.app_client_tasks,
-            #[cfg(feature = "ota")]
-            ota_service_task: Default::default(),
             max_concurrent_connections: self.max_concurrent_connections,
             network: Some(network),
         }
@@ -400,8 +398,6 @@ where
             http2_server_port: self.http2_server_port,
             wifi_manager: Rc::new(self.wifi_manager),
             app_client_tasks: self.app_client_tasks,
-            #[cfg(feature = "ota")]
-            ota_service_task: None,
             max_concurrent_connections: self.max_concurrent_connections,
             network: None,
         }
@@ -420,8 +416,6 @@ pub struct ViamServer<Storage, C, M> {
     http2_server_port: u16,
     wifi_manager: Rc<Option<Box<dyn WifiManager>>>,
     app_client_tasks: Vec<Box<dyn PeriodicAppClientTask>>,
-    #[cfg(feature = "ota")]
-    ota_service_task: Option<Task<()>>,
     max_concurrent_connections: usize,
     network: Option<Box<dyn Network>>,
 }
@@ -486,6 +480,14 @@ where
             self.provision().await;
         }
 
+        #[cfg(feature = "ota")]
+        {
+            match self.storage.get_ota_metadata() {
+                Ok(metadata) => log::info!("firmware version: {}", metadata.version),
+                Err(e) => log::warn!("not OTA firmware metadata available: {}", e),
+            };
+        }
+
         // Since provisioning was run and completed, credentials are properly populated
         // if wifi manager is configured loop forever until wifi is connected
         if let Some(wifi) = self.wifi_manager.as_ref().as_ref() {
@@ -576,45 +578,12 @@ where
         let config_monitor_task = Box::new(ConfigMonitor::new(
             config.clone(),
             self.storage.clone(),
+            #[cfg(feature = "ota")]
+            self.executor.clone(),
             || std::process::exit(0),
         ));
         self.app_client_tasks.push(config_monitor_task);
 
-        #[cfg(feature = "ota")]
-        {
-            log::debug!("ota feature enabled");
-
-            if let Some(service) = config
-                .services
-                .iter()
-                .find(|&service| service.model == *ota::OTA_MODEL_TRIPLET)
-            {
-                match ota::OtaService::from_config(
-                    service,
-                    self.storage.clone(),
-                    self.executor.clone(),
-                ) {
-                    Ok(mut ota) => {
-                        self.ota_service_task
-                            .replace(self.executor.spawn(async move {
-                                if let Err(e) = ota.update().await {
-                                    log::error!("failed to complete ota update {}", e);
-                                }
-                            }));
-                    }
-                    Err(e) => {
-                        log::error!("failed to build ota service: {}", e.to_string());
-                        log::error!("ota service config: {:?}", service);
-                    }
-                };
-            } else {
-                log::error!(
-                    "ota enabled, but no service of type `{}` found in robot config",
-                    ota::OTA_MODEL_TYPE
-                );
-            }
-        }
-
         let mut robot = LocalRobot::from_cloud_config(
             self.executor.clone(),
             robot_creds.robot_id.clone(),

micro-rdk/src/common/ota.rs

@@ -59,7 +59,8 @@ use thiserror::Error;
 #[cfg(not(feature = "esp32"))]
 use {bincode::Decode, futures_lite::AsyncWriteExt};
 
-const CONN_RETRY_SECS: u64 = 60;
+const CONN_RETRY_SECS: u64 = 1;
+const NUM_RETRY_CONN: usize = 3;
 const SIZEOF_APPDESC: usize = 256;
 const MAX_VER_LEN: usize = 128;
 pub const OTA_MODEL_TYPE: &str = "ota_service";
@@ -121,7 +122,7 @@ pub(crate) enum ConfigError {
 
 #[allow(dead_code)]
 #[derive(Error, Debug)]
-pub(crate) enum OtaError {
+pub(crate) enum OtaError<S: OtaMetadataStorage> {
     #[error("error occured during abort process: {0}")]
     AbortError(String),
     #[error("{0}")]
@@ -136,6 +137,10 @@ pub(crate) enum OtaError {
     InvalidImageSize(usize, usize),
     #[error("failed to retrieve firmware header info from binary, firmware may not be valid for this system: {0}")]
     InvalidFirmware(String),
+    #[error("failed to update OTA metadata: expected updated version to be `{0}`, found `{1}`")]
+    UpdateMetadata(String, String),
+    #[error(transparent)]
+    StorageError(<S as OtaMetadataStorage>::Error),
     #[error("error writing firmware to update partition: {0}")]
     WriteError(String),
     #[error("{0}")]
@@ -156,9 +161,6 @@ impl OtaMetadata {
     pub fn new(version: String) -> Self {
         Self { version }
     }
-    pub(crate) fn version(&self) -> &str {
-        &self.version
-    }
 }
 
 pub(crate) struct OtaService<S: OtaMetadataStorage> {
@@ -172,13 +174,23 @@ pub(crate) struct OtaService<S: OtaMetadataStorage> {
 }
 
 impl<S: OtaMetadataStorage> OtaService<S> {
+    pub(crate) fn stored_metadata(&self) -> Result<OtaMetadata, OtaError<S>> {
+        if !self.storage.has_ota_metadata() {
+            log::info!("no OTA metadata currently stored in NVS");
+        }
+
+        self.storage
+            .get_ota_metadata()
+            .map_err(OtaError::StorageError)
+    }
+
     pub(crate) fn from_config(
         new_config: &ServiceConfig,
         storage: S,
         exec: Executor,
-    ) -> Result<Self, OtaError> {
+    ) -> Result<Self, OtaError<S>> {
         let kind = new_config.attributes.as_ref().ok_or_else(|| {
-            ConfigError::Other("ota service config has no attributes".to_string())
+            ConfigError::Other("OTA service config has no attributes".to_string())
         })?;
 
         let url = kind
@@ -259,28 +271,17 @@ impl<S: OtaMetadataStorage> OtaService<S> {
         })
     }
 
-    pub(crate) async fn update(&mut self) -> Result<(), OtaError> {
-        let stored_metadata = if !self.storage.has_ota_metadata() {
-            log::info!("no ota metadata currently stored in NVS");
-            OtaMetadata::default()
-        } else {
-            self.storage
-                .get_ota_metadata()
-                .inspect_err(|e| log::warn!("failed to get ota metadata from nvs: {}", e))
-                .unwrap_or_default()
-        };
+    pub(crate) fn needs_update(&self) -> bool {
+        self.stored_metadata().unwrap_or_default().version != self.pending_version
+    }
 
-        if self.pending_version == stored_metadata.version() {
-            log::info!("firmware is up-to-date: `{}`", stored_metadata.version);
-            return Ok(());
+    /// Attempts to perform an OTA update.
+    /// On success, returns an `Ok(true)` or `Ok(false)` indicating if a reboot is necessary.
+    pub(crate) async fn update(&mut self) -> Result<bool, OtaError<S>> {
+        if !(self.needs_update()) {
+            return Ok(false);
         }
 
-        log::info!(
-            "firmware is out of date, proceeding with update from version `{}` to version `{}`",
-            stored_metadata.version,
-            self.pending_version
-        );
-
         let mut uri = self
             .url
             .parse::<hyper::Uri>()
@@ -304,11 +305,19 @@ impl<S: OtaMetadataStorage> OtaService<S> {
             uri = hyper::Uri::from_parts(parts).map_err(|e| OtaError::Other(e.to_string()))?;
         };
 
+        let mut num_tries = 0;
         let (mut sender, conn) = loop {
+            num_tries += 1;
+            if num_tries == NUM_RETRY_CONN + 1 {
+                return Err(OtaError::Other(
+                    "failed to establish connection".to_string(),
+                ));
+            }
             match self.connector.connect_to(&uri) {
                 Ok(connection) => {
                     match connection.await {
                         Ok(io) => {
+                            // TODO(RSDK-9617): add timeout for stalled download
                             match http2::Builder::new(self.exec.clone())
                                 .max_frame_size(16_384) // lowest configurable
                                 .timer(H2Timer)
@@ -519,16 +528,27 @@ impl<S: OtaMetadataStorage> OtaService<S> {
             })
             .map_err(|e| OtaError::Other(e.to_string()))?;
 
-        log::info!("firmware update complete");
+        // verifies nvs was stored correctly
+        let curr_metadata = self
+            .stored_metadata()
+            .inspect_err(|e| log::error!("OTA update failed to store new metadata: {e}"))?;
+        if curr_metadata.version != self.pending_version {
+            return Err(OtaError::UpdateMetadata(
+                self.pending_version.clone(),
+                curr_metadata.version,
+            ));
+        };
+        log::info!(
+            "firmware update successful: version `{}`",
+            curr_metadata.version
+        );
 
-        // Test experimental ffi accesses here to be recoverable without flashing
+        // Note: test experimental ota ffi accesses here to be recoverable without flashing
         #[cfg(feature = "esp32")]
         {
-            log::info!("rebooting to load firmware from `{:#x}`", self.address);
-            // TODO(RSDK-9464): flush logs to app.viam before restarting
-            esp_idf_svc::hal::reset::restart();
+            log::info!("next reboot will load firmware from `{:#x}`", self.address);
         }
 
-        Ok(())
+        Ok(true)
     }
 }