Merge pull request #3614 from jarhodes314/refactor/flush-file-buffers

refactor: flush after calling write_all

Author: jarhodes314

crates/common/certificate/src/parse_root_certificate/mod.rs

@@ -282,16 +282,18 @@ mod tests {
         let temp_dir = TempDir::new().unwrap();
 
         // Add a first chain with 1 certificate
-        let mut cert_1 = File::create(temp_dir.path().join("cert_1")).unwrap();
-        cert_1
-            .write_all(include_str!("../test_root_cert_1.txt").as_bytes())
-            .unwrap();
+        fs::write(
+            temp_dir.path().join("cert_1"),
+            include_str!("../test_root_cert_1.txt"),
+        )
+        .unwrap();
 
         // Add a second chain with 2 certificates
-        let mut cert_2 = File::create(temp_dir.path().join("cert_2")).unwrap();
-        cert_2
-            .write_all(include_str!("../test_root_cert_2.txt").as_bytes())
-            .unwrap();
+        fs::write(
+            temp_dir.path().join("cert_2"),
+            include_str!("../test_root_cert_2.txt"),
+        )
+        .unwrap();
 
         let root_certs = new_root_store(temp_dir.path()).unwrap();
         assert_eq!(root_certs.len(), 3);
@@ -302,17 +304,19 @@ mod tests {
         let temp_dir = TempDir::new().unwrap();
 
         // Add a first chain with 1 certificate
-        let mut cert_1 = File::create(temp_dir.path().join("cert_1")).unwrap();
-        cert_1
-            .write_all(include_str!("../test_root_cert_1.txt").as_bytes())
-            .unwrap();
+        fs::write(
+            temp_dir.path().join("cert_1"),
+            include_str!("../test_root_cert_1.txt"),
+        )
+        .unwrap();
 
         // Add a second chain with 2 certificates in a sub directory
         fs::create_dir(temp_dir.path().join("sub_certs")).unwrap();
-        let mut cert_2 = File::create(temp_dir.path().join("sub_certs/cert_2")).unwrap();
-        cert_2
-            .write_all(include_str!("../test_root_cert_2.txt").as_bytes())
-            .unwrap();
+        fs::write(
+            temp_dir.path().join("sub_certs/cert_2"),
+            include_str!("../test_root_cert_2.txt"),
+        )
+        .unwrap();
 
         let root_certs = new_root_store(temp_dir.path()).unwrap();
         assert_eq!(root_certs.len(), 3);

crates/common/download/src/download.rs

@@ -415,6 +415,7 @@ async fn save_chunks_to_file_at(
     while let Some(bytes) = response.chunk().await? {
         writer.write_all(&bytes)?;
     }
+    writer.flush()?;
     Ok(())
 }
 
@@ -800,6 +801,7 @@ mod tests {
                         let msg = format!("{header}\r\n{size}\r\n{body}\r\n");
                         debug!("sending message = {msg}");
                         writer.write_all(msg.as_bytes()).await.unwrap();
+                        writer.flush().await.unwrap();
                     } else {
                         let header = "\
                             HTTP/1.1 200 OK\r\n\
@@ -811,6 +813,7 @@ mod tests {
                         let body = "AAAA";
                         let msg = format!("{header}\r\n4\r\n{body}\r\n");
                         writer.write_all(msg.as_bytes()).await.unwrap();
+                        writer.flush().await.unwrap();
                     }
                 };
                 tokio::spawn(response_task);
@@ -1005,6 +1008,7 @@ mod tests {
         }
 
         file.write_all(buffer.as_bytes()).unwrap();
+        file.flush().unwrap();
 
         Ok(file)
     }

crates/common/tedge_config/src/system_toml/log_level.rs

@@ -87,7 +87,6 @@ pub fn set_log_level(log_level: tracing::Level) {
 mod tests {
     use super::*;
     use camino::Utf8PathBuf;
-    use std::io::Write;
     use tempfile::TempDir;
     use tracing::Level;
 
@@ -154,8 +153,7 @@ mod tests {
         let temp_dir = TempDir::new()?;
         let config_root = temp_dir.path().to_path_buf();
         let config_file_path = config_root.join("system.toml");
-        let mut file = std::fs::File::create(config_file_path.as_path())?;
-        file.write_all(content.as_bytes())?;
+        std::fs::write(config_file_path.as_path(), content.as_bytes())?;
         Ok((temp_dir, config_root.try_into().unwrap()))
     }
 }

crates/common/tedge_config/src/system_toml/mod.rs

@@ -89,7 +89,6 @@ impl Default for SystemSpecificCommands {
 mod tests {
     use super::*;
     use camino::Utf8PathBuf;
-    use std::io::Write;
     use tempfile::TempDir;
 
     #[test]
@@ -196,8 +195,7 @@ mod tests {
         let temp_dir = TempDir::new()?;
         let config_root = Utf8Path::from_path(temp_dir.path()).unwrap().to_owned();
         let config_file_path = config_root.join(SYSTEM_CONFIG_FILE);
-        let mut file = std::fs::File::create(config_file_path.as_path())?;
-        file.write_all(content.as_bytes())?;
+        std::fs::write(config_file_path.as_path(), content.as_bytes())?;
         Ok((temp_dir, config_root))
     }
 }

crates/common/tedge_utils/src/file.rs

@@ -314,18 +314,19 @@ impl PermissionEntry {
         match options.create_new(true).write(true).open(file).await {
             Ok(mut f) => {
                 self.clone().apply(file).await?;
-                f.sync_all().await.map_err(|from| FileError::FailedToSync {
-                    file: file.to_path_buf(),
-                    from,
-                })?;
                 if let Some(default_content) = default_content {
                     f.write_all(default_content.as_bytes())
                         .map_err(|e| FileError::WriteContentFailed {
                             file: file.display().to_string(),
                             from: e,
                         })
                         .await?;
+                    f.flush().await?;
                 }
+                f.sync_all().await.map_err(|from| FileError::FailedToSync {
+                    file: file.to_path_buf(),
+                    from,
+                })?;
                 Ok(())
             }
 
@@ -347,18 +348,19 @@ pub async fn overwrite_file(file: &Path, content: &str) -> Result<(), FileError>
         .await
     {
         Ok(mut f) => {
-            f.sync_all()
-                .map_err(|from| FileError::FailedToSync {
-                    file: file.to_path_buf(),
-                    from,
-                })
-                .await?;
             f.write_all(content.as_bytes())
                 .map_err(|e| FileError::WriteContentFailed {
                     file: file.display().to_string(),
                     from: e,
                 })
                 .await?;
+            f.flush().await?;
+            f.sync_all()
+                .map_err(|from| FileError::FailedToSync {
+                    file: file.to_path_buf(),
+                    from,
+                })
+                .await?;
             Ok(())
         }
         Err(e) => Err(FileError::FileCreateFailed {

crates/common/tedge_utils/src/paths.rs

@@ -6,6 +6,8 @@ use std::path::PathBuf;
 
 use async_tempfile::TempFile;
 use tokio::io::AsyncWrite;
+use tokio::io::AsyncWriteExt;
+use tokio::io::BufWriter;
 
 #[derive(thiserror::Error, Debug)]
 pub enum PathsError {
@@ -43,8 +45,13 @@ pub fn create_directories(dir_path: impl AsRef<Path>) -> Result<(), PathsError>
         .map_err(|error| PathsError::DirCreationFailed(error, dir_path.into()))
 }
 
-pub async fn persist_tempfile(file: TempFile, path_to: impl AsRef<Path>) -> Result<(), PathsError> {
-    tokio::fs::rename(file.file_path(), &path_to)
+pub async fn persist_tempfile(
+    mut file: BufWriter<TempFile>,
+    path_to: impl AsRef<Path>,
+) -> Result<(), PathsError> {
+    file.flush().await?;
+    file.get_ref().sync_all().await?;
+    tokio::fs::rename(file.get_ref().file_path(), &path_to)
         .await
         .map_err(|error| PathsError::FileCreationFailed(error, path_to.as_ref().into()))?;
 
@@ -65,7 +72,7 @@ pub fn ok_if_not_found(err: std::io::Error) -> std::io::Result<()> {
 #[pin_project::pin_project]
 pub struct DraftFile {
     #[pin]
-    file: TempFile,
+    file: BufWriter<TempFile>,
     target: PathBuf,
     mode: Option<u32>,
 }
@@ -82,7 +89,7 @@ impl DraftFile {
             .ok_or_else(|| PathsError::ParentDirNotFound {
                 path: target.as_os_str().into(),
             })?;
-        let file = TempFile::new_in(dir).await?;
+        let file = BufWriter::new(TempFile::new_in(dir).await?);
 
         let target = target.to_path_buf();
 

crates/common/upload/src/upload.rs

@@ -531,5 +531,6 @@ mod tests {
         }
 
         file.write_all(buffer.as_bytes()).await.unwrap();
+        file.flush().await.unwrap();
     }
 }

crates/core/plugin_sm/src/plugin.rs

@@ -466,7 +466,8 @@ impl Plugin for ExternalPluginCommand {
                 }
             };
 
-            child_stdin.write_all(action.as_bytes()).await?
+            child_stdin.write_all(action.as_bytes()).await?;
+            child_stdin.flush().await?;
         }
 
         let output = child.wait_with_output(command_log).await?;

crates/core/plugin_sm/tests/plugin.rs

@@ -4,7 +4,6 @@ mod tests {
     use plugin_sm::plugin::deserialize_module_info;
     use plugin_sm::plugin::sm_path;
     use plugin_sm::plugin::ExternalPluginCommand;
-    use std::io::Write;
     use std::path::Path;
     use std::path::PathBuf;
     use tedge_api::SoftwareError;
@@ -206,8 +205,7 @@ mod tests {
         let toml_conf = &format!("[software]\nmax_packages = {max_packages}");
 
         let config_location = TEdgeConfigLocation::from_custom_root(dir.path());
-        let mut file = std::fs::File::create(config_location.tedge_config_file_path())?;
-        file.write_all(toml_conf.as_bytes())?;
+        std::fs::write(config_location.tedge_config_file_path(), toml_conf)?;
         Ok(dir)
     }
 }

crates/core/tedge/src/bridge/common_mosquitto_config.rs

@@ -137,6 +137,7 @@ impl CommonMosquittoConfig {
 
         self.internal_listener.write(writer).await?;
         self.external_listener.write(writer).await?;
+        writer.flush().await?;
 
         Ok(())
     }

crates/core/tedge/src/bridge/config.rs

@@ -135,6 +135,8 @@ impl BridgeConfig {
             writeln_async!(writer, "topic {}", topic)?;
         }
 
+        writer.flush().await?;
+
         Ok(())
     }
 

crates/core/tedge/src/cli/certificate/create.rs

@@ -169,6 +169,7 @@ async fn persist_private_key(
     // Make sure the key is secret, before write
     File::set_permissions(&key_file, Permissions::from_mode(0o600)).await?;
     key_file.write_all(cert_key.as_bytes()).await?;
+    key_file.flush().await?;
     key_file.sync_all().await?;
 
     // Prevent the key to be overwritten
@@ -178,6 +179,7 @@ async fn persist_private_key(
 
 async fn persist_public_key(mut key_file: File, cert_pem: String) -> Result<(), std::io::Error> {
     key_file.write_all(cert_pem.as_bytes()).await?;
+    key_file.flush().await?;
     key_file.sync_all().await?;
 
     // Make the file public

crates/core/tedge/tests/mqtt.rs

@@ -3,7 +3,6 @@ mod tests {
 
     // These test cases need mosquitto on localhost on GH hosted machine.
 
-    use std::io::Write;
     use std::time::Duration;
 
     use assert_cmd::assert::OutputAssertExt;
@@ -19,8 +18,10 @@ mod tests {
         let toml_conf = &format!("[mqtt]\nclient_port = {port}");
 
         let config_location = TEdgeConfigLocation::from_custom_root(dir.path());
-        let mut file = std::fs::File::create(config_location.tedge_config_file_path())?;
-        file.write_all(toml_conf.as_bytes())?;
+        std::fs::write(
+            config_location.tedge_config_file_path(),
+            toml_conf.as_bytes(),
+        )?;
         Ok(dir)
     }
 

crates/core/tedge_agent/src/state_repository/state.rs

@@ -4,8 +4,6 @@ use log::info;
 use log::warn;
 use serde::de::DeserializeOwned;
 use serde::Serialize;
-use std::fs::File;
-use std::io::Write;
 use std::marker::PhantomData;
 use tedge_utils::fs::atomically_write_file_async;
 use tokio::fs;
@@ -29,9 +27,9 @@ pub fn agent_default_state_dir(tedge_root: Utf8PathBuf) -> Utf8PathBuf {
 pub fn agent_state_dir(state_dir: Utf8PathBuf, tedge_root: Utf8PathBuf) -> Utf8PathBuf {
     // Check that the given directory is actually writable, by creating an empty test file
     let test_file = state_dir.join(state_dir.join(".--test--"));
-    match File::create(test_file.clone()).and_then(|mut file| file.write_all(b"")) {
+    match std::fs::write(&test_file, "") {
         Ok(_) => {
-            let _ = std::fs::remove_file(test_file);
+            let _ = std::fs::remove_file(&test_file);
             state_dir
         }
         Err(err) => {

crates/extensions/tedge_log_manager/src/manager/log_utils.rs

@@ -37,11 +37,14 @@ pub fn new_read_logs(
                 break;
             }
             Err(error) => {
+                temp_file.flush()?;
                 return Err(error);
             }
         };
     }
 
+    temp_file.flush()?;
+
     Ok(temp_path)
 }
 
@@ -329,6 +332,7 @@ mod tests {
         let data = "this is the first line.\nthis is the second line.\nthis is the third line.\nthis is the forth line.\nthis is the fifth line.";
 
         log_file.write_all(data.as_bytes()).unwrap();
+        log_file.flush().unwrap();
 
         let line_counter = 0;
         let max_lines = 4;
@@ -381,6 +385,7 @@ mod tests {
             let data = &format!("this is the first line of {file_name}.\nthis is the second line of {file_name}.\nthis is the third line of {file_name}.\nthis is the forth line of {file_name}.\nthis is the fifth line of {file_name}.");
 
             log_file.write_all(data.as_bytes()).unwrap();
+            log_file.flush().unwrap();
 
             let new_mtime = FileTime::from_unix_time(m_time, 0);
             set_file_mtime(file_path, new_mtime).unwrap();

crates/extensions/tedge_uploader_ext/src/tests.rs

@@ -1,5 +1,3 @@
-use std::fs::File;
-use std::io::Write;
 use std::time::Duration;
 
 use camino::Utf8Path;
@@ -42,8 +40,7 @@ async fn upload_without_auth() -> Result<(), DynError> {
         .unwrap()
         .join("file_to_upload.txt");
 
-    let mut tmp_file = File::create(&target_path).unwrap();
-    tmp_file.write_all(b"Hello, world!").unwrap();
+    std::fs::write(&target_path, "Hello, world!").unwrap();
 
     let server_url = server.url();
     let download_request = UploadRequest::new(&server_url, &target_path);
@@ -80,8 +77,7 @@ async fn upload_with_auth() -> Result<(), DynError> {
         .unwrap()
         .join("file_to_upload.txt");
 
-    let mut tmp_file = File::create(&target_path).unwrap();
-    tmp_file.write_all(b"Hello, world!").unwrap();
+    std::fs::write(&target_path, "Hello, world!").unwrap();
 
     let server_url = server.url();
     let download_request =