fs: internal flush fix (#21)

src/filesystem/manifest.rs

@@ -335,6 +335,7 @@ impl Manifest {
                 {
                     if let ChunkLocation::Memory(offset) = location {
                         *location = ChunkLocation::WAL(wal_length_before_flush + *offset);
+                        in_memory_file.wal_chunks.push(start);
                     }
                 }
             }
@@ -347,6 +348,7 @@ impl Manifest {
                     }
                 }
             }
+            in_memory_file.mem_chunks.clear();
         }
 
         memory_buffer.clear();
@@ -358,7 +360,6 @@ impl Manifest {
         // called from main, locks manifest
         // other flush_to_wal gets buffer and others passed in.
         // potentially unify with options.
-
         let mut manifest = self.manifest.write().await;
         let mut memory_buffer = self.memory_buffer.write().await;
         let mut wal_file = self.wal_file.write().await;
@@ -374,6 +375,7 @@ impl Manifest {
                 {
                     if let ChunkLocation::Memory(offset) = location {
                         *location = ChunkLocation::WAL(wal_length_before_flush + *offset);
+                        in_memory_file.wal_chunks.push(start);
                     }
                 }
             }
@@ -386,10 +388,10 @@ impl Manifest {
                     }
                 }
             }
+            in_memory_file.mem_chunks.clear();
         }
 
         memory_buffer.clear();
-
         Ok(())
     }
 
@@ -948,8 +950,13 @@ impl Manifest {
         let mut hash_index = self.hash_index.write().await;
         let mut memory_buffer = self.memory_buffer.write().await;
 
+        let mut to_flush: Vec<(
+            FileIdentifier,
+            Vec<([u8; 32], u64, u64, ChunkLocation, bool)>,
+        )> = Vec::new();
         for (file_id, in_memory_file) in manifest_lock.iter_mut() {
             let mut chunks_to_flush: Vec<([u8; 32], u64, u64, ChunkLocation, bool)> = Vec::new();
+
             for &start in &in_memory_file.mem_chunks {
                 if let Some((hash, length, location, encrypted)) = in_memory_file.chunks.get(&start)
                 {
@@ -979,8 +986,14 @@ impl Manifest {
                     }
                 }
             }
+            if !chunks_to_flush.is_empty() {
+                to_flush.push((file_id.clone(), chunks_to_flush));
+            }
+        }
 
-            for (hash, start, length, location, encrypted) in chunks_to_flush.iter() {
+        for (file_id, chunks) in to_flush.iter() {
+            let in_memory_file = manifest_lock.get_mut(file_id).unwrap();
+            for (hash, start, length, location, encrypted) in chunks.iter() {
                 let total_len = if *encrypted {
                     length + ENCRYPTION_OVERHEAD as u64
                 } else {
@@ -1027,7 +1040,7 @@ impl Manifest {
                     client.put_object(req).await?;
                 } else {
                     let path = self.fs_directory_path.join(hex::encode(hash));
-                    fs::write(path, &buffer).await?;
+                    fs::write(path, buffer).await?;
                 }
                 // add a manifest entry with the new hash and removed wal_position
                 in_memory_file.chunks.insert(
@@ -1039,12 +1052,12 @@ impl Manifest {
                         *encrypted,
                     ),
                 );
-
                 chunk_hashes.insert(*hash, !self.cloud_enabled);
             }
+            in_memory_file.mem_chunks.clear();
+            in_memory_file.wal_chunks.clear();
 
-            if !chunks_to_flush.is_empty() {
+            // chunks have been flushed, let's add a manifest entry.
-                // add updated manifest entries to the manifest file
             let entry = ManifestRecord::Backup(BackupEntry {
                 file: file_id.clone(),
                 chunks: in_memory_file
@@ -1059,6 +1072,7 @@ impl Manifest {
                     })
                     .collect::<Vec<_>>(),
             });
+
             let serialized_entry = bincode::serialize(&entry).unwrap();
             let entry_length = serialized_entry.len() as u64;
 
@@ -1083,12 +1097,9 @@ impl Manifest {
             }
             hash_index.insert(in_memory_file.hash(), file_id.clone());
         }
-        }
-
         // clear the WAL file and memory buffer
         wal_file.set_len(0).await?;
         memory_buffer.clear();
-
         Ok(())
     }
 

src/filesystem/mod.rs

@@ -571,6 +571,7 @@ async fn handle_request(
                     action: "Write".into(),
                 });
             };
+            // println!("fs: got write from {:?} with len {:?}", source.process, &payload.bytes.len());
 
             let file_uuid = FileIdentifier::new_uuid();
             match manifest.write(&file_uuid, &payload.bytes).await {
@@ -712,6 +713,7 @@ async fn handle_request(
                 });
             };
 
+            // println!("setting state for process {:?} with len {:?}", process_id, &payload.bytes.len());
             let file = FileIdentifier::Process(process_id);
             match manifest.write(&file, &payload.bytes).await {
                 Ok(_) => (),