hottier deletion bug

Author: parmesant

src/hottier.rs

@@ -857,7 +857,16 @@ impl HotTierManager {
                 );
                 return Ok(false);
             }
-            sht.available_size -= parquet_file.file_size;
+            sht.available_size = if let Some(val) =
+                sht.available_size.checked_sub(parquet_file.file_size)
+            {
+                val
+            } else {
+                tracing::error!(
+                    "file_size > sht.available_size, setting used_available_sizesize to 0 and moving on"
+                );
+                0
+            };
             self.put_hot_tier(stream, &mut sht, tenant_id).await?;
             info!(
                 stream = %stream,
@@ -1113,7 +1122,7 @@ impl HotTierManager {
         path.exists()
     }
 
-    ///delete the parquet file from the hot tier directory for the stream
+    /// delete entire parquet file minute from the hot tier directory for the stream
     /// loop through all manifests in the hot tier directory for the stream
     /// loop through all parquet files in the manifest
     /// check for the oldest entry to delete if the path exists in hot tier
@@ -1155,61 +1164,66 @@ impl HotTierManager {
                 let file = fs::read(manifest_file.path()).await?;
                 let mut manifest: Manifest = serde_json::from_slice(&file)?;
 
-                manifest.files.sort_by_key(|file| file.file_path.clone());
-                manifest.files.reverse();
-
-                'loop_files: while let Some(file_to_delete) = manifest.files.pop() {
-                    let file_size = file_to_delete.file_size;
-                    let path_to_delete = self.hot_tier_path.join(&file_to_delete.file_path);
-
-                    if path_to_delete.exists() {
-                        if let (Some(download_date_time), Some(delete_date_time)) = (
-                            extract_datetime(download_file_path.to_str().unwrap()),
-                            extract_datetime(path_to_delete.to_str().unwrap()),
-                        ) && download_date_time <= delete_date_time
-                        {
-                            info!(
-                                stream = %stream,
-                                tenant = ?tenant_id,
-                                candidate = %file_to_delete.file_path,
-                                target = %download_file_path.display(),
-                                "skip evict: candidate newer than target"
-                            );
-                            delete_successful = false;
-                            break 'loop_files;
-                        }
-
-                        fs::write(manifest_file.path(), serde_json::to_vec(&manifest)?).await?;
+                if manifest.files.is_empty() {
+                    continue;
+                }
 
-                        fs::remove_dir_all(path_to_delete.parent().unwrap()).await?;
-                        delete_empty_directory_hot_tier(
-                            path_to_delete.parent().unwrap().to_path_buf(),
-                        )
-                        .await?;
+                // sort in an ascending manner
+                // idx0: minute=00
+                // idx59: minute=59
+                manifest.files.sort_by_key(|file| file.file_path.clone());
 
-                        stream_hot_tier.used_size -= file_size;
-                        stream_hot_tier.available_size += file_size;
-                        self.put_hot_tier(stream, stream_hot_tier, tenant_id)
-                            .await?;
-                        delete_successful = true;
-                        freed_total += file_size;
+                // get first file's parent (/hottier/stream/date=d/hour=h/minute=m)
+                let first_file = manifest.files.first().unwrap();
+                let first_file_path = self.hot_tier_path.join(&first_file.file_path);
+                let minute_to_delete = first_file_path.parent().unwrap();
+
+                if minute_to_delete.exists() {
+                    if let (Some(download_date_time), Some(delete_date_time)) = (
+                        extract_datetime(download_file_path.to_str().unwrap()),
+                        extract_datetime(first_file_path.to_str().unwrap()),
+                    ) && download_date_time <= delete_date_time
+                    {
                         info!(
                             stream = %stream,
                             tenant = ?tenant_id,
-                            evicted_file = %file_to_delete.file_path,
-                            evicted_size = file_size,
-                            freed_total,
-                            new_available = stream_hot_tier.available_size,
-                            "evicted"
+                            candidate = %minute_to_delete.display(),
+                            target = %download_file_path.display(),
+                            "skip evict: candidate newer than target"
                         );
+                        delete_successful = false;
+                        continue;
+                    }
 
-                        if stream_hot_tier.available_size <= parquet_file_size {
-                            continue 'loop_files;
+                    manifest.files.retain(|file| {
+                        let file_path = self.hot_tier_path.join(&file.file_path);
+                        let file_minute = file_path.parent().unwrap();
+                        if file_minute == minute_to_delete {
+                            // DELETE
+                            stream_hot_tier.used_size = if let Some(val) =
+                                stream_hot_tier.used_size.checked_sub(file.file_size)
+                            {
+                                val
+                            } else {
+                                tracing::error!(
+                                    "file_size > sht.used_size, setting used_size to 0 and moving on"
+                                );
+                                0
+                            };
+                            stream_hot_tier.available_size += file.file_size;
+                            freed_total += file.file_size;
+                            false
                         } else {
-                            break 'loop_dates;
+                            true
                         }
+                    });
+                    fs::write(manifest_file.path(), serde_json::to_vec(&manifest)?).await?;
+                    fs::remove_dir_all(minute_to_delete).await?;
+                    delete_empty_directory_hot_tier(minute_to_delete.to_path_buf()).await?;
+                    if stream_hot_tier.available_size <= parquet_file_size {
+                        continue;
                     } else {
-                        fs::write(manifest_file.path(), serde_json::to_vec(&manifest)?).await?;
+                        break 'loop_dates;
                     }
                 }
             }
@@ -1448,4 +1462,8 @@ pub enum HotTierError {
     HotTierValidationError(#[from] HotTierValidationError),
     #[error("{0}")]
     Anyhow(#[from] anyhow::Error),
+    #[error("{0}")]
+    Underflow(String),
+    #[error("{0}")]
+    Overflow(String),
 }