Add remove_incomplete_bank_snapshot_dir

runtime/src/snapshot_utils.rs

@@ -203,7 +203,12 @@ impl BankSnapshotInfo {
         // There is a time window from the slot directory being created, and the content being completely
         // filled.  Check the completion to avoid using a highest found slot directory with missing content.
         let completion_flag_file = bank_snapshot_dir.join(SNAPSHOT_STATE_COMPLETE_FILENAME);
-        if !fs::metadata(&completion_flag_file)?.is_file() {
+        if !completion_flag_file.is_file() {
+            // If the directory is incomplete, it should be removed.
+            // There are also possible hardlink files under <account_path>/snapshot/<slot>/, referred by this
+            // snapshot dir's symlinks.  They are cleaned up in clean_orphaned_account_snapshot_dirs() at the
+            // boot time.
+            fs::remove_dir_all(bank_snapshot_dir)?;
             return Err(SnapshotNewFromDirError::IncompleteDir(completion_flag_file));
         }
 
@@ -5089,8 +5094,13 @@ mod tests {
 
         let complete_flag_file = snapshot.snapshot_dir.join(SNAPSHOT_STATE_COMPLETE_FILENAME);
         fs::remove_file(complete_flag_file).unwrap();
+        // The incomplete snapshot dir should still exist
+        let snapshot_dir_4 = snapshot.snapshot_dir;
+        assert!(snapshot_dir_4.exists());
         let snapshot = get_highest_bank_snapshot(bank_snapshots_dir).unwrap();
         assert_eq!(snapshot.slot, 3);
+        // The incomplete snapshot dir should have been deleted
+        assert!(!snapshot_dir_4.exists());
 
         let snapshot_version_file = snapshot.snapshot_dir.join(SNAPSHOT_VERSION_FILENAME);
         fs::remove_file(snapshot_version_file).unwrap();