Persist workflow definition of pending commands on start

When the agent restarts and loads pending operations,
these operations might not have been assign a workflow version.
This is notably the case after a self update from a previous version of
the agent that was oblivious of workflow versions.

In such a case, the agent assumes the current user-defined version of
the workflow is the version to be used for the pending command.
The workflow definition is persisted in `workflows-in-use` directory
and assigned to the command.

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

Author: didier-wenzek

crates/core/tedge_agent/src/operation_workflows/actor.rs

@@ -513,6 +513,7 @@ impl WorkflowActor {
                 for command in self
                     .workflow_repository
                     .load_pending_commands(pending_commands)
+                    .await
                 {
                     // Make sure the latest state is visible over MQTT
                     self.mqtt_publisher

crates/core/tedge_agent/src/operation_workflows/persist.rs

@@ -346,7 +346,25 @@ impl WorkflowRepository {
         None
     }
 
-    pub fn load_pending_commands(&mut self, commands: CommandBoard) -> Vec<GenericCommandState> {
+    pub async fn load_pending_commands(
+        &mut self,
+        mut commands: CommandBoard,
+    ) -> Vec<GenericCommandState> {
+        // If the resumed commands have been triggered by an agent without workflow version management
+        // then these commands are assigned the current version of the operation workflow.
+        // These currents versions have also to be marked as in use and persisted.
+        for (_, ref mut command) in commands.iter_mut() {
+            if command.workflow_version().is_none() {
+                if let Some(operation) = command.operation() {
+                    if let Some(current_version) = self.workflows.use_current_version(&operation) {
+                        self.persist_workflow_definition(&operation, &current_version)
+                            .await;
+                        *command = command.clone().set_workflow_version(&current_version);
+                    }
+                }
+            }
+        }
+
         self.workflows.load_pending_commands(commands)
     }
 

crates/core/tedge_api/src/workflow/supervisor.rs

@@ -76,25 +76,7 @@ impl WorkflowSupervisor {
     }
 
     /// Update on start the set of pending commands
-    pub fn load_pending_commands(
-        &mut self,
-        mut commands: CommandBoard,
-    ) -> Vec<GenericCommandState> {
-        // If the resumed commands have been triggered by an agent without workflow version management
-        // then these commands are assigned the current version for the operation.
-        // These currents versions have also to be marked as in use.
-        for (_, ref mut command) in commands.iter_mut() {
-            if command.workflow_version().is_none() {
-                if let Some(versions) = command
-                    .operation()
-                    .and_then(|operation| self.workflows.get_mut(&operation.as_str().into()))
-                {
-                    if let Some(current_version) = versions.use_current_version() {
-                        *command = command.clone().set_workflow_version(current_version);
-                    }
-                }
-            }
-        }
+    pub fn load_pending_commands(&mut self, commands: CommandBoard) -> Vec<GenericCommandState> {
         self.commands = commands;
         self.commands
             .iter()
@@ -150,6 +132,16 @@ impl WorkflowSupervisor {
         }
     }
 
+    /// Mark the current version of an operation workflow as being in use.
+    ///
+    /// Return the current version if any.
+    pub fn use_current_version(&mut self, operation: &OperationName) -> Option<WorkflowVersion> {
+        match self.workflows.get_mut(&operation.as_str().into()) {
+            Some(versions) => versions.use_current_version().cloned(),
+            None => None,
+        }
+    }
+
     /// Update the state of the command board on reception of a message sent by a peer over MQTT
     ///
     /// Return the new CommandRequest state if any.
@@ -399,7 +391,7 @@ impl WorkflowVersions {
         }
     }
 
-    // Mark the current version as being in-use.
+    /// Mark the current version as being in-use.
     fn use_current_version(&mut self) -> Option<&WorkflowVersion> {
         match self.current.as_ref() {
             Some((version, workflow)) => {
@@ -416,7 +408,7 @@ impl WorkflowVersions {
         }
     }
 
-    // Remove the current version from this list of versions, restoring the built-in version if any
+    /// Remove the current version from this list of versions, restoring the built-in version if any
     fn remove(&mut self, version: &WorkflowVersion) {
         if self.current.as_ref().map(|(v, _)| v == version) == Some(true) {
             self.current = None;