Sync log3 journal files

HP can now sync files in log3 format and provide them via getLog API.

A significant difference from log2 is that a snapshot is not recorded
at the begging of each file, but rather scattered in the whole file.
This saves space but requires to process an extra file when creating a
result in getLog implemenation.

Author: j4r0u53k

Cargo.lock

@@ -3,7 +3,7 @@ name = "historyprovider"
 description = "historyprovider-rs"
 license = "MIT"
 repository = "https://github.com/silicon-heaven/historyprovider-rs"
-version = "2.7.11"
+version = "2.8.0"
 edition = "2024"
 
 [[bin]]
@@ -12,8 +12,8 @@ name = "hp"
 [dependencies]
 tokio = { version = "1.50.0", features = ["io-util", "macros", "net", "rt", "rt-multi-thread", "signal", "sync", "time"], default-features = false }
 shvproto = "6.1.4"
-shvrpc = { version = "14.0", features = ["journal"] }
-shvclient = { version = "3.0", features = ["tokio"] }
+shvrpc = { version = "15.0", features = ["journal"] }
+shvclient = { version = "4.0", features = ["tokio"] }
 futures = "0.3.32"
 log = "0.4.29"
 clap = { version = "4.6.0", features = ["derive"] }
@@ -42,5 +42,5 @@ size = { version = "0.5.0", features = ["serde"] }
 [dev-dependencies]
 async-broadcast = "0.7.2"
 async-trait = "0.1.80"
-shvclient = { version = "3.0", features = ["mocking", "tokio"] }
+shvclient = { version = "4.0", features = ["mocking", "tokio"] }
 tempfile = "3.27.0"

Cargo.toml

@@ -1,5 +1,6 @@
 use log::error;
 use log::info;
+use shvrpc::journalrw::datetime_to_log3_filename;
 use tokio::fs;
 use tokio::io;
 use std::collections::HashMap;
@@ -14,7 +15,7 @@ pub(crate) struct LogFile {
     pub(crate) size: u64,
 }
 
-pub(crate) async fn collect_log2_files(dir: impl AsRef<Path>) -> io::Result<Vec<LogFile>> {
+pub(crate) async fn collect_log_files(dir: impl AsRef<Path>) -> io::Result<Vec<LogFile>> {
     let mut result = Vec::new();
     let mut dirs = vec![dir.as_ref().to_path_buf()];
 
@@ -27,7 +28,7 @@ pub(crate) async fn collect_log2_files(dir: impl AsRef<Path>) -> io::Result<Vec<
 
             if metadata.is_dir() {
                 dirs.push(path);
-            } else if metadata.is_file() && path.extension().is_some_and(|ext| ext == "log2")
+            } else if metadata.is_file() && path.extension().is_some_and(|ext| ext == "log2" || ext == "log3")
                 && let (Some(file_name), Some(parent)) = (path.file_name(), path.parent()) {
                     result.push(LogFile {
                         name: file_name.into(),
@@ -41,9 +42,9 @@ pub(crate) async fn collect_log2_files(dir: impl AsRef<Path>) -> io::Result<Vec<
     Ok(result)
 }
 
-/// Prune `.log2` files while keeping the newest one per directory
-pub(crate) async fn cleanup_log2_files(dir: impl AsRef<Path>, size_limit: u64, days_to_keep: i64) -> io::Result<()> {
-    let files = collect_log2_files(dir).await?;
+/// Prune `.log[2,3]` files while keeping the newest one per directory
+pub(crate) async fn cleanup_log_files(dir: impl AsRef<Path>, size_limit: u64, days_to_keep: i64) -> io::Result<()> {
+    let files = collect_log_files(dir).await?;
     let mut files_size: u64 = files.iter().map(|f| f.size).sum();
 
     info!("log2 files size: {files_size}, size limit: {size_limit}, days_to_keep: {days_to_keep}");
@@ -61,14 +62,26 @@ pub(crate) async fn cleanup_log2_files(dir: impl AsRef<Path>, size_limit: u64, d
     let mut deletable_files = Vec::new();
 
     // This file doesn't have to exist, I'm only constructing the filename for log retention.
-    let oldest_file_to_keep = PathBuf::from(msec_to_log2_filename(shvproto::DateTime::now().add_days(-days_to_keep).epoch_msec()));
-    info!("keeping files younger than {filename}", filename = oldest_file_to_keep.to_string_lossy());
+    let oldest_log2_file_to_keep = PathBuf::from(msec_to_log2_filename(shvproto::DateTime::now().add_days(-days_to_keep).epoch_msec()));
+    let oldest_log3_file_to_keep = PathBuf::from(datetime_to_log3_filename(shvproto::DateTime::now().add_days(-days_to_keep)));
+    info!("keeping files younger than {filename_log2} or {filename_log3}",
+        filename_log2 = oldest_log2_file_to_keep.to_string_lossy(),
+        filename_log3 = oldest_log3_file_to_keep.to_string_lossy(),
+    );
 
     for (_dir, mut group) in grouped {
         // Sort descending (newest first)
         group.sort_by(|a, b| b.name.cmp(&a.name));
-        // Keep the newest file, and keep files newer than oldest_file_to_delete
-        deletable_files.extend(group.into_iter().skip(1).filter(|log_file| log_file.name < oldest_file_to_keep));
+        // Keep the newest two files (for snapshots), and keep files newer than oldest_file_to_delete
+        deletable_files.extend(group
+            .into_iter()
+            .skip(2)
+            .filter(|log_file| if log_file.name.ends_with(".log2") {
+                log_file.name < oldest_log2_file_to_keep
+            } else {
+                log_file.name < oldest_log3_file_to_keep
+            })
+        );
     }
 
     // Sort deletable files (oldest first for deletion)

src/cleanup.rs

@@ -13,12 +13,12 @@ use shvclient::{ClientCommandSender, ClientEventsReceiver};
 use shvproto::DateTime as ShvDateTime;
 use shvrpc::metamethod::AccessLevel;
 use shvrpc::journalentry::JournalEntry;
-use shvrpc::journalrw::{JournalReaderLog2, JournalWriterLog2};
+use shvrpc::journalrw::{JournalReaderLog2, JournalReaderLog3, JournalWriterLog2};
 use shvrpc::datachange::{DataChange, ValueFlags};
 use tokio_util::compat::TokioAsyncReadCompatExt;
 
 use crate::sites::ParsedNotification;
-use crate::util::{get_files, is_log2_file};
+use crate::util::{get_files, is_log_file};
 use crate::State;
 
 
@@ -127,7 +127,7 @@ pub(crate) async fn dirtylog_task(
                 };
 
                 let latest_entry = {
-                    let mut log_files = match get_files(journal_dir.join(&site), is_log2_file).await {
+                    let mut log_files = match get_files(journal_dir.join(&site), is_log_file).await {
                         Ok(files) => files,
                         Err(err) => {
                             error!("Cannot trim dirty log. Cannot read journal dir entries: {err}");
@@ -142,8 +142,13 @@ pub(crate) async fn dirtylog_task(
                             async move {
                                 match tokio::fs::File::open(&file_path).await {
                                     Ok(file) => {
-                                        let reader = JournalReaderLog2::new(BufReader::new(file.compat()));
-                                        reader.fold(None, async |_, entry| entry.ok()).await
+                                        if file_path.ends_with(".log3") {
+                                            let reader = JournalReaderLog3::new(BufReader::new(file.compat()));
+                                            reader.fold(None, async |_, entry| entry.ok()).await
+                                        } else {
+                                            let reader = JournalReaderLog2::new(BufReader::new(file.compat()));
+                                            reader.fold(None, async |_, entry| entry.ok()).await
+                                        }
                                     }
                                     Err(err) => {
                                         error!("Cannot open file {file_path} while getting the last journal entry for trim dirtylog: {err}",
@@ -265,19 +270,19 @@ pub(crate) async fn dirtylog_task(
                 DirtyLogCommand::Get { site, response_tx } => {
                     request_scheduler.schedule_new(site, Request::Get(response_tx));
                 }
-                DirtyLogCommand::ProcessNotification(ParsedNotification { site_path, property_path, signal, param }) => {
+                DirtyLogCommand::ProcessNotification(ParsedNotification { site_path, property_path, signal, source, param }) => {
                     let data_change = DataChange::from(param);
                     let journal_entry = JournalEntry {
                         epoch_msec: data_change.date_time.unwrap_or_else(ShvDateTime::now).epoch_msec(),
                         path: property_path,
                         signal,
-                        source: Default::default(),
+                        source,
                         value: data_change.value,
                         access_level: AccessLevel::Read as _,
                         short_time: data_change.short_time.unwrap_or(-1),
-                        user_id: Default::default(),
+                        user_id: shvproto::RpcValue::null(),
                         repeat: !data_change.value_flags.contains(ValueFlags::SPONTANEOUS),
-                        provisional: true, // data_change.value_flags & (1 << VALUE_FLAG_PROVISIONAL_BIT) != 0,
+                        provisional: true,
                     };
                     // Schedule next task
                     request_scheduler.schedule_new(site_path, Request::Append(journal_entry));
@@ -354,15 +359,15 @@ mod tests {
             TestCase {
                 name: "ProcessNotification: journaldir doesn't exist",
                 steps: &[
-                    Box::new(TestDirtyLogCommand(DirtyLogCommand::ProcessNotification(crate::sites::ParsedNotification { site_path: "site1".into(), property_path: "some_value_node".into(), signal: "chng".into(), param: 20.into() })))
+                    Box::new(TestDirtyLogCommand(DirtyLogCommand::ProcessNotification(crate::sites::ParsedNotification { site_path: "site1".into(), property_path: "some_value_node".into(), signal: "chng".into(), source: "get".into(), param: 20.into() })))
                 ],
                 starting_files: vec![],
                 expected_file_paths: vec![],
             },
             TestCase {
                 name: "ProcessNotification: notifications get written to disk",
                 steps: &[
-                    Box::new(TestDirtyLogCommand(DirtyLogCommand::ProcessNotification(crate::sites::ParsedNotification { site_path: "site1".into(), property_path: "some_value_node".into(), signal: "chng".into(), param: DataChange{
+                    Box::new(TestDirtyLogCommand(DirtyLogCommand::ProcessNotification(crate::sites::ParsedNotification { site_path: "site1".into(), property_path: "some_value_node".into(), signal: "chng".into(), source: "get".into(), param: DataChange{
                         value: 20.into(),
                         date_time: Some(DateTime::from_iso_str("2022-07-07T00:00:00.000").expect("DateTime must work")),
                         value_flags: ValueFlags::empty(),
@@ -417,7 +422,7 @@ mod tests {
                             value: 20.into(),
                             access_level: 8,
                             short_time: -1,
-                            user_id: None,
+                            user_id: ().into(),
                             repeat: true,
                             provisional: true
                         }],
@@ -441,7 +446,7 @@ mod tests {
                             value: 20.into(),
                             access_level: 8,
                             short_time: -1,
-                            user_id: None,
+                            user_id: ().into(),
                             repeat: true,
                             provisional: true
                         }],

src/dirtylog.rs

@@ -14,8 +14,8 @@ use tokio_util::compat::TokioAsyncReadCompatExt;
 
 use crate::dirtylog::DirtyLogCommand;
 use shvrpc::journalentry::JournalEntry;
-use shvrpc::journalrw::{matches_path_pattern, GetLog2Params, GetLog2Since, JournalReaderLog2};
-use crate::util::{get_files, is_log2_file};
+use shvrpc::journalrw::{GetLog2Params, GetLog2Since, JournalReaderLog2, JournalReaderLog3, matches_path_pattern};
+use crate::util::{get_files, is_log_file};
 use crate::State;
 
 pub(crate) struct GetLogResult {
@@ -32,14 +32,17 @@ pub(crate) struct GetLogResult {
 }
 
 fn file_name_to_file_msec(filename: &str) -> Result<i64, String> {
-    let without_ext = filename
-        .strip_suffix(".log2")
-        .ok_or_else(|| format!("Invalid file extension in '{filename}'"))?;
-
-    let datetime = chrono::NaiveDateTime::parse_from_str(without_ext, "%Y-%m-%dT%H-%M-%S-%3f")
-        .map_err(|e| format!("Failed to parse '{filename}': {e}"))?;
+    let (stripped, format) = if let Some(stripped) = filename.strip_suffix(".log2") {
+        (stripped, "%Y-%m-%dT%H-%M-%S-%3f")
+    } else if let Some(stripped) = filename.strip_suffix(".log3") {
+        (stripped, "Y-%m-%dT%H-%M-%S")
+    } else {
+        return Err(format!("Invalid file extension in '{filename}'"));
+    };
 
-    Ok(chrono::Utc.from_utc_datetime(&datetime).timestamp_millis())
+    chrono::NaiveDateTime::parse_from_str(stripped, format)
+        .map(|datetime| chrono::Utc.from_utc_datetime(&datetime).timestamp_millis())
+        .map_err(|e| format!("Failed to parse '{filename}': {e}"))
 }
 
 
@@ -53,7 +56,7 @@ pub(crate) async fn getlog_handler(
     }
     let local_journal_path = Path::new(&app_state.config.journal_dir).join(site_path);
     info!("getLog handler, site: {site_path}, params: {params}");
-    let mut log_files = get_files(&local_journal_path, is_log2_file)
+    let mut log_files = get_files(&local_journal_path, is_log_file)
         .await
         .map_err(|err| RpcError::new(RpcErrorCode::InternalError, format!("Cannot read log files: {err}")))?;
     log_files.sort_by_key(|entry| entry.file_name());
@@ -65,10 +68,11 @@ pub(crate) async fn getlog_handler(
             async move {
                 match tokio::fs::File::open(&file_path).await {
                     Ok(file) => {
-                        JournalReaderLog2::new(BufReader::new(file.compat()))
-                            .next()
-                            .await
-                            .is_some()
+                        if file_path.ends_with(".log3") {
+                            JournalReaderLog3::new(BufReader::new(file.compat())).next().await.is_some()
+                        } else {
+                            JournalReaderLog2::new(BufReader::new(file.compat())).next().await.is_some()
+                        }
                     }
                     Err(err) => {
                         error!("Cannot open file {file_path} in call to getLog: {err}",
@@ -82,7 +86,7 @@ pub(crate) async fn getlog_handler(
         .collect::<Vec<_>>()
         .await;
 
-    let file_start_index = {
+    let file_start_index: usize = {
         if log_files.is_empty() {
             0
         } else {
@@ -102,6 +106,8 @@ pub(crate) async fn getlog_handler(
                 GetLog2Since::LastEntry => log_files.len() - 1,
                 GetLog2Since::None => 0,
             }
+            .saturating_sub(1) // Take previous file to have a complete snapshot when processing log3
+                               // files
         }
     };
 
@@ -366,7 +372,7 @@ mod tests {
             access_level: 32,
             source: "".to_string(),
             value: value.into(),
-            user_id: None,
+            user_id: ().into(),
             repeat: false,
             provisional: false,
         })