Tests and code adjustments post bumps

- The async module bump lead to the need of using callbacks on functions like
doWhilst.
- Also we need to add for mongo some parameters to have the expected
return , in this specific case  returnDocument: 'after',
includeResultMetadata: true.
- In our tests, bucketInfos used to send an empty array,
this commit actually uses fill to send the intended array as before
async used to just ignore the empty array and return the expected result.

Issue: S3UTILS-171

Author: benzekrimaha

CRR/ReplicationStatusUpdater.js

@@ -352,7 +352,7 @@ class ReplicationStatusUpdater {
                     });
                 },
             ),
-            () => {
+            async () => {
                 if (this._nUpdated >= this.maxUpdates || this._nProcessed >= this.maxScanned) {
                     this._logProgress();
                     let remainingBuckets;

CompareRaftMembers/BlockDigestsStorage.js

@@ -27,7 +27,7 @@ class BlockDigestsStorage extends stream.Writable {
     constructor(params) {
         super({ objectMode: true });
         const { levelPath, db } = params;
-        if (params.levelPath) {
+        if (levelPath) {
             this.db = new Level(params.levelPath);
         } else {
             this.db = db;
@@ -38,16 +38,16 @@ class BlockDigestsStorage extends stream.Writable {
     }
 
     _write(blockInfo, encoding, callback) {
-        const { size, digest } = blockInfo;
+        const { size, digest, lastKey } = blockInfo;
         this.cargo.push({
             type: 'put',
-            key: blockInfo.lastKey, // index by last block key for efficient lookup
+            key: lastKey, // index by last block key for efficient lookup
             value: JSON.stringify({ size, digest }),
         });
         // heuristic to have basic flow control: delay the callback
         // while queue size is above a reasonable size
         async.whilst(
-            () => this.cargo.length() > MAX_QUEUE_SIZE,
+            async () => this.cargo.length() > MAX_QUEUE_SIZE,
             cb => setTimeout(cb, 100),
             () => callback(),
         );
@@ -57,9 +57,9 @@ class BlockDigestsStorage extends stream.Writable {
         if (this.cargo.idle()) {
             this.db.close(callback);
         } else {
-            this.cargo.drain = () => {
+            this.cargo.drain(() => {
                 this.db.close(callback);
-            };
+            });
         }
     }
 }

CountItems/CountManager.js

@@ -174,7 +174,9 @@ class CountManager {
         }));
         this.store.buckets += bucketCount;
         this.store.bucketList = this.store.bucketList.concat(transformedInfos);
-        this.q.push(bucketInfos);
+        bucketInfos.forEach(bucketInfo => {
+            this.q.push(bucketInfo);
+        });
         this.log.debug('added work', {
             workInQueue: this.q.length(),
             workInProgress: this.q.running(),
@@ -195,18 +197,22 @@ class CountManager {
             Object.values(this.workers)
                 .forEach(worker => this.workerList.push(worker.id));
         }
-        this.q.error = err => {
+        this.q.error(err => {
+            this.log.error('error processing bucket', {
+                error: err,
+                method: 'CountManager::addWork',
+            });
             this.q.pause();
             this.q.kill();
             return process.nextTick(onceCB, err);
-        };
-        this.q.drain = () => {
+        });
+        this.q.drain(() => {
             if (this.q.idle()) {
                 this.q.pause();
                 this.q.kill();
                 process.nextTick(onceCB);
             }
-        };
+        });
         this.q.resume();
     }
 }

CountItems/CountWorker.js

@@ -36,7 +36,7 @@ class CountWorker {
         if (bucketInfoObj._websiteConfiguration) {
             Object.assign(bucketInfoObj, { _websiteConfiguration: null });
         }
-        const bucketInfo = BucketInfo.fromObj(bucketInfoObj);
+        const bucketInfo = BucketInfo.fromObj(bucketInfoObj.bucketInfo || bucketInfoObj);
         const bucketName = bucketInfo.getName();
         this.log.info(`${process.pid} handling ${bucketName}`);
         return async.waterfall([

CountItems/CountWorkerObj.js

@@ -21,9 +21,13 @@ class CountWorkerObj {
             }
         });
         this._worker.on('message', data => {
-            if (data.owner !== 'scality') {return;}
+            if (data.owner !== 'scality') {
+                return;
+            }
             const cb = this._getCallback(data.id);
-            if (!cb) {return;}
+            if (!cb) {
+                return;
+            }
             switch (data.type) {
             case 'setup':
                 if (data.status === 'passed') {
@@ -81,7 +85,9 @@ class CountWorkerObj {
     }
 
     _getCallback(id) {
-        if (!this.callbacks.has(id)) {return null;}
+        if (!this.callbacks.has(id)) {
+            return null;
+        }
         const ret = this.callbacks.get(id);
         this.callbacks.delete(id);
         return ret.callback;

StalledRetry/StalledRequestHandler.js

@@ -17,7 +17,9 @@ class StalledRequestHandler {
     }
 
     queueSetup() {
-        this._queue = async.queue(({ bucket, batch, getNext }, done) => {
+        this._queue = async.queue((task, done) => {
+            const { bucket, batch, getNext } = task;
+            
             if (this.queueError !== null) {
                 return done();
             }
@@ -42,15 +44,16 @@ class StalledRequestHandler {
             });
         }, this.concurrentRequests);
 
-        this._queue.error = (err, { bucket, batch }) => {
+        this._queue.error((err, task) => {
+            const { bucket, batch } = task;
             this.log.error('error occurred while processing request', {
                 error: err,
                 lastBatch: batch,
                 bucket,
             });
             this.queueError = err;
             this.kill();
-        };
+        });
     }
 
     isInProgress() {
@@ -68,7 +71,7 @@ class StalledRequestHandler {
 
     _waitForCompletion(cb) {
         async.whilst(
-            () => (!this.killed && this.isInProgress()),
+            async () => (!this.killed && this.isInProgress()),
             done => setTimeout(done, 1000),
             cb,
         );
@@ -110,7 +113,7 @@ class StalledRequestHandler {
 
         return async.times(
             this.concurrentRequests,
-            (_, cb) => nextBatch(cb),
+            (n, next) => nextBatch(next),
             err => {
                 if (err) {
                     this.log.error('failed to populate queue', {

VerifyReplication/verifyReplication.js

@@ -88,7 +88,7 @@ function handlePrefixes(prefixList, cb) {
             bucket: statusObj.srcBucket,
             prefix,
             listingLimit,
-        };       
+        };
         return listAndCompare(params, done);
     }, cb);
 }

compareBuckets/compareBuckets.js

@@ -1,7 +1,3 @@
-/* eslint-disable max-len */
-/* eslint-disable no-console */
-/* eslint-disable comma-dangle */
-
 const async = require('async');
 
 const { listBucketMasterKeys } = require('./utils');
@@ -91,14 +87,13 @@ function compareBuckets(params, log, cb) {
                     if (srcDone || srcContents.length > 0) {
                         return process.nextTick(_done);
                     }
-
                     return listBucketMasterKeys(
                         bucketdSrcParams,
                         (err, isTruncated, marker, contents) => {
                             if (err) {
+                                log.error('Error fetching src bucket', { error: err });
                                 return _done(err);
                             }
-
                             srcContents = contents;
                             srcDone = !isTruncated;
                             bucketdSrcParams.marker = marker;
@@ -116,9 +111,9 @@ function compareBuckets(params, log, cb) {
                         bucketdDstParams,
                         (err, isTruncated, marker, contents) => {
                             if (err) {
+                                log.error('Error fetching dst bucket', { error: err });
                                 return _done(err);
                             }
-
                             dstContents = contents;
                             dstDone = !isTruncated;
                             bucketdDstParams.marker = marker;
@@ -129,6 +124,7 @@ function compareBuckets(params, log, cb) {
                 },
             }, err => {
                 if (err) {
+                    log.error('Parallel fetch error', { error: err });
                     return done(err);
                 }
 
@@ -224,15 +220,14 @@ function compareBuckets(params, log, cb) {
                 return process.nextTick(() => done(null));
             });
         },
-        () => (!srcDone || !dstDone || srcContents.length > 0 || dstContents.length > 0),
+        async () => (!srcDone || !dstDone || srcContents.length > 0 || dstContents.length > 0),
         err => {
             statusObj.dstBucketInProgress = null;
             statusObj.srcBucketInProgress = null;
             cb(err);
         }
     );
 }
-
 module.exports = {
     compareBuckets,
     compareObjectsReport,

repairDuplicateVersionIds.js

@@ -154,8 +154,6 @@ function repairObject(objInfo, cb) {
             // use "versionId" from the parsed metadata instead of
             // `objInfo.firstVersionId`, since it may have changed
             // since the scan ran
-            //
-             
             [md.versionId] = versionIds;
             return putObjectMetadata(objInfo.objectUrl, md, err => {
                 if (err) {

service-level-sidecar/bucketd.js

@@ -36,8 +36,7 @@ async function* listBuckets(log) {
 
     while (true) {
         let res;
-
-        try {             
+        try {   
             res = await listObjects(usersBucket, { ...listingParams, gt }, log);
         } catch (error) {
             if (error.NoSuchBucket) {

tests/constants.js

@@ -29,6 +29,36 @@ module.exports = {
         },
         ingestion: null,
     },
+    internalTestBucketMD: {
+        acl: {
+            Canned: 'private',
+            FULL_CONTROL: [],
+            WRITE: [],
+            WRITE_ACP: [],
+            READ: [],
+            READ_ACP: [],
+        },
+        name: 'test-bucket',
+        owner: 'd1d40abd2bd8250962f7f5774af1bbbeaec9b77a0853749d41ec46f142e66fe4',
+        ownerDisplayName: 'test-name',
+        creationDate: new Date(1678284806000).toString(),
+        mdBucketModelVersion: 10,
+        transient: false,
+        deleted: false,
+        serverSideEncryption: null,
+        versioningConfiguration: null,
+        locationConstraint: 'us-east-1',
+        readLocationConstraint: null,
+        cors: null,
+        replicationConfiguration: null,
+        lifecycleConfiguration: null,
+        uid: '',
+        isNFS: null,
+        capabilities: {
+            VeeamSOSApi: undefined,
+        },
+        ingestion: null,
+    },
     testAccountCanonicalId: 'd1d40abd2bd8250962f7f5774af1bbbeaec9b77a0853749d41ec46f142e66fe4',
     testBucketCreationDate: 1678284808000,
     testUserBucketInfo: { value: { creationDate: new Date(1678284808000).toString() } },