NC | Concurrency & refactoring | Add delay, version move checks and GPFS refactoring

romayalon · romayalon · commit 6cd373142bb5 · 2024-09-29T19:16:37.000+03:00
Signed-off-by: Romy &lt;35330373+romayalon@users.noreply.github.com&gt;
diff --git a/config.js b/config.js
@@ -766,6 +766,7 @@ config.NSFS_SEM_WARNING_TIMEOUT = 10 * 60 * 1000;
 // number of rename retries in case of deleted destination directory
 config.NSFS_RENAME_RETRIES = 10;
 config.NSFS_MKDIR_PATH_RETRIES = 3;
+config.NSFS_DELETE_VERSION_RETRIES = 2;
 
 config.NSFS_VERSIONING_ENABLED = true;
 config.NSFS_UPDATE_ISSUES_REPORT_ENABLED = true;
diff --git a/src/sdk/namespace_fs.js b/src/sdk/namespace_fs.js
@@ -1391,9 +1391,10 @@ class NamespaceFS {
     }
 
     // 1. get latest version_id
-    // 2. if versioning is suspended - 
+    // 2. if versioning is suspended -
     //     2.1 if version ID of the latest version is null - 
-    //       2.1.1 remove latest version
+    //       2.1.1. if it's POSIX backend - unlink the null version
+    //       2.1.2. if it's GPFS backend - nothing to do, the linkatif will override it
     //     2.2 else (version ID of the latest version is unique or there is no latest version) -
     //       2.2.1 remove a version (or delete marker) with null version ID from .versions/ (if exists)
     //  3. if latest version exists -
@@ -1417,10 +1418,8 @@ class NamespaceFS {
                 const versioned_path = latest_ver_info && this._get_version_path(key, latest_ver_info.version_id_str);
                 const versioned_info = latest_ver_info && await this._get_version_info(fs_context, versioned_path);
 
-                gpfs_options = is_gpfs ?
-                    await this._open_files_gpfs(fs_context, new_ver_tmp_path, latest_ver_path, upload_file,
-                        latest_ver_info, open_mode, undefined, versioned_info) :
-                    undefined;
+                gpfs_options = await this._open_files_gpfs(fs_context, new_ver_tmp_path, latest_ver_path, upload_file,
+                    latest_ver_info, open_mode, undefined, versioned_info);
                 const bucket_tmp_dir_path = this.get_bucket_tmpdir_full_path();
                 dbg.log1('Namespace_fs._move_to_dest_version:', latest_ver_info, new_ver_info, gpfs_options);
 
@@ -1439,13 +1438,20 @@ class NamespaceFS {
                 if (latest_ver_info &&
                     ((this._is_versioning_enabled()) ||
                         (this._is_versioning_suspended() && latest_ver_info.version_id_str !== NULL_VERSION_ID))) {
-                    dbg.log1('NamespaceFS._move_to_dest_version version ID of the latest version is a unique ID - the file will be moved it to .versions/ directory');
+                    dbg.log1('NamespaceFS._move_to_dest_version version ID of the latet version is a unique ID - the file will be moved it to .versions/ directory');
+                    await this._check_version_moved(fs_context, key, latest_ver_info.version_id_str, latest_ver_path);
                     await native_fs_utils._make_path_dirs(versioned_path, fs_context);
                     await native_fs_utils.safe_move(fs_context, latest_ver_path, versioned_path, latest_ver_info,
-                        gpfs_options && gpfs_options.move_to_versions, bucket_tmp_dir_path);
+                        gpfs_options?.move_to_versions, bucket_tmp_dir_path);
                 }
                 try {
                     // move new version to latest_ver_path (key path)
+                    if (latest_ver_info) {
+                        await this._check_version_moved(fs_context, key, latest_ver_info.version_id_str, latest_ver_path);
+                    } else {
+                        const latest_ver_info_verification = await this._get_version_info(fs_context, latest_ver_path);
+                        if (latest_ver_info_verification) throw new RpcError('VERSION_MOVED');
+                    }
                     await native_fs_utils.safe_move(fs_context, new_ver_tmp_path, latest_ver_path, new_ver_info,
                         gpfs_options && gpfs_options.move_to_dst, bucket_tmp_dir_path);
                 } catch (err) {
@@ -1460,6 +1466,7 @@ class NamespaceFS {
                 dbg.warn(`NamespaceFS._move_to_dest_version retrying retries=${retries} should_retry=${should_retry}` +
                     ` new_ver_tmp_path=${new_ver_tmp_path} latest_ver_path=${latest_ver_path}`, err);
                 if (!should_retry || retries <= 0) throw err;
+                await P.delay(100);
             } finally {
                 if (gpfs_options) await this._close_files_gpfs(fs_context, gpfs_options.move_to_dst, open_mode);
             }
@@ -2680,9 +2687,18 @@ class NamespaceFS {
         // if stat failed, undefined will return
     }
 
-    // 1. if version exists in .versions/ folder - return its path
-    // 2. else if version is latest version - return latest version path
-    // 3. throw ENOENT error
+    /**
+     * _find_version_path returns the path of the version
+     * 1. if version_id is not defined, it returns the key file
+     * 2. else, 
+     *    2.1. check version format
+     *    2.2. check if latest version exists and it matches the versio_id parameter the latest version path returns
+     *    2.3. else, return the version path under .versions/ 
+     * @param {import('./nb').NativeFSContext} fs_context 
+     * @param {{key: string, version_id?: string}} params
+     * @param {boolean} [return_md_path] 
+     * @returns {Promise<string>}
+     */
     async _find_version_path(fs_context, { key, version_id }, return_md_path) {
         const cur_ver_path = return_md_path ? this._get_file_md_path({ key }) : this._get_file_path({ key });
         if (!version_id) return cur_ver_path;
@@ -2740,7 +2756,7 @@ class NamespaceFS {
      */
     // we can use this function when versioning is enabled or suspended
     async _delete_single_object_versioned(fs_context, key, version_id) {
-        let retries = config.NSFS_RENAME_RETRIES;
+        let retries = config.NSFS_DELETE_VERSION_RETRIES;
         const is_gpfs = native_fs_utils._is_gpfs(fs_context);
         const latest_version_path = this._get_file_path({ key });
         for (;;) {
@@ -2754,15 +2770,15 @@ class NamespaceFS {
 
                 const deleted_latest = file_path === latest_version_path;
                 if (deleted_latest) {
-                    gpfs_options = is_gpfs ?
-                        await this._open_files_gpfs(fs_context, file_path, undefined, undefined, undefined, undefined, true) :
-                        undefined;
+                    gpfs_options = await this._open_files_gpfs(fs_context, file_path, undefined, undefined, undefined, undefined, true);
                     const bucket_tmp_dir_path = this.get_bucket_tmpdir_full_path();
                     await native_fs_utils.safe_unlink(fs_context, file_path, version_info,
                         gpfs_options?.delete_version, bucket_tmp_dir_path);
+                    await this._check_version_moved(fs_context, key, version_id, file_path);
                     return { ...version_info, latest: true };
                 } else {
                     await native_fs_utils.unlink_ignore_enoent(fs_context, file_path);
+                    await this._check_version_moved(fs_context, key, version_id, file_path);
                 }
                 return version_info;
             } catch (err) {
@@ -2776,6 +2792,7 @@ class NamespaceFS {
                 // 3. concurrent delete of this version - will get ENOENT, doing a retry will return successfully 
                 // after we will see that the version was already deleted
                 if (retries <= 0 || !native_fs_utils.should_retry_link_unlink(is_gpfs, err)) throw err;
+                await P.delay(100);
             } finally {
                 if (gpfs_options) await this._close_files_gpfs(fs_context, gpfs_options.delete_version, undefined, true);
             }
@@ -2886,6 +2903,7 @@ class NamespaceFS {
                 }
                 if (err.code !== 'ENOENT') throw err;
                 dbg.warn(`NamespaceFS: _promote_version_to_latest failed retries=${retries}`, err);
+                await P.delay(100);
             }
         }
     }
@@ -2920,19 +2938,17 @@ class NamespaceFS {
 
                 dbg.log1('Namespace_fs._delete_latest_version:', latest_ver_info, versioned_path, versioned_info);
                 if (latest_ver_info) {
-                    gpfs_options = is_gpfs ?
-                    await this._open_files_gpfs(fs_context, latest_ver_path,
-                        undefined, undefined, undefined, undefined, true, versioned_info) :
-                    undefined;
+                    gpfs_options = await this._open_files_gpfs(fs_context, latest_ver_path, undefined, undefined, undefined,
+                        undefined, true, versioned_info);
 
                     const suspended_and_latest_is_not_null = this._is_versioning_suspended() &&
                         latest_ver_info.version_id_str !== NULL_VERSION_ID;
                     const bucket_tmp_dir_path = this.get_bucket_tmpdir_full_path();
                     if (this._is_versioning_enabled() || suspended_and_latest_is_not_null) {
                         await native_fs_utils._make_path_dirs(versioned_path, fs_context);
-                         await native_fs_utils.safe_move(fs_context, latest_ver_path, versioned_path, latest_ver_info,
-                            gpfs_options && gpfs_options.delete_version, bucket_tmp_dir_path);
-                         if (suspended_and_latest_is_not_null) {
+                        await native_fs_utils.safe_move(fs_context, latest_ver_path, versioned_path, latest_ver_info,
+                            gpfs_options?.delete_version, bucket_tmp_dir_path);
+                        if (suspended_and_latest_is_not_null) {
                             // remove a version (or delete marker) with null version ID from .versions/ (if exists)
                             await this._delete_null_version_from_versions_directory(params.key, fs_context);
                         }
@@ -2948,6 +2964,7 @@ class NamespaceFS {
                 retries -= 1;
                 if (retries <= 0 || !native_fs_utils.should_retry_link_unlink(is_gpfs, err)) throw err;
                 dbg.warn(`NamespaceFS._delete_latest_version: Retrying retries=${retries} latest_ver_path=${latest_ver_path}`, err);
+                await P.delay(100);
             } finally {
                 if (gpfs_options) await this._close_files_gpfs(fs_context, gpfs_options.delete_version, undefined, true);
             }
@@ -2965,29 +2982,29 @@ class NamespaceFS {
     // This function removes an object version or delete marker with a null version ID inside .version/ directory 
     async _delete_null_version_from_versions_directory(key, fs_context) {
         const is_gpfs = native_fs_utils._is_gpfs(fs_context);
-        let retries = config.NSFS_RENAME_RETRIES;
         const null_versioned_path = this._get_version_path(key, NULL_VERSION_ID);
         await this._check_path_in_bucket_boundaries(fs_context, null_versioned_path);
-
+        let gpfs_options;
+        let retries = config.NSFS_RENAME_RETRIES;
         for (;;) {
             try {
                 const null_versioned_path_info = await this._get_version_info(fs_context, null_versioned_path);
                 dbg.log1('Namespace_fs._delete_null_version_from_versions_directory:', null_versioned_path, null_versioned_path_info);
-                if (null_versioned_path_info) {
-                    const gpfs_options = is_gpfs ?
-                        await this._open_files_gpfs(fs_context, null_versioned_path, undefined, undefined, undefined, undefined, true) :
-                        undefined;
-                    const bucket_tmp_dir_path = this.get_bucket_tmpdir_full_path();
-                    await native_fs_utils.safe_unlink(fs_context, null_versioned_path, null_versioned_path_info,
-                        gpfs_options?.delete_version, bucket_tmp_dir_path);
+                if (!null_versioned_path_info) return;
 
-                    if (gpfs_options) await this._close_files_gpfs(fs_context, gpfs_options?.delete_version, undefined, true);
-                }
+                gpfs_options = await this._open_files_gpfs(fs_context, null_versioned_path, undefined, undefined, undefined,
+                    undefined, true);
+                const bucket_tmp_dir_path = this.get_bucket_tmpdir_full_path();
+                await native_fs_utils.safe_unlink(fs_context, null_versioned_path, null_versioned_path_info,
+                    gpfs_options?.delete_version, bucket_tmp_dir_path);
                 break;
             } catch (err) {
                 retries -= 1;
                 if (retries <= 0 || !native_fs_utils.should_retry_link_unlink(is_gpfs, err)) throw err;
                 dbg.warn(`NamespaceFS._delete_null_version_from_versions_directory Retrying retries=${retries} null_versioned_path=${null_versioned_path}`, err);
+            } finally {
+                if (gpfs_options) await this._close_files_gpfs(fs_context, gpfs_options.delete_version, undefined, true);
+                await P.delay(100);
             }
         }
     }
@@ -3025,6 +3042,7 @@ class NamespaceFS {
                     return delete_marker_version_id;
                 }
                 dbg.warn(`NamespaceFS: _create_delete_marker failed retries=${retries}`, err);
+                await P.delay(100);
             } finally {
                 if (upload_params) await this.complete_object_upload_finally(undefined, undefined, upload_params.target_file, fs_context);
             }
@@ -3070,6 +3088,8 @@ class NamespaceFS {
     // eslint-disable-next-line max-params
     async _open_files_gpfs(fs_context, src_path, dst_path, upload_or_dir_file, dst_ver_info, open_mode, delete_version, versioned_info) {
         dbg.log1('Namespace_fs._open_files_gpfs:', src_path, src_path && path.dirname(src_path), dst_path, upload_or_dir_file, dst_ver_info, open_mode, delete_version, versioned_info);
+        const is_gpfs = native_fs_utils._is_gpfs(fs_context);
+        if (!is_gpfs) return;
 
         let src_file;
         let dst_file;
@@ -3133,6 +3153,29 @@ class NamespaceFS {
         }
     }
 
+    /**
+     * _check_version_moved recieves cur_path and checks if the version moved
+     * 1. if cur_path is in .versions/ - the function will check if the verison moved to latest
+     * 2. if cur_path is the latest version - the function will check if the version moved to .versions/
+     * @param {import('./nb').NativeFSContext} fs_context 
+     * @param {string} key 
+     * @param {string} version_id 
+     * @param {string} cur_path 
+     */
+    async _check_version_moved(fs_context, key, version_id, cur_path) {
+        const latest_version_path = this._get_file_path({ key });
+        const verisoned_path = this._get_version_path(key, version_id);
+        if (cur_path === latest_version_path) {
+            const verisoned_path_info = await this._get_version_info(fs_context, verisoned_path);
+            if (verisoned_path_info) throw error_utils.new_error_code('VERSION_MOVED', 'version file moved from latest to .versions/, retrying');
+        } else if (cur_path === verisoned_path) {
+            const latest_ver_info = await this._get_version_info(fs_context, latest_version_path);
+            if (latest_ver_info && latest_ver_info.version_id_str === version_id) throw error_utils.new_error_code('VERSION_MOVED', 'version file moved from .versions/ to latest, retrying');
+        } else {
+            throw new RpcError('INTERNAL_ERROR', `namespace_fs._check_version_moved failed, invalid cur_path=${cur_path}`);
+        }
+    }
+
     async _throw_if_storage_class_not_supported(storage_class) {
         if (!await this._is_storage_class_supported(storage_class)) {
             throw new S3Error(S3Error.InvalidStorageClass);
diff --git a/src/test/unit_tests/jest_tests/test_versioning_concurrency.test.js b/src/test/unit_tests/jest_tests/test_versioning_concurrency.test.js
@@ -2,7 +2,6 @@
 'use strict';
 
 const path = require('path');
-const config = require('../../../../config');
 const P = require('../../../util/promise');
 const fs_utils = require('../../../util/fs_utils');
 const NamespaceFS = require('../../../sdk/namespace_fs');
@@ -41,32 +40,31 @@ const nsfs = new NamespaceFS({
 
 const DUMMY_OBJECT_SDK = make_dummy_object_sdk(true);
 describe('test versioning concurrency', () => {
-    const prior_value_of_nsfs_rename_retries = config.NSFS_RENAME_RETRIES;
 
     beforeEach(async () => {
         await fs_utils.create_fresh_path(tmp_fs_path);
     });
 
     afterEach(async () => {
         await fs_utils.folder_delete(tmp_fs_path);
-        config.NSFS_RENAME_RETRIES = prior_value_of_nsfs_rename_retries;
     });
 
     it('multiple puts of the same key', async () => {
         const bucket = 'bucket1';
         const key = 'key1';
         const failed_operations = [];
-        for (let i = 0; i < 5; i++) {
+        const num_of_concurrency = 10;
+        for (let i = 0; i < num_of_concurrency; i++) {
             const random_data = Buffer.from(String(i));
             const body = buffer_utils.buffer_to_read_stream(random_data);
             nsfs.upload_object({ bucket: bucket, key: key, source_stream: body }, DUMMY_OBJECT_SDK)
                 .catch(err => failed_operations.push(err));
         }
-        await P.delay(1000);
+        await P.delay(3000);
         expect(failed_operations.length).toBe(0);
         const versions = await nsfs.list_object_versions({ bucket: bucket }, DUMMY_OBJECT_SDK);
-        expect(versions.objects.length).toBe(5);
-    });
+        expect(versions.objects.length).toBe(num_of_concurrency);
+    }, 6000);
 
     it('multiple delete version id and key', async () => {
         const bucket = 'bucket1';
@@ -163,7 +161,6 @@ describe('test versioning concurrency', () => {
         const failed_put_operations = [];
         const failed_head_operations = [];
         const number_of_iterations = 10;
-        config.NSFS_RENAME_RETRIES = 40;
         for (let i = 0; i < number_of_iterations; i++) {
             const random_data = Buffer.from(String(i));
             const body = buffer_utils.buffer_to_read_stream(random_data);
@@ -181,6 +178,40 @@ describe('test versioning concurrency', () => {
         const versions = await nsfs.list_object_versions({ bucket: bucket }, DUMMY_OBJECT_SDK);
         expect(versions.objects.length).toBe(number_of_iterations + 1); // 1 version before + 10 versions concurrent
     });
+
+    it('concurrent delete objects by version id/latest', async () => {
+        const bucket = 'bucket1';
+        const key = 'key5';
+        const delete_ver_res_arr = [];
+        const delete_ver_err_arr = [];
+        const delete_res_arr = [];
+        const delete_err_arr = [];
+        const initial_num_of_versions = 1;
+        const versions_arr = await _upload_versions(bucket, key, initial_num_of_versions);
+
+        const version_id_to_delete = versions_arr[initial_num_of_versions - 1];
+        const num_of_concurrency = initial_num_of_versions;
+        for (let i = 0; i < num_of_concurrency; i++) {
+            nsfs.delete_object({ bucket: bucket, key: key, version_id: version_id_to_delete }, DUMMY_OBJECT_SDK)
+                .then(res => delete_ver_res_arr.push(res.deleted_version_id))
+                .catch(err => delete_ver_err_arr.push(err));
+            nsfs.delete_object({ bucket: bucket, key: key }, DUMMY_OBJECT_SDK)
+                .then(res => delete_res_arr.push(res))
+                .catch(err => delete_err_arr.push(err));
+        }
+        await P.delay(5000);
+        expect(delete_ver_res_arr).toHaveLength(num_of_concurrency);
+        expect(delete_res_arr).toHaveLength(num_of_concurrency);
+        expect(delete_ver_err_arr).toHaveLength(0);
+        expect(delete_err_arr).toHaveLength(0);
+        const versions = await nsfs.list_object_versions({ bucket: bucket }, DUMMY_OBJECT_SDK);
+
+        expect(versions.objects).toHaveLength(num_of_concurrency);
+        const num_of_delete_markers = (versions.objects.filter(version => version.delete_marker === true)).length;
+        expect(num_of_delete_markers).toBe(num_of_concurrency);
+        const num_of_latest_versions = (versions.objects.filter(version => version.is_latest === true)).length;
+        expect(num_of_latest_versions).toBe(1);
+    }, 6000);
 });
 
 /**
diff --git a/src/util/native_fs_utils.js b/src/util/native_fs_utils.js
