Merge pull request #2145 from oasisprotocol/kostko/feature/rofl-watchdog

runtime-sdk/modules/rofl: Add watchdog task

rofl-containers/Cargo.toml

@@ -1,6 +1,6 @@
 [package]
 name = "rofl-containers"
-version = "0.3.3"
+version = "0.3.5"
 edition = "2021"
 
 [dependencies]

runtime-sdk/src/modules/rofl/app/mod.rs

@@ -26,6 +26,7 @@ mod notifier;
 pub mod prelude;
 mod processor;
 mod registration;
+mod watchdog;
 
 pub use crate::modules::rofl::app_id::AppId;
 pub use client::Client;

runtime-sdk/src/modules/rofl/app/processor.rs

@@ -16,7 +16,7 @@ use crate::{
 };
 use rand::rngs::OsRng;
 
-use super::{notifier, registration, App, Environment};
+use super::{notifier, registration, watchdog, App, Environment};
 
 /// Size of the processor command queue.
 const CMDQ_BACKLOG: usize = 32;
@@ -31,6 +31,8 @@ pub(super) enum Command {
     GetLatestRound(oneshot::Sender<u64>),
     /// Notification that initial registration has been completed.
     InitialRegistrationCompleted,
+    /// Registration refreshed.
+    RegistrationRefreshed,
 }
 
 /// Processor state.
@@ -45,6 +47,7 @@ pub(super) struct State<A: App> {
 struct Tasks<A: App> {
     registration: registration::Task<A>,
     notifier: notifier::Task<A>,
+    watchdog: watchdog::Task,
 }
 
 /// Processor.
@@ -89,6 +92,7 @@ where
             tasks: Tasks {
                 registration: registration::Task::new(state.clone(), env.clone()),
                 notifier: notifier::Task::new(state.clone(), env.clone()),
+                watchdog: watchdog::Task::new(),
             },
             state,
             env,
@@ -125,6 +129,7 @@ where
         // Start the tasks.
         self.tasks.registration.start();
         self.tasks.notifier.start();
+        self.tasks.watchdog.start();
 
         slog::info!(self.logger, "entering processor loop");
         while let Some(cmd) = self.cmdq.recv().await {
@@ -146,6 +151,7 @@ where
             Command::InitialRegistrationCompleted => {
                 self.cmd_initial_registration_completed().await
             }
+            Command::RegistrationRefreshed => self.tasks.watchdog.keep_alive().await,
         }
     }
 

runtime-sdk/src/modules/rofl/app/registration.rs

@@ -156,6 +156,11 @@ where
         }
         self.last_registration_epoch = Some(epoch);
 
+        // Notify about registration refresh.
+        self.env
+            .send_command(processor::Command::RegistrationRefreshed)
+            .await?;
+
         Ok(())
     }
 }

runtime-sdk/src/modules/rofl/app/watchdog.rs

@@ -0,0 +1,77 @@
+use anyhow::Result;
+use tokio::{sync::mpsc, time};
+
+use crate::core::common::{logger::get_logger, process};
+
+/// Interval in which at least one keep-alive must be delivered to avoid the watchdog from
+/// terminating the application.
+const WATCHDOG_TRIGGER_INTERVAL: u64 = 6 * 3600; // 6 hours
+
+/// Application watchdog task.
+pub(super) struct Task {
+    imp: Option<Impl>,
+    tx: mpsc::Sender<()>,
+}
+
+impl Task {
+    /// Create an application watchdog task.
+    pub(super) fn new() -> Self {
+        let (tx, rx) = mpsc::channel(16);
+
+        let imp = Impl {
+            logger: get_logger("modules/rofl/app/watchdog"),
+            notify: rx,
+        };
+
+        Self { imp: Some(imp), tx }
+    }
+
+    /// Start the application watchdog task.
+    pub(super) fn start(&mut self) {
+        if let Some(imp) = self.imp.take() {
+            imp.start();
+        }
+    }
+
+    /// Notify the watchdog that we are still alive.
+    pub(super) async fn keep_alive(&self) -> Result<()> {
+        self.tx.send(()).await?;
+        Ok(())
+    }
+}
+
+struct Impl {
+    logger: slog::Logger,
+
+    notify: mpsc::Receiver<()>,
+}
+
+impl Impl {
+    /// Start the application watchdog task.
+    pub(super) fn start(self) {
+        tokio::task::spawn(self.run());
+    }
+
+    /// Run the application watchdog task.
+    async fn run(mut self) {
+        slog::info!(self.logger, "starting watchdog task");
+
+        loop {
+            tokio::select! {
+                Some(()) = self.notify.recv() => {
+                    // Keep-alive received, reset watchdog.
+                },
+
+                _ = time::sleep(time::Duration::from_secs(WATCHDOG_TRIGGER_INTERVAL)) => {
+                    // Watchdog triggered, kill the process.
+                    slog::error!(self.logger, "keep-alive not received, terminating application");
+                    process::abort();
+                },
+
+                else => break,
+            }
+        }
+
+        slog::info!(self.logger, "watchdog task stopped");
+    }
+}