bf: search with aggregate instead of find

The search was using find().sort() and was disrupting user defined
search queries and custom indexes. The sort() is needed to implement
a stateless paging system. The combo of user defined query and sort is
now implemented with a 2 stage aggregate on server side.
We always limit the execution time maxTimeMs to 5mn (tunable by an
environment variable).

Author: vrancurel

lib/storage/metadata/mongoclient/MongoClientInterface.js

@@ -39,6 +39,8 @@ const CONNECT_TIMEOUT_MS = 5000;
 // MongoDB default
 const SOCKET_TIMEOUT_MS = 360000;
 const CONCURRENT_CURSORS = 10;
+// Search
+const MAX_TIME_MS = 300000;
 
 const initialInstanceID = process.env.INITIAL_INSTANCE_ID;
 
@@ -109,6 +111,11 @@ class MongoClientInterface {
             !Number.isNaN(process.env.CONCURRENT_CURSORS))
             ? Number.parseInt(process.env.CONCURRENT_CURSORS, 10)
             : CONCURRENT_CURSORS;
+
+        this.maxTimeMs = (process.env.MAX_TIME_MS &&
+            !Number.isNaN(process.env.MAX_TIME_MS))
+            ? Number.parseInt(process.env.MAX_TIME_MS, 10)
+            : MAX_TIME_MS;
     }
 
     setup(cb) {
@@ -941,6 +948,7 @@ class MongoClientInterface {
 
     internalListObject(bucketName, params, extension, log, cb) {
         const c = this.getCollection(bucketName);
+        Object.assign(params, { maxTimeMs: this.maxTimeMs });
         const stream = new MongoReadStream(c, params, params.mongifiedSearch);
         const skip = new Skip({
             extension,

lib/storage/metadata/mongoclient/readStream.js

@@ -60,12 +60,21 @@ class MongoReadStream extends Readable {
         }
 
         if (searchOptions) {
-            Object.assign(query, searchOptions);
+            this._cursor = c.aggregate([
+                { $match: searchOptions }, // user query
+                { $match: query }, // for paging
+            ],
+                {
+                    cursor: {}, // use default batchSize
+                    maxTimeMs: options.maxTimeMs, // always limit
+                    allowDiskUse: true, // stage large queries on disk
+                },
+                null); // will return a cursor
+        } else {
+            this._cursor = c.find(query).sort({
+                _id: options.reverse ? -1 : 1,
+            });
         }
-
-        this._cursor = c.find(query).sort({
-            _id: options.reverse ? -1 : 1,
-        });
         if (options.limit && options.limit !== -1) {
             this._cursor = this._cursor.limit(options.limit);
         }