Replace polling with sleeps in handling intervals in tedge flows

Signed-off-by: James Rhodes <jarhodes314@gmail.com>

Author: jarhodes314

crates/extensions/tedge_flows/Cargo.toml

@@ -14,6 +14,7 @@ async-trait = { workspace = true }
 camino = { workspace = true, features = ["serde1"] }
 cfg-if = { workspace = true }
 fjall = { workspace = true, optional = true }
+futures = { workspace = true }
 humantime = { workspace = true }
 rquickjs = { git = "https://github.com/thin-edge/rquickjs", rev = "4ed04ac9af3de453dd41ff09fdd1837c7ceb1f1c", default-features = false, features = [
     # disable bindgen and rely on pre-generated bindings due to problems

crates/extensions/tedge_flows/src/actor.rs

@@ -7,6 +7,8 @@ use crate::InputMessage;
 use crate::OutputMessage;
 use async_trait::async_trait;
 use camino::Utf8PathBuf;
+use futures::future::Either;
+use std::future::pending;
 use tedge_actors::Actor;
 use tedge_actors::MessageReceiver;
 use tedge_actors::RuntimeError;
@@ -16,8 +18,7 @@ use tedge_file_system_ext::FsWatchEvent;
 use tedge_mqtt_ext::MqttMessage;
 use tedge_mqtt_ext::SubscriptionDiff;
 use tedge_mqtt_ext::TopicFilter;
-use tokio::time::interval;
-use tokio::time::Duration;
+use tokio::time::sleep_until;
 use tracing::error;
 use tracing::info;
 
@@ -34,11 +35,14 @@ impl Actor for FlowsMapper {
     }
 
     async fn run(mut self) -> Result<(), RuntimeError> {
-        let mut interval = interval(Duration::from_secs(1));
-
         loop {
+            let deadline_future = match self.processor.next_interval_deadline() {
+                Some(deadline) => Either::Left(sleep_until(deadline)),
+                None => Either::Right(pending()),
+            };
+
             tokio::select! {
-                _ = interval.tick() => {
+                _ = deadline_future => {
                     let drained_messages = self.drain_db().await?;
                     self.on_messages(MessageSource::MeaDB, drained_messages).await?;
 
@@ -147,8 +151,7 @@ impl FlowsMapper {
         for (flow_id, flow_messages) in self.processor.on_interval(timestamp).await {
             match flow_messages {
                 Ok(messages) => {
-                    self.publish_messages(flow_id.clone(), messages)
-                        .await?;
+                    self.publish_messages(flow_id.clone(), messages).await?;
                 }
                 Err(err) => {
                     error!(target: "flows", "{flow_id}: {err}");
@@ -184,15 +187,18 @@ impl FlowsMapper {
                     }
                 }
                 FlowOutput::MeaDB { output_series } => {
-                    let messages = messages.into_iter().map(|m| (DateTime::now(), m)).collect::<Vec<_>>();
-                        if let Err(err) = self
-                            .processor
-                            .database
-                            .store_many(output_series, messages)
-                            .await
-                        {
-                            error!(target: "flows", "{flow_id}: fail to persist message: {err}");
-                        }
+                    let messages = messages
+                        .into_iter()
+                        .map(|m| (DateTime::now(), m))
+                        .collect::<Vec<_>>();
+                    if let Err(err) = self
+                        .processor
+                        .database
+                        .store_many(output_series, messages)
+                        .await
+                    {
+                        error!(target: "flows", "{flow_id}: fail to persist message: {err}");
+                    }
                 }
             }
         }

crates/extensions/tedge_flows/src/config.rs

@@ -118,15 +118,20 @@ impl FlowConfig {
             js_runtime.load_script(&mut step.script).await?;
             step.check(&source);
             step.fix();
+            step.script.init_next_execution();
             steps.push(step);
         }
         let output = self.output.try_into()?;
-        Ok(Flow {
+        let mut flow = Flow {
             input,
             steps,
             source,
             output,
-        })
+            next_drain: None,
+            last_drain: None,
+        };
+        flow.init_next_drain();
+        Ok(flow)
     }
 }
 

crates/extensions/tedge_flows/src/flow.rs

@@ -26,6 +26,12 @@ pub struct Flow {
 
     /// Target of the transformed messages
     pub output: FlowOutput,
+
+    /// Next time to drain database for MeaDB inputs (for deadline-based wakeup)
+    pub next_drain: Option<tokio::time::Instant>,
+
+    /// Last time database was drained (for frequency checking)
+    pub last_drain: Option<DateTime>,
 }
 
 /// A message transformation step
@@ -94,6 +100,46 @@ impl Flow {
         }
     }
 
+    pub fn init_next_drain(&mut self) {
+        if let FlowInput::MeaDB { frequency, .. } = &self.input {
+            if !frequency.is_zero() {
+                self.next_drain = Some(tokio::time::Instant::now() + *frequency);
+            }
+        }
+    }
+
+    pub fn should_drain_at(&mut self, timestamp: DateTime) -> bool {
+        if let FlowInput::MeaDB { frequency, .. } = &self.input {
+            if frequency.is_zero() {
+                return false;
+            }
+
+            // Check if enough time has passed since last drain
+            match self.last_drain {
+                Some(last_drain) => {
+                    let elapsed_secs = timestamp.seconds.saturating_sub(last_drain.seconds);
+                    let frequency_secs = frequency.as_secs();
+                    if elapsed_secs >= frequency_secs {
+                        self.last_drain = Some(timestamp);
+                        // Also update the deadline for the actor loop
+                        self.next_drain = Some(tokio::time::Instant::now() + *frequency);
+                        true
+                    } else {
+                        false
+                    }
+                }
+                None => {
+                    // First drain
+                    self.last_drain = Some(timestamp);
+                    self.next_drain = Some(tokio::time::Instant::now() + *frequency);
+                    true
+                }
+            }
+        } else {
+            false
+        }
+    }
+
     pub fn topics(&self) -> TopicFilter {
         let mut topics = self.input.topics();
         for step in self.steps.iter() {
@@ -159,7 +205,7 @@ impl Flow {
     ) -> Result<Vec<Message>, FlowError> {
         let stated_at = stats.flow_on_interval_start(self.source.as_str());
         let mut messages = vec![];
-        for step in self.steps.iter() {
+        for step in self.steps.iter_mut() {
             let js = step.script.source();
             // Process first the messages triggered upstream by the tick
             let mut transformed_messages = vec![];
@@ -175,7 +221,8 @@ impl Flow {
                 transformed_messages.extend(step_output?);
             }
 
-            // Only then process the tick
+            // Only then process the tick if it's time to execute
+            if step.script.should_execute_interval() {
             let step_started_at = stats.flow_step_start(&js, "onInterval");
             let tick_output = step.script.on_interval(js_runtime, timestamp).await;
             match &tick_output {
@@ -185,6 +232,7 @@ impl Flow {
                 Err(_) => stats.flow_step_failed(&js, "onInterval"),
             }
             transformed_messages.extend(tick_output?);
+            }
 
             // Iterate with all the messages collected at this step
             messages = transformed_messages;

crates/extensions/tedge_flows/src/js_script.rs

@@ -6,6 +6,7 @@ use crate::js_runtime::JsRuntime;
 use crate::js_value::JsonValue;
 use camino::Utf8Path;
 use camino::Utf8PathBuf;
+use tokio::time::Instant;
 use tracing::debug;
 
 #[derive(Clone)]
@@ -14,6 +15,7 @@ pub struct JsScript {
     pub path: Utf8PathBuf,
     pub config: JsonValue,
     pub interval: std::time::Duration,
+    pub next_execution: Option<Instant>,
     pub no_js_on_message_fun: bool,
     pub no_js_on_config_update_fun: bool,
     pub no_js_on_interval_fun: bool,
@@ -27,6 +29,7 @@ impl JsScript {
             path,
             config: JsonValue::default(),
             interval: std::time::Duration::ZERO,
+            next_execution: None,
             no_js_on_message_fun: true,
             no_js_on_config_update_fun: true,
             no_js_on_interval_fun: true,
@@ -116,24 +119,51 @@ impl JsScript {
         Ok(())
     }
 
+    /// Initialize the next execution time for this script's interval
+    /// Should be called after the script is loaded and interval is set
+    pub fn init_next_execution(&mut self) {
+        if !self.no_js_on_interval_fun && !self.interval.is_zero() {
+            self.next_execution = Some(Instant::now() + self.interval);
+        }
+    }
+
+    /// Check if this script should execute its interval function now
+    /// Returns true and updates next_execution if it's time to execute
+    pub fn should_execute_interval(&mut self) -> bool {
+        if self.no_js_on_interval_fun || self.interval.is_zero() {
+            return false;
+        }
+
+        let now = Instant::now();
+        match self.next_execution {
+            Some(deadline) if now >= deadline => {
+                // Time to execute - schedule next execution
+                self.next_execution = Some(now + self.interval);
+                true
+            }
+            None => {
+                // First execution - initialize and execute
+                self.next_execution = Some(now + self.interval);
+                true
+            }
+            _ => false,
+        }
+    }
+
     /// Trigger the onInterval function of the JS module
     ///
     /// The "onInterval" function is passed 2 arguments
     /// - the current timestamp
     /// - the current flow step config
     ///
     /// Return zero, one or more messages
+    ///
+    /// Note: Caller should check should_execute_interval() before calling this
     pub async fn on_interval(
         &self,
         js: &JsRuntime,
         timestamp: DateTime,
     ) -> Result<Vec<Message>, FlowError> {
-        if self.no_js_on_interval_fun {
-            return Ok(vec![]);
-        }
-        if !timestamp.tick_now(self.interval) {
-            return Ok(vec![]);
-        }
         debug!(target: "flows", "{}: onInterval({timestamp:?})", self.module_name());
         let input = vec![timestamp.into(), self.config.clone()];
         js.call_function(&self.module_name(), "onInterval", input)

crates/extensions/tedge_flows/src/runtime.rs

@@ -155,6 +155,20 @@ impl MessageProcessor {
         topics
     }
 
+    /// Get the next deadline for interval execution across all scripts and database drains
+    /// Returns None if no scripts have intervals configured and no database drains are scheduled
+    pub fn next_interval_deadline(&self) -> Option<tokio::time::Instant> {
+        let script_deadlines = self
+            .flows
+            .values()
+            .flat_map(|flow| &flow.steps)
+            .filter_map(|step| step.script.next_execution);
+
+        let drain_deadlines = self.flows.values().filter_map(|flow| flow.next_drain);
+
+        script_deadlines.chain(drain_deadlines).min()
+    }
+
     pub async fn on_message(
         &mut self,
         source: MessageSource,
@@ -206,20 +220,19 @@ impl MessageProcessor {
         timestamp: DateTime,
     ) -> Vec<(String, Result<Vec<(DateTime, Message)>, DatabaseError>)> {
         let mut out_messages = vec![];
-        for (flow_id, flow) in self.flows.iter() {
-            if let FlowInput::MeaDB {
-                series: input_series,
-                frequency: input_frequency,
-                max_age: input_span,
-            } = &flow.input
-            {
-                if timestamp.tick_now(*input_frequency) {
+        for (flow_id, flow) in self.flows.iter_mut() {
+            if flow.should_drain_at(timestamp) {
+                if let FlowInput::MeaDB {
+                    series: input_series,
+                    max_age: input_span,
+                    ..
+                } = &flow.input
+                {
                     let cutoff_time = timestamp - *input_span;
                     let drained_messages = self
                         .database
                         .drain_older_than(cutoff_time, input_series)
-                        .await
-                        .map_err(DatabaseError::from);
+                        .await;
                     out_messages.push((flow_id.to_owned(), drained_messages));
                 }
             }
@@ -241,6 +254,7 @@ impl MessageProcessor {
                 if step.script.path() == path {
                     match self.js_runtime.load_script(&mut step.script).await {
                         Ok(()) => {
+                            step.script.init_next_execution();
                             info!(target: "flows", "Reloaded flow script {path}");
                         }
                         Err(e) => {