fix(eon): handle rebirth commands and timestamp-less CMD payloads

CHANGELOG.md

@@ -24,6 +24,11 @@
   subscribe to `spBv1.0/STATE/<primary_host_id>`, defer NBIRTH and DBIRTH until the
   host publishes an online STATE, publish NDEATH and reconnect when the host goes
   offline, and ignore STATE messages with older timestamps
+- Accept command payloads that omit the payload timestamp by using the time the
+  command was received, instead of discarding the command
+- Stop data publishes immediately when a rebirth command is received, so no DATA
+  messages are sent after the command, and publish the node and device births from
+  the same session
 - Stop accepting data publishes before publishing the NDEATH on shutdown, and use
   `Client::disconnect_with_will` when the client supports it. This stops DATA
   arriving after the death certificate and satisfies the MQTT 5 intentional

srad-eon/src/device.rs

@@ -14,7 +14,7 @@ use tokio::{select, sync::mpsc, task};
 
 use crate::{
     birth::{BirthInitializer, BirthObjectType},
-    metric::{MetricPublisher, PublishError, PublishMetric},
+    metric::{MessageMetrics, MetricPublisher, PublishError, PublishMetric},
     metric_manager::manager::DynDeviceMetricManager,
     node::{EoNState, TemplateRegistry},
     BirthType, StateError,
@@ -257,15 +257,23 @@ impl Device {
                     return;
                 }
             };
-            let message_metrics = match payload.try_into() {
-                Ok(metrics) => metrics,
-                Err(_) => {
-                    warn!(
-                        "Got invalid CMD payload for device - ignoring. Device = {}",
-                        self.state.name
-                    );
-                    return;
+            let message_metrics = if payload.timestamp.is_some() {
+                match payload.try_into() {
+                    Ok(metrics) => metrics,
+                    Err(e) => {
+                        warn!(
+                            "Got invalid CMD payload for device - ignoring. Device = {} error={e}",
+                            self.state.name
+                        );
+                        return;
+                    }
                 }
+            } else {
+                warn!(
+                    "Got CMD payload without a timestamp for device. using the current time. Device = {}",
+                    self.state.name
+                );
+                MessageMetrics::from_payload_lenient(payload)
             };
             self.dev_impl.on_dcmd(handle, message_metrics).await
         }

srad-eon/src/metric.rs

@@ -242,6 +242,21 @@ impl MessageMetrics {
     pub fn is_empty(&self) -> bool {
         self.metrics.len() == 0
     }
+
+    /// Create metrics from a payload, using the current time if the payload
+    /// does not include a timestamp.
+    ///
+    /// Some Sparkplug implementations omit the payload timestamp on command
+    /// messages. The command is still actionable, so use the receive time
+    /// instead of rejecting it.
+    #[must_use]
+    pub fn from_payload_lenient(value: Payload) -> Self {
+        let timestamp = value.timestamp.unwrap_or_else(timestamp);
+        Self {
+            timestamp,
+            metrics: value.metrics,
+        }
+    }
 }
 
 /// Errors that can occur when converting a Sparkplug [Metric] into a [MessageMetric].
@@ -397,4 +412,18 @@ mod tests {
             Err(MessageMetricsError::MissingTimestamp)
         ));
     }
+
+    #[test]
+    fn payload_without_timestamp_can_be_read_leniently() {
+        let mut metric = Metric::new();
+        metric.set_name("metric".into());
+        let payload = Payload {
+            metrics: vec![metric],
+            ..Default::default()
+        };
+
+        let metrics = MessageMetrics::from_payload_lenient(payload);
+        assert!(metrics.timestamp > 0);
+        assert_eq!(metrics.len(), 1);
+    }
 }

srad-eon/src/node.rs

@@ -582,6 +582,23 @@ impl Node {
         }
     }
 
+    /// Rebirth the node in response to a rebirth command.
+    ///
+    /// Data publishes are stopped before the birth sequence so that no DATA
+    /// messages are sent after the command is received.
+    async fn command_rebirth(&mut self) {
+        if !self.state.birthed() {
+            return;
+        }
+        self.state.set_dead();
+        if let Err(e) = self.birth(BirthType::Rebirth).await {
+            error!(
+                "Publishing node rebirth message failed. node={}: {e}",
+                self.state.edge_node_id
+            );
+        }
+    }
+
     fn death(&self) {
         self.state.set_dead();
         self.state.bdseq.fetch_add(1, Ordering::SeqCst);
@@ -789,15 +806,23 @@ impl Node {
                 }
             }
 
-            let message_metrics: MessageMetrics = match payload.try_into() {
-                Ok(metrics) => metrics,
-                Err(_) => {
-                    warn!(
-                        "Received invalid CMD payload - ignoring request. node={}",
-                        self.state.edge_node_id
-                    );
-                    return;
+            let message_metrics: MessageMetrics = if payload.timestamp.is_some() {
+                match payload.try_into() {
+                    Ok(metrics) => metrics,
+                    Err(e) => {
+                        warn!(
+                            "Received invalid CMD payload - ignoring request. node={} error={e}",
+                            self.state.edge_node_id
+                        );
+                        return;
+                    }
                 }
+            } else {
+                warn!(
+                    "Received CMD payload without a timestamp; using the current time. node={}",
+                    self.state.edge_node_id
+                );
+                MessageMetrics::from_payload_lenient(payload)
             };
 
             self.metric_manager.on_ncmd(handle, message_metrics).await;
@@ -815,7 +840,7 @@ impl Node {
                     "Got Rebirth CMD - Rebirthing Node. node={}",
                     self.state.edge_node_id
                 );
-                self.rebirth().await;
+                self.command_rebirth().await;
                 self.last_node_rebirth_request = now;
             }
         }