Add plugin restarting on panic

This patch adds an experimental implementation of plugin-restart on
panic.

The strategy can be defined in the configuration via the restart_policy
key, which can either be "Never", "Once" or "Always", default being
"Never".

This re-runs the plugin "main" if configured so.

This is highly experimental, not tested at all and might be completely
rubbish.

Signed-off-by: Matthias Beyer <matthias.beyer@ifm.com>

2 files changed, 45 insertions, 3 deletions

diff --git a/crates/core/tedge_core/src/configuration.rs b/crates/core/tedge_core/src/configuration.rs
index 56defb99..d6d58f2d 100644
--- a/crates/core/tedge_core/src/configuration.rs
+++ b/crates/core/tedge_core/src/configuration.rs
@@ -51,6 +51,10 @@ impl PluginInstanceConfiguration {
     pub fn configuration(&self) -> &InstanceConfiguration {
         &self.configuration
     }
+
+    pub fn restart_policy(&self) -> &RestartPolicy {
+        &self.restart_policy
+    }
 }
 
 #[derive(serde::Deserialize, Debug)]
diff --git a/crates/core/tedge_core/src/reactor.rs b/crates/core/tedge_core/src/reactor.rs
index 07d06c01..c97358e0 100644
--- a/crates/core/tedge_core/src/reactor.rs
+++ b/crates/core/tedge_core/src/reactor.rs
@@ -215,9 +215,47 @@ impl Reactor {
         let _main_results = all_plugins
             .iter_mut()
             .map(|plugin_task| {
-                let span =
-                    tracing::debug_span!("plugin.main", plugin.name = %plugin_task.plugin_name());
-                plugin_task.run_main().instrument(span)
+                async {
+                    let mut restarted_once = false;
+                    loop {
+                        let span =
+                            tracing::debug_span!("plugin.main", plugin.name = %plugin_task.plugin_name());
+
+                        match plugin_task.run_main().instrument(span).await {
+                            err @ Err(crate::errors::PluginLifecycleError::PluginMainPanicked(_)) => {
+                                let restart_policy = self.0.config().plugins().get(plugin_task.plugin_name())
+                                    .map(|pl| pl.restart_policy());
+
+                                match restart_policy {
+                                    Some(crate::configuration::RestartPolicy::Never) => {
+                                        return err
+                                    },
+                                    Some(crate::configuration::RestartPolicy::Once) => {
+                                        if restarted_once {
+                                            return err
+                                        }
+
+                                        restarted_once = true;
+                                        tracing::error!(error = ?err, "Error during plugin lifecycle");
+                                    },
+                                    Some(crate::configuration::RestartPolicy::Always) => {
+                                        restarted_once = true;
+                                        tracing::error!(error = ?err, "Error during plugin lifecycle");
+                                    },
+                                    None => {
+                                        // cannot happen
+                                        unimplemented!()
+                                    }
+                                }
+                            },
+
+                            Err(other_err) => return Err(other_err),
+                            Ok(()) => break,
+                        }
+                    }
+
+                    Ok(())
+                }
             })
             .collect::<futures::stream::FuturesUnordered<_>>()
             .collect::<Vec<Result<(), _>>>()