Make entity store persistent #2428

Persist the entity store as a JSON lines file.
Every registration message and twin data message is persisted as JSON lines.
On startup the in-memory entity store is rebuilt by replaying these messages.

crates/common/mqtt_channel/Cargo.toml

@@ -23,4 +23,5 @@ zeroize = { workspace = true }
 [dev-dependencies]
 anyhow = { workspace = true }
 mqtt_tests = { workspace = true }
+serde_json = { workspace = true }
 serial_test = { workspace = true }

crates/common/mqtt_channel/src/messages.rs

@@ -2,20 +2,43 @@ use crate::errors::MqttError;
 use crate::topics::Topic;
 use rumqttc::Publish;
 use rumqttc::QoS;
+use serde::Deserialize;
+use serde::Serialize;
 use std::fmt::Debug;
 use std::fmt::Formatter;
 use std::fmt::Write;
 
 /// A message to be sent to or received from MQTT.
-#[derive(Debug, Clone, Eq, PartialEq)]
+#[derive(Debug, Clone, Eq, PartialEq, Serialize, Deserialize)]
 pub struct Message {
     pub topic: Topic,
     pub payload: DebugPayload,
+    #[serde(serialize_with = "serialize_qos", deserialize_with = "deserialize_qos")]
     pub qos: QoS,
     pub retain: bool,
 }
 
-#[derive(Clone, Eq, PartialEq)]
+fn serialize_qos<S>(qos: &QoS, serializer: S) -> Result<S::Ok, S::Error>
+where
+    S: serde::Serializer,
+{
+    (*qos as u8).serialize(serializer)
+}
+
+fn deserialize_qos<'de, D>(deserializer: D) -> Result<QoS, D::Error>
+where
+    D: serde::Deserializer<'de>,
+{
+    let value = u8::deserialize(deserializer)?;
+    match value {
+        0 => Ok(QoS::AtMostOnce),
+        1 => Ok(QoS::AtLeastOnce),
+        2 => Ok(QoS::ExactlyOnce),
+        _ => Err(serde::de::Error::custom("Invalid QoS value")),
+    }
+}
+
+#[derive(Clone, Eq, PartialEq, Serialize, Deserialize)]
 pub struct DebugPayload(Payload);
 
 impl Debug for DebugPayload {
@@ -188,4 +211,18 @@ mod tests {
             "Invalid UTF8 payload: invalid utf-8 sequence of 1 bytes from index 0: ..."
         );
     }
+
+    #[test]
+    fn message_serialize_deserialize() {
+        let message = Message {
+            topic: Topic::new("test").unwrap(),
+            payload: DebugPayload("payload".as_bytes().to_vec()),
+            qos: QoS::AtMostOnce,
+            retain: true,
+        };
+
+        let json = serde_json::to_string(&message).expect("Serialization failed");
+        let deserialized: Message = serde_json::from_str(&json).expect("Deserialization failed");
+        assert_eq!(deserialized, message);
+    }
 }

crates/common/tedge_config/src/tedge_config_cli/tedge_config.rs

@@ -700,6 +700,13 @@ define_tedge_config! {
         path: Utf8PathBuf,
     },
 
+    db: {
+        /// The directory used to store any persistent data that survives tedge updates and even firmware updates
+        #[tedge_config(example = "/data/tedge", default(value = "/data/tedge"))]
+        #[doku(as = "PathBuf")]
+        path: Utf8PathBuf,
+    },
+
     data: {
         /// The directory used to store data like cached files, runtime metadata, etc.
         #[tedge_config(example = "/var/tedge", default(value = "/var/tedge"))]

crates/core/c8y_api/src/json_c8y.rs

@@ -404,6 +404,7 @@ mod tests {
     use tedge_api::event::ThinEdgeEventData;
     use tedge_api::messages::SoftwareListCommandPayload;
     use tedge_api::mqtt_topics::EntityTopicId;
+    use tedge_api::mqtt_topics::MqttSchema;
     use test_case::test_case;
     use time::macros::datetime;
 
@@ -768,11 +769,16 @@ mod tests {
         ;"convert to clear alarm"
     )]
     fn check_alarm_translation(tedge_alarm: ThinEdgeAlarm, expected_c8y_alarm: C8yAlarm) {
+        let temp_dir = tempfile::tempdir().unwrap();
         let main_device = EntityRegistrationMessage::main_device("test-main".into());
-        let mut entity_store = EntityStore::with_main_device(
+        let mut entity_store = EntityStore::with_main_device_and_default_service_type(
+            MqttSchema::default(),
             main_device,
+            "service".into(),
             dummy_external_id_mapper,
             dummy_external_id_validator,
+            5,
+            &temp_dir,
         )
         .unwrap();
 
@@ -800,11 +806,16 @@ mod tests {
             }),
         };
 
+        let temp_dir = tempfile::tempdir().unwrap();
         let main_device = EntityRegistrationMessage::main_device("test-main".into());
-        let entity_store = EntityStore::with_main_device(
+        let entity_store = EntityStore::with_main_device_and_default_service_type(
+            MqttSchema::default(),
             main_device,
+            "service".into(),
             dummy_external_id_mapper,
             dummy_external_id_validator,
+            5,
+            &temp_dir,
         )
         .unwrap();
 

crates/core/tedge/src/cli/init.rs

@@ -172,6 +172,15 @@ impl TEdgeInitCmd {
             ),
         )?;
 
+        create_directory(
+            &config.db.path,
+            PermissionEntry::new(
+                Some(self.user.clone()),
+                Some(self.group.clone()),
+                Some(0o775),
+            ),
+        )?;
+
         Ok(())
     }
 }

crates/core/tedge_api/Cargo.toml

@@ -35,6 +35,7 @@ clock = { workspace = true }
 maplit = { workspace = true }
 mockall = { workspace = true }
 regex = { workspace = true }
+tempfile = { workspace = true }
 test-case = { workspace = true }
 time = { workspace = true, features = ["macros"] }
 toml = { workspace = true }