fix: properly flush and sync all writers (#232)

src/annotate/seqvars/mod.rs

@@ -50,6 +50,7 @@ use crate::annotate::seqvars::provider::MehariProvider;
 use crate::common::GenomeRelease;
 
 use crate::db::create::txs::data::TxSeqDatabase;
+use crate::finalize_buf_writer;
 use crate::ped::{PedigreeByName, Sex};
 
 use self::ann::{AnnField, Consequence, FeatureBiotype};
@@ -653,6 +654,12 @@ impl VarFishSeqvarTsvWriter {
         }
     }
 
+    /// Flush buffers.
+    pub fn flush(&mut self) -> Result<(), anyhow::Error> {
+        self.inner.flush()?;
+        Ok(())
+    }
+
     /// Fill `record` coordinate fields.
     ///
     /// # Returns
@@ -1560,11 +1567,21 @@ pub fn run(_common: &crate::common::Args, args: &Args) -> Result<(), anyhow::Err
                     .map(BufWriter::new)
                     .map(BgzfWriter::new)?,
             );
+
             run_with_writer(&mut writer, args)?;
+
+            let bgzf_writer = writer.into_inner();
+            let mut buf_writer = bgzf_writer
+                .finish()
+                .map_err(|e| anyhow::anyhow!("problem finishing BGZF: {}", e))?;
+            finalize_buf_writer!(buf_writer);
         } else {
             let mut writer = VcfWriter::new(File::create(path_output_vcf).map(BufWriter::new)?);
 
             run_with_writer(&mut writer, args)?;
+
+            let mut buf_writer = writer.into_inner();
+            finalize_buf_writer!(buf_writer);
         }
     } else {
         // Load the HGNC xlink map.
@@ -1601,6 +1618,10 @@ pub fn run(_common: &crate::common::Args, args: &Args) -> Result<(), anyhow::Err
 
         writer.set_hgnc_map(hgnc_map);
         run_with_writer(&mut writer, args)?;
+
+        writer
+            .flush()
+            .map_err(|e| anyhow::anyhow!("problem flushing file: {}", e))?;
     }
 
     Ok(())

src/annotate/strucvars/mod.rs

@@ -12,6 +12,7 @@ use std::{fs::File, io::BufWriter};
 
 use crate::common::noodles::{open_vcf_reader, AsyncVcfReader};
 use crate::common::GenomeRelease;
+use crate::finalize_buf_writer;
 use crate::ped::PedigreeByName;
 use annonars::common::cli::CANONICAL;
 use annonars::freqs::cli::import::reading::guess_assembly;
@@ -923,7 +924,7 @@ impl AnnotatedVcfWriter for VarFishStrucvarTsvWriter {
 }
 
 impl VarFishStrucvarTsvWriter {
-    // Create new TSV writer from path.
+    /// Create new TSV writer from path.
     pub fn with_path<P>(p: P) -> Self
     where
         P: AsRef<Path>,
@@ -942,6 +943,12 @@ impl VarFishStrucvarTsvWriter {
             header: None,
         }
     }
+
+    /// Flush buffers.
+    pub fn flush(&mut self) -> Result<(), anyhow::Error> {
+        self.inner.flush()?;
+        Ok(())
+    }
 }
 
 /// Enumeration for describing the orientation of a paired-end read.
@@ -2990,13 +2997,24 @@ pub async fn run(_common: &crate::common::Args, args: &Args) -> Result<(), anyho
             );
             writer.set_assembly(assembly);
             writer.set_pedigree(&pedigree);
+
             run_with_writer(&mut writer, &args, &pedigree, &header).await?;
+
+            let bgzf_writer = writer.into_inner();
+            let mut buf_writer = bgzf_writer
+                .finish()
+                .map_err(|e| anyhow::anyhow!("problem finishing BGZF: {}", e))?;
+            finalize_buf_writer!(buf_writer);
         } else {
             let mut writer =
                 noodles_vcf::Writer::new(File::create(path_output_vcf).map(BufWriter::new)?);
             writer.set_assembly(assembly);
             writer.set_pedigree(&pedigree);
+
             run_with_writer(&mut writer, &args, &pedigree, &header).await?;
+
+            let mut buf_writer = writer.into_inner();
+            finalize_buf_writer!(buf_writer);
         }
     } else {
         let path_output_tsv = args
@@ -3009,6 +3027,10 @@ pub async fn run(_common: &crate::common::Args, args: &Args) -> Result<(), anyho
         writer.set_pedigree(&pedigree);
 
         run_with_writer(&mut writer, &args, &pedigree, &header).await?;
+
+        writer
+            .flush()
+            .map_err(|e| anyhow::anyhow!("problem flushing file: {}", e))?;
     }
 
     Ok(())

src/common/io/std.rs

@@ -94,6 +94,20 @@ where
     Ok(buffer)
 }
 
+/// Given a `BufWriter<File>`, flush buffers and sync the file.
+#[macro_export]
+macro_rules! finalize_buf_writer {
+    ($a:expr) => {
+        $a.flush()
+            .map_err(|e| anyhow::anyhow!("problem flushing buffers: {}", e))?;
+        let file = $a
+            .into_inner()
+            .map_err(|e| anyhow::anyhow!("problem getting inner file: {}", e))?;
+        file.sync_all()
+            .map_err(|e| anyhow::anyhow!("problem syncing file: {}", e))?;
+    };
+}
+
 #[cfg(test)]
 mod test {
     use std::io::Read;