NSFS | Fix delete last object deletes directory object if its size is 0

Signed-off-by: Romy <35330373+romayalon@users.noreply.github.com>
(cherry picked from commit 79ed147)

Author: romayalon

src/sdk/namespace_fs.js

@@ -2565,12 +2565,32 @@ class NamespaceFS {
         }
     }
 
+    /**
+     * _delete_path_dirs deletes all the paths in the hierarchy that are empty after a successful delete
+     * if the original file_path to be deleted is a regular object which means file_path is not a directory and it's not a directory object path - 
+     * before deletion of the parent directory  - 
+     * if the parent directory is a directory object (has CONTENT_DIR xattr) - stop the deletion loop
+     * else - delete the directory - if dir is not empty it will stop at the first non empty dir
+     * NOTE - the directory object check is needed because when object size is zero we won't create a .folder file and the dir will be empty
+     * therefore the deletion will succeed although we shouldn't delete the directory object
+     * @param {String} file_path 
+     * @param {nb.NativeFSContext} fs_context 
+     */
     async _delete_path_dirs(file_path, fs_context) {
         try {
-            let dir = path.dirname(file_path);
-            while (dir !== this.bucket_path) {
-                await nb_native().fs.rmdir(fs_context, dir);
-                dir = path.dirname(dir);
+            let dir_path = path.dirname(file_path);
+            const deleted_file_is_dir = file_path.endsWith('/');
+            const deleted_file_is_dir_object = file_path.endsWith(config.NSFS_FOLDER_OBJECT_NAME);
+            let should_check_dir_path_is_content_dir = !deleted_file_is_dir && !deleted_file_is_dir_object;
+            while (dir_path !== this.bucket_path) {
+                if (should_check_dir_path_is_content_dir) {
+                    const dir_stat = await nb_native().fs.stat(fs_context, dir_path);
+                    const file_is_disabled_dir_content = dir_stat.xattr && dir_stat.xattr[XATTR_DIR_CONTENT] !== undefined;
+                    if (file_is_disabled_dir_content) break;
+                }
+                await nb_native().fs.rmdir(fs_context, dir_path);
+                dir_path = path.dirname(dir_path);
+                should_check_dir_path_is_content_dir = true;
             }
         } catch (err) {
             if (err.code !== 'ENOTEMPTY' &&
@@ -2999,10 +3019,18 @@ class NamespaceFS {
         return res;
     }
 
-    // delete version_id -
-    // 1. get version info, if it's empty - return
-    // 2. unlink key
-    // 3. if version is latest version - promote second latest -> latest
+    /**
+     * delete version_id does the following - 
+     * 1. get version info, if it's empty - return
+     * 2. unlink key
+     * 3. if version is latest version - promote second latest -> latest
+     * 4. if it's the latest version - delete the directory hirerachy of the key if it's empty
+     *    if it's a past version - delete .versions/ and the directory hirerachy if it's empty
+     * @param {nb.NativeFSContext} fs_context 
+     * @param {String} file_path 
+     * @param {Object} params 
+     * @returns {Promise<{deleted_delete_marker?: string, version_id?: string}>}
+     */
     async _delete_version_id(fs_context, file_path, params) {
         // TODO optimization - GPFS link overrides, no need to unlink before promoting, but if there is nothing to promote we should unlink
         const del_obj_version_info = await this._delete_single_object_versioned(fs_context, params.key, params.version_id);

src/test/unit_tests/test_namespace_fs.js

@@ -1200,6 +1200,36 @@ mocha.describe('namespace_fs folders tests', function() {
 
         });
 
+        mocha.it('delete inner object in directory object size 0 - no .folder file but directory still exists', async function() {
+            const inner_key = '/inner_obj';
+            const key = upload_key_3 + inner_key;
+            const source = buffer_utils.buffer_to_read_stream(data);
+            await upload_object(ns_tmp, upload_bkt, key, dummy_object_sdk, source);
+            const p1 = path.join(ns_tmp_bucket_path, upload_key_3);
+            const p2 = path.join(ns_tmp_bucket_path, key);
+            await fs_utils.file_must_not_exist(path.join(p1, config.NSFS_FOLDER_OBJECT_NAME));
+            const full_xattr1 = await get_xattr(p1);
+            assert.deepEqual(full_xattr1, { ...user_md_and_dir_content_xattr, [XATTR_DIR_CONTENT]: obj_sizes_map[upload_key_3] });
+            await ns_tmp.delete_object({ bucket: upload_bkt, key: key }, dummy_object_sdk);
+            await fs_utils.file_must_exist(p1);
+            await fs_utils.file_must_not_exist(p2);
+        });
+
+        mocha.it('delete inner directory object size > 0 in directory object size 0 - no .folder file but directory still exists', async function() {
+            const inner_dir_obj_key = '/inner_dir_obj_key';
+            const key = upload_key_3 + inner_dir_obj_key;
+            const source = buffer_utils.buffer_to_read_stream(data);
+            await upload_object(ns_tmp, upload_bkt, key, dummy_object_sdk, source);
+            const p1 = path.join(ns_tmp_bucket_path, upload_key_3);
+            const p2 = path.join(ns_tmp_bucket_path, key);
+            await fs_utils.file_must_not_exist(path.join(p1, config.NSFS_FOLDER_OBJECT_NAME));
+            const full_xattr1 = await get_xattr(p1);
+            assert.deepEqual(full_xattr1, { ...user_md_and_dir_content_xattr, [XATTR_DIR_CONTENT]: obj_sizes_map[upload_key_3] });
+            await ns_tmp.delete_object({ bucket: upload_bkt, key: key }, dummy_object_sdk);
+            await fs_utils.file_must_exist(p1);
+            await fs_utils.file_must_not_exist(p2);
+        });
+
         mocha.it('delete object content 0 - no .folder file', async function() {
             const p1 = path.join(ns_tmp_bucket_path, upload_key_3);
             const full_xattr1 = await get_xattr(p1);