encryption: fix key collision issue in tablet snapshot (#15098)

close #15059

- Properly clean up encryption keys when cleaning up snapshot cache file.
- Allow overwriting stale encryption keys when importing remote keys.

Signed-off-by: tabokie <xy.tao@outlook.com>

Co-authored-by: tonyxuqqi <tonyxuqi@outlook.com>

components/encryption/src/manager/mod.rs

@@ -1033,7 +1033,12 @@ impl<'a> DataKeyImporter<'a> {
             if let Entry::Vacant(e) = file_dict.files.entry(fname.to_owned()) {
                 e.insert(file.clone());
             } else {
-                return Err(box_err!("file name collides with existing file: {}", fname));
+                // check for physical file.
+                if Path::new(fname).exists() {
+                    return Err(box_err!("file name collides with existing file: {}", fname));
+                } else {
+                    warn!("overwriting existing unused encryption key"; "fname" => fname);
+                }
             }
             file_dict.files.len() as _
         };
@@ -1825,10 +1830,17 @@ mod tests {
         let mut importer = DataKeyImporter::new(&manager);
         let file0 = manager.new_file("0").unwrap();
 
-        // conflict
+        // conflict with actual file.
+        let f = tmp_dir.path().join("0").to_str().unwrap().to_owned();
+        let _ = manager.new_file(&f).unwrap();
+        File::create(&f).unwrap();
         importer
-            .add("0", file0.iv.clone(), DataKey::default())
+            .add(&f, file0.iv.clone(), DataKey::default())
             .unwrap_err();
+        // conflict with only key.
+        importer
+            .add("0", file0.iv.clone(), DataKey::default())
+            .unwrap();
         // same key
         importer
             .add(

src/server/tablet_snap.rs

@@ -314,6 +314,9 @@ async fn cleanup_cache(
             }
         }
         fs::remove_file(entry.path())?;
+        if let Some(m) = key_manager {
+            m.delete_file(entry.path().to_str().unwrap(), None)?;
+        }
     }
     let mut missing = vec![];
     loop {
@@ -329,7 +332,10 @@ async fn cleanup_cache(
                     continue;
                 }
                 // We should not write to the file directly as it's hard linked.
-                fs::remove_file(p)?;
+                fs::remove_file(&p)?;
+                if let Some(m) = key_manager {
+                    m.delete_file(p.to_str().unwrap(), None)?;
+                }
             }
             missing.push(meta.file_name);
         }
@@ -338,7 +344,10 @@ async fn cleanup_cache(
         }
     }
     for (_, p) in exists {
-        fs::remove_file(p)?;
+        fs::remove_file(&p)?;
+        if let Some(m) = key_manager {
+            m.delete_file(p.to_str().unwrap(), None)?;
+        }
     }
     let mut resp = TabletSnapshotResponse::default();
     resp.mut_files().set_file_name(missing.clone().into());