Update c8y-firmare-manager to await download results

This is an intermediate step. The next one will be
to spawn a task to handle the download and
subsequent steps so a the manager is not blocked during that time.
The failing test (handle_child_response_while_busy_downloading) is a
witness of this missing requirement.

Signed-off-by: Didier Wenzek <didier.wenzek@free.fr>

Author: didier-wenzek

crates/extensions/c8y_firmware_manager/src/actor.rs

@@ -19,6 +19,7 @@ use std::os::unix::fs as unix_fs;
 use std::path::Path;
 use tedge_actors::fan_in_message_type;
 use tedge_actors::Actor;
+use tedge_actors::ClientMessageBox;
 use tedge_actors::DynSender;
 use tedge_actors::LoggingReceiver;
 use tedge_actors::MessageReceiver;
@@ -54,13 +55,11 @@ pub type OperationTimeout = Timeout<OperationKey>;
 pub type IdDownloadResult = (String, DownloadResult);
 pub type IdDownloadRequest = (String, DownloadRequest);
 
-fan_in_message_type!(FirmwareInput[MqttMessage, OperationTimeout, IdDownloadResult] : Debug);
-fan_in_message_type!(FirmwareOutput[MqttMessage, OperationSetTimeout, IdDownloadRequest] : Debug);
+fan_in_message_type!(FirmwareInput[MqttMessage, OperationTimeout] : Debug);
 
 pub struct FirmwareManagerActor {
     config: FirmwareManagerConfig,
     active_child_ops: HashMap<OperationKey, ActiveOperationState>,
-    reqs_pending_download: HashMap<String, SmartRestFirmwareRequest>,
     message_box: FirmwareManagerMessageBox,
 }
 
@@ -90,9 +89,6 @@ impl Actor for FirmwareManagerActor {
                 FirmwareInput::OperationTimeout(timeout) => {
                     self.process_operation_timeout(timeout).await?;
                 }
-                FirmwareInput::IdDownloadResult((id, result)) => {
-                    self.process_downloaded_firmware(&id, result).await?
-                }
             }
         }
         Ok(())
@@ -104,7 +100,6 @@ impl FirmwareManagerActor {
         Self {
             config,
             active_child_ops: HashMap::new(),
-            reqs_pending_download: HashMap::new(),
             message_box,
         }
     }
@@ -262,12 +257,13 @@ impl FirmwareManagerActor {
                 DownloadRequest::new(firmware_url, cache_file_path.as_std_path())
             };
 
-            self.message_box
+            let (_, download_result) = self
+                .message_box
                 .download_sender
-                .send((operation_id.to_string(), download_request))
+                .await_response((operation_id.to_string(), download_request))
+                .await?;
+            self.process_downloaded_firmware(operation_id, smartrest_request, download_result)
                 .await?;
-            self.reqs_pending_download
-                .insert(operation_id.to_string(), smartrest_request);
         }
         Ok(())
     }
@@ -278,39 +274,32 @@ impl FirmwareManagerActor {
     async fn process_downloaded_firmware(
         &mut self,
         operation_id: &str,
+        smartrest_request: SmartRestFirmwareRequest,
         download_result: DownloadResult,
     ) -> Result<(), FirmwareManagementError> {
-        if let Some(smartrest_request) = self.reqs_pending_download.remove(operation_id) {
-            let child_id = smartrest_request.device.clone();
-            match download_result {
-                Ok(response) => {
-                    if let Err(err) =
-                        // Publish a firmware update request to child device.
-                        self
-                            .handle_firmware_update_request_with_downloaded_file(
-                                smartrest_request,
-                                operation_id,
-                                &response.file_path,
-                            )
-                            .await
-                    {
-                        self.fail_operation_in_cloud(
-                            &child_id,
-                            Some(operation_id),
-                            &err.to_string(),
+        let child_id = smartrest_request.device.clone();
+        match download_result {
+            Ok(response) => {
+                if let Err(err) =
+                    // Publish a firmware update request to child device.
+                    self
+                        .handle_firmware_update_request_with_downloaded_file(
+                            smartrest_request,
+                            operation_id,
+                            &response.file_path,
                         )
-                        .await?;
-                    }
-                }
-                Err(err) => {
-                    let firmware_url = smartrest_request.url;
-                    let failure_reason = format!("Download from {firmware_url} failed with {err}");
-                    self.fail_operation_in_cloud(&child_id, Some(operation_id), &failure_reason)
+                        .await
+                {
+                    self.fail_operation_in_cloud(&child_id, Some(operation_id), &err.to_string())
                         .await?;
                 }
             }
-        } else {
-            error!("Unexpected: Download completed for unknown operation: {operation_id}");
+            Err(err) => {
+                let firmware_url = smartrest_request.url;
+                let failure_reason = format!("Download from {firmware_url} failed with {err}");
+                self.fail_operation_in_cloud(&child_id, Some(operation_id), &failure_reason)
+                    .await?;
+            }
         }
         Ok(())
     }
@@ -743,7 +732,7 @@ pub struct FirmwareManagerMessageBox {
     mqtt_publisher: DynSender<MqttMessage>,
     jwt_retriever: JwtRetriever,
     timer_sender: DynSender<SetTimeout<OperationKey>>,
-    download_sender: DynSender<IdDownloadRequest>,
+    download_sender: ClientMessageBox<IdDownloadRequest, IdDownloadResult>,
 }
 
 impl FirmwareManagerMessageBox {
@@ -752,7 +741,7 @@ impl FirmwareManagerMessageBox {
         mqtt_publisher: DynSender<MqttMessage>,
         jwt_retriever: JwtRetriever,
         timer_sender: DynSender<SetTimeout<OperationKey>>,
-        download_sender: DynSender<IdDownloadRequest>,
+        download_sender: ClientMessageBox<IdDownloadRequest, IdDownloadResult>,
     ) -> Self {
         Self {
             input_receiver,

crates/extensions/c8y_firmware_manager/src/lib.rs

@@ -19,6 +19,7 @@ use c8y_http_proxy::credentials::JwtRetriever;
 pub use config::*;
 use tedge_actors::futures::channel::mpsc;
 use tedge_actors::Builder;
+use tedge_actors::ClientMessageBox;
 use tedge_actors::DynSender;
 use tedge_actors::LinkError;
 use tedge_actors::LoggingReceiver;
@@ -40,7 +41,7 @@ pub struct FirmwareManagerBuilder {
     mqtt_publisher: DynSender<MqttMessage>,
     jwt_retriever: JwtRetriever,
     timer_sender: DynSender<SetTimeout<OperationKey>>,
-    download_sender: DynSender<IdDownloadRequest>,
+    download_sender: ClientMessageBox<IdDownloadRequest, IdDownloadResult>,
     signal_sender: mpsc::Sender<RuntimeRequest>,
 }
 
@@ -50,7 +51,11 @@ impl FirmwareManagerBuilder {
         mqtt_actor: &mut impl ServiceProvider<MqttMessage, MqttMessage, TopicFilter>,
         jwt_actor: &mut impl Service<(), JwtResult>,
         timer_actor: &mut impl ServiceProvider<OperationSetTimeout, OperationTimeout, NoConfig>,
-        downloader_actor: &mut impl ServiceProvider<IdDownloadRequest, IdDownloadResult, NoConfig>,
+        downloader_actor: &mut impl ServiceProvider<
+            RequestEnvelope<IdDownloadRequest, IdDownloadResult>,
+            NoMessage,
+            ReplyToRequester,
+        >,
     ) -> Result<FirmwareManagerBuilder, FileError> {
         Self::init(&config.data_dir)?;
 
@@ -66,7 +71,7 @@ impl FirmwareManagerBuilder {
             mqtt_actor.connect_consumer(Self::subscriptions(), input_sender.clone().into());
         let jwt_retriever = JwtRetriever::new(jwt_actor);
         let timer_sender = timer_actor.connect_consumer(NoConfig, input_sender.clone().into());
-        let download_sender = downloader_actor.connect_consumer(NoConfig, input_sender.into());
+        let download_sender = ClientMessageBox::new(downloader_actor);
         Ok(Self {
             config,
             input_receiver,

crates/extensions/c8y_firmware_manager/src/tests.rs

@@ -699,7 +699,7 @@ async fn spawn_firmware_manager(
         TimedMessageBox<SimpleMessageBox<MqttMessage, MqttMessage>>,
         TimedMessageBox<FakeServerBox<JwtRequest, JwtResult>>,
         SimpleMessageBox<OperationSetTimeout, OperationTimeout>,
-        TimedMessageBox<SimpleMessageBox<IdDownloadRequest, IdDownloadResult>>,
+        TimedMessageBox<FakeServerBox<IdDownloadRequest, IdDownloadResult>>,
     ),
     DynError,
 > {
@@ -726,8 +726,8 @@ async fn spawn_firmware_manager(
     let mut jwt_builder: FakeServerBoxBuilder<JwtRequest, JwtResult> = FakeServerBox::builder();
     let mut timer_builder: SimpleMessageBoxBuilder<OperationSetTimeout, OperationTimeout> =
         SimpleMessageBoxBuilder::new("Timer", 5);
-    let mut downloader_builder: SimpleMessageBoxBuilder<IdDownloadRequest, IdDownloadResult> =
-        SimpleMessageBoxBuilder::new("Downloader", 5);
+    let mut downloader_builder: FakeServerBoxBuilder<IdDownloadRequest, IdDownloadResult> =
+        FakeServerBox::builder();
 
     let firmware_manager_builder = FirmwareManagerBuilder::try_new(
         config,