fix: only remove file handles once a deleted segment actually drops

Author: marvin-j97

src/compaction/worker.rs

@@ -484,17 +484,6 @@ fn merge_segments(
     levels.show_segments(payload.segment_ids.iter().copied());
     drop(levels);
 
-    log::trace!(
-        "Closing file handles for old segment files: {:?}",
-        payload.segment_ids
-    );
-
-    for segment_id in &payload.segment_ids {
-        opts.config
-            .descriptor_table
-            .remove((opts.tree_id, *segment_id).into());
-    }
-
     log::trace!("Compaction successful");
 
     Ok(())
@@ -548,11 +537,6 @@ fn drop_segments(
         segment.mark_as_deleted();
     }
 
-    for key in segment_ids {
-        log::trace!("Closing file handles for segment data file");
-        opts.config.descriptor_table.remove(*key);
-    }
-
     log::trace!("Dropped {} segments", segment_ids.len());
 
     Ok(())

src/segment/inner.rs

@@ -42,14 +42,18 @@ pub struct Inner {
 
 impl Drop for Inner {
     fn drop(&mut self) {
+        let global_id = (self.tree_id, self.metadata.id).into();
+
         if self.is_deleted.load(std::sync::atomic::Ordering::Acquire) {
             if let Err(e) = std::fs::remove_file(&self.path) {
                 log::warn!(
-                    "Failed to cleanup deleted segment {} at {:?}: {e:?}",
-                    self.metadata.id,
+                    "Failed to cleanup deleted segment {global_id:?} at {:?}: {e:?}",
                     self.path,
                 );
             }
         }
+
+        log::trace!("Closing file handles for old segment file {global_id:?}",);
+        self.descriptor_table.remove(global_id);
     }
 }