NSFS | fix copy_object issues

Signed-off-by: nadav mizrahi <[email protected]>

Author: nadavMiz

src/sdk/namespace_fs.js

@@ -483,7 +483,7 @@ class NamespaceFS {
     }
 
     /**
-     * @param {nb.ObjectSDK} object_sdk 
+     * @param {nb.ObjectSDK} object_sdk
      * @returns {nb.NativeFSContext}
      */
     prepare_fs_context(object_sdk) {
@@ -1090,7 +1090,7 @@ class NamespaceFS {
             // end the stream
             res.end();
 
-            await stream_utils.wait_finished(res, { signal: object_sdk.abort_controller.signal });
+            await stream_utils.wait_finished(res, { readable: false, signal: object_sdk.abort_controller.signal });
             object_sdk.throw_if_aborted();
 
             dbg.log0('NamespaceFS: read_object_stream completed file', file_path, {
@@ -1209,9 +1209,7 @@ class NamespaceFS {
         }
 
         if (copy_res) {
-            if (copy_res === copy_status_enum.FALLBACK) {
-                params.copy_source.nsfs_copy_fallback();
-            } else {
+            if (copy_res !== copy_status_enum.FALLBACK) {
                 // open file after copy link/same inode should use read open mode
                 open_mode = config.NSFS_OPEN_READ_MODE;
                 if (copy_res === copy_status_enum.SAME_INODE) open_path = file_path;
@@ -1294,10 +1292,8 @@ class NamespaceFS {
         let stat = await target_file.stat(fs_context);
         this._verify_encryption(params.encryption, this._get_encryption_info(stat));
 
-        // handle xattr
-        // assign user xattr on non copy / copy with xattr_copy header provided
         const copy_xattr = params.copy_source && params.xattr_copy;
-        let fs_xattr = copy_xattr ? undefined : to_fs_xattr(params.xattr);
+        let fs_xattr = to_fs_xattr(params.xattr);
 
         // assign noobaa internal xattr - content type, md5, versioning xattr
         if (params.content_type) {
@@ -1511,7 +1507,7 @@ class NamespaceFS {
     // Can be finetuned further on if needed and inserting the Semaphore logic inside
     // Instead of wrapping the whole _upload_stream function (q_buffers lives outside of the data scope of the stream)
     async _upload_stream({ fs_context, params, target_file, object_sdk, offset }) {
-        const { source_stream } = params;
+        const { source_stream, copy_source } = params;
         try {
             // Not using async iterators with ReadableStreams due to unsettled promises issues on abort/destroy
             const md5_enabled = this._is_force_md5_enabled(object_sdk);
@@ -1526,8 +1522,12 @@ class NamespaceFS {
                 large_buf_size: multi_buffer_pool.get_buffers_pool(undefined).buf_size
             });
             chunk_fs.on('error', err1 => dbg.error('namespace_fs._upload_stream: error occured on stream ChunkFS: ', err1));
-            await stream_utils.pipeline([source_stream, chunk_fs]);
-            await stream_utils.wait_finished(chunk_fs);
+            if (copy_source) {
+                await this.read_object_stream(copy_source, object_sdk, chunk_fs);
+            } else {
+                await stream_utils.pipeline([source_stream, chunk_fs]);
+                await stream_utils.wait_finished(chunk_fs);
+            }
             return { digest: chunk_fs.digest, total_bytes: chunk_fs.total_bytes };
         } catch (error) {
             dbg.error('_upload_stream had error: ', error);
@@ -1813,6 +1813,7 @@ class NamespaceFS {
             upload_params.params.xattr = create_params_parsed.xattr;
             upload_params.params.storage_class = create_params_parsed.storage_class;
             upload_params.digest = MD5Async && (((await MD5Async.digest()).toString('hex')) + '-' + multiparts.length);
+            upload_params.params.content_type = create_params_parsed.content_type;
 
             const upload_info = await this._finish_upload(upload_params);
 

src/sdk/object_sdk.js

@@ -106,8 +106,8 @@ class ObjectSDK {
      * in order to handle aborting requests gracefully. The `abort_controller` member will
      * be used to signal async flows that abort was detected.
      * @see {@link https://nodejs.org/docs/latest/api/globals.html#class-abortcontroller}
-     * @param {import('http').IncomingMessage} req 
-     * @param {import('http').ServerResponse} res 
+     * @param {import('http').IncomingMessage} req
+     * @param {import('http').ServerResponse} res
      */
     setup_abort_controller(req, res) {
         res.once('error', err => {
@@ -158,7 +158,7 @@ class ObjectSDK {
     }
 
     /**
-     * @param {string} name 
+     * @param {string} name
      * @returns {Promise<nb.Namespace>}
      */
     async _get_bucket_namespace(name) {
@@ -268,7 +268,7 @@ class ObjectSDK {
         return Boolean(fs_root_path || fs_root_path === '');
     }
 
-    // validates requests for non nsfs buckets from accounts which are nsfs_only 
+    // validates requests for non nsfs buckets from accounts which are nsfs_only
     has_non_nsfs_bucket_access(account, ns) {
         dbg.log1('validate_non_nsfs_bucket: ', account, ns?.write_resource?.resource);
         if (!account) return false;
@@ -524,7 +524,7 @@ class ObjectSDK {
     /**
      * Calls the op and report time and error to stats collector.
      * on_success can be added to update read/write stats (but on_success shouln't throw)
-     * 
+     *
      * @template T
      * @param {{
      *      op_name: string;
@@ -642,7 +642,7 @@ class ObjectSDK {
             params.content_type = source_md.content_type;
         }
         try {
-            if (params.xattr) params.xattr = _.omitBy(params.xattr, (val, name) => name.startsWith('noobaa-namespace'));
+            if (params.xattr) params.xattr = _.omitBy(params.xattr, (val, name) => name.startsWith?.('noobaa-namespace'));
         } catch (e) {
             dbg.log3("Got an error while trying to omitBy param.xattr:", params.xattr, "error:", e);
         }
@@ -658,12 +658,6 @@ class ObjectSDK {
             params.copy_source.bucket = actual_source_ns.get_bucket(bucket);
             params.copy_source.obj_id = source_md.obj_id;
             params.copy_source.version_id = source_md.version_id;
-            if (source_ns instanceof NamespaceFS) {
-                params.copy_source.nsfs_copy_fallback = () => {
-                    this._populate_nsfs_copy_fallback({ source_params, source_ns, params });
-                    params.copy_source = null;
-                };
-            }
         } else {
             // source cannot be copied directly (different plaforms, accounts, etc.)
             // set the source_stream to read from the copy source
@@ -701,9 +695,9 @@ class ObjectSDK {
         }
     }
 
-    // nsfs copy_object & server side copy consisted of link and a fallback to 
+    // nsfs copy_object & server side copy consisted of link and a fallback to
     // read stream and then upload stream
-    // nsfs copy object when can't server side copy - fallback directly 
+    // nsfs copy object when can't server side copy - fallback directly
     _populate_nsfs_copy_fallback({ source_ns, params, source_params }) {
         const read_stream = new stream.PassThrough();
         source_ns.read_object_stream(source_params, this, read_stream)

src/test/system_tests/ceph_s3_tests/s3-tests-lists/nsfs_s3_tests_pending_list.txt

@@ -157,7 +157,6 @@ s3tests_boto3/functional/test_s3.py::test_object_copy_to_itself_with_metadata
 s3tests_boto3/functional/test_s3.py::test_object_copy_canned_acl
 s3tests_boto3/functional/test_s3.py::test_object_copy_retaining_metadata
 s3tests_boto3/functional/test_s3.py::test_object_copy_replacing_metadata
-s3tests_boto3/functional/test_s3.py::test_object_copy_versioning_multipart_upload
 s3tests_boto3/functional/test_s3.py::test_list_multipart_upload
 s3tests_boto3/functional/test_s3.py::test_multipart_upload_missing_part
 s3tests_boto3/functional/test_s3.py::test_multipart_upload_incorrect_etag