feat(get): add a new threads to limit load on server

Downloading large collections caused an unpredictable amount of requests to run in parallel. The new
limit makes download requests not to run in parallel.

Also use Promise.all / correct awaits for promises in general.

Author: DrRataplan

commands/get.js

@@ -1,6 +1,8 @@
 import { resolve, join, posix, dirname, basename } from 'node:path'
-import { writeFileSync, statSync, existsSync, mkdirSync } from 'node:fs'
+import { statSync, existsSync, mkdirSync } from 'node:fs'
+import { writeFile } from 'node:fs/promises'
 import { connect } from '@existdb/node-exist'
+import Bottleneck from 'bottleneck'
 
 /**
  * @typedef { import("@existdb/node-exist").NodeExist } NodeExist
@@ -14,6 +16,8 @@ import { connect } from '@existdb/node-exist'
  * @prop {Number} depth how many levels to traverse down for recursive and tree views
  * @prop {String[]} include filter items
  * @prop {String[]} exclude filter items
+ * @prop {Number} threads How many resources should be downloaded at the same time
+ * @prop {Number} mintime How long a downloads should take at least
  */
 
 /**
@@ -57,10 +61,7 @@ const xmlBooleanSetting = {
     return value
   }
 }
-const serializationOptionNames = [
-  'insert-final-newline',
-  'omit-xml-declaration'
-]
+const serializationOptionNames = ['insert-final-newline', 'omit-xml-declaration']
 
 const serializationDefaults = {
   // "exist:indent": "no",
@@ -71,7 +72,7 @@ const serializationDefaults = {
 
 function getSerializationOptions (options) {
   const serializationOptions = serializationDefaults
-  serializationOptionNames.forEach(o => {
+  serializationOptionNames.forEach((o) => {
     if (o in options) {
       serializationOptions[o] = options[o]
     }
@@ -83,6 +84,7 @@ function getSerializationOptions (options) {
 /**
  * Download a single resource into an existdb instance
  * @param {NodeExist.BoundModules} db NodeExist client
+ * @param {GetOptions} options
  * @param {Boolean} verbose
  * @param {ResourceInfo} resource
  * @param {String} directory
@@ -100,7 +102,7 @@ async function downloadResource (db, options, resource, directory, collection, r
     }
     const localName = rename || posix.basename(resource.name)
     const localPath = join(directory, localName)
-    await writeFileSync(localPath, fileContents)
+    await writeFile(localPath, fileContents)
 
     if (verbose) {
       console.log(`✔︎ downloaded resource ${localPath}`)
@@ -115,11 +117,13 @@ async function downloadResource (db, options, resource, directory, collection, r
 /**
  * download a collection from an existdb instance
  * @param {NodeExist} db NodeExist client
+ * @param {GetOptions} options
  * @param {boolean} verbose
  * @param {String} collection
  * @param {String} baseCollection
+ * @param {Bottleneck} limiter
  */
-async function downloadCollection (db, options, collection, baseCollection, directory) {
+async function downloadCollection (db, options, collection, baseCollection, directory, limiter) {
   const absCollection = posix.join(baseCollection, collection)
   const { verbose } = options
   try {
@@ -134,12 +138,21 @@ async function downloadCollection (db, options, collection, baseCollection, dire
     }
 
     const targetDir = posix.join(directory, collection)
-    await collectionMeta.documents.forEach(
-      async resource => downloadResource(db, options, resource, targetDir, absCollection))
+    // Download all documents. Do this in parallel, but not everything at once. Pool that work so we don't take down the
+    // server
+    await Promise.all(
+      collectionMeta.documents.map(async (resource) => {
+        await limiter.schedule(() => downloadResource(db, options, resource, targetDir, absCollection))
+      })
+    )
 
     // recursive (optional?)
-    await collectionMeta.collections.forEach(
-      async collection => downloadCollection(db, options, collection, absCollection, targetDir))
+
+    // There should always be fewer collections than resources, so no need for pooling. Go over them one by one. No need
+    // to do this in parallel
+    for (const collection of collectionMeta.collections) {
+      await downloadCollection(db, options, collection, absCollection, targetDir, limiter)
+    }
 
     return true
   } catch (e) {
@@ -189,12 +202,13 @@ async function getPathInfo (db, path) {
  */
 async function downloadCollectionOrResource (db, source, target, options) {
   // read parameters
-//  const start = Date.now()
+  //  const start = Date.now()
   const root = resolve(target)
 
   if (options.verbose) {
     console.log('Downloading:', source, 'to', root)
-    console.log('Server:',
+    console.log(
+      'Server:',
       (db.client.isSecure ? 'https' : 'http') + '://' + db.client.options.host + ':' + db.client.options.port,
       '(v' + options.version + ')'
     )
@@ -205,6 +219,7 @@ async function downloadCollectionOrResource (db, source, target, options) {
     if (options.exclude.length) {
       console.log('Exclude:\n', ...options.exclude, '\n')
     }
+    console.log(`Downloading up to ${options.threads} resources at a time`)
   }
 
   // initial file
@@ -260,10 +275,13 @@ async function downloadCollectionOrResource (db, source, target, options) {
     throw Error(`${source} is a collection but ${root} is not a directory`)
   }
 
+  const limiter = new Bottleneck({
+    maxConcurrent: options.threads,
+    minTime: options.mintime
+  })
+
   // download collection into a folder
-  return await downloadCollection(db, options,
-    posix.basename(info.name),
-    posix.dirname(info.name), root)
+  return await downloadCollection(db, options, posix.basename(info.name), posix.dirname(info.name), root, limiter)
 }
 
 export const command = ['get [options] <source> <target>', 'download', 'fetch']
@@ -301,14 +319,34 @@ export function builder (yargs) {
       type: 'boolean',
       default: false
     })
+    .option('t', {
+      alias: 'threads',
+      describe: 'The maximum number of concurrent threads that will be used to dowload data',
+      type: 'number',
+      default: 4
+    })
+    .option('m', {
+      alias: 'mintime',
+      describe: 'The minimum time each dowload will take',
+      type: 'number',
+      default: 0
+    })
     .nargs({ i: 1, e: 1 })
 }
 
 export async function handler (argv) {
   if (argv.help) {
     return 0
   }
-  const { source } = argv
+
+  const { threads, mintime, source } = argv
+
+  if (typeof mintime !== 'number' || mintime < 0) {
+    throw Error('Invalid value for option "mintime"; must be an integer equal or greater than zero.')
+  }
+  if (typeof threads !== 'number' || threads <= 0) {
+    throw Error('Invalid value for option "threads"; must be an integer equal or greater than zero.')
+  }
 
   const target = argv.target ? argv.target : '.'
 

spec/tests/get.js

@@ -7,12 +7,16 @@ const testCollection = '/db/' + testCollectionName
 
 async function removeLocalDownload () {
   const { stderrRmLocal } = await run('rm', ['-rf', testCollectionName])
-  if (stderrRmLocal) { return console.error(stderrRmLocal) }
+  if (stderrRmLocal) {
+    return console.error(stderrRmLocal)
+  }
 }
 
 async function removeRemoteCollection (t) {
   const { stderrRmRemote } = await run('xst', ['rm', '-rf', testCollection])
-  if (stderrRmRemote) { return console.error(stderrRmRemote) }
+  if (stderrRmRemote) {
+    return console.error(stderrRmRemote)
+  }
 }
 
 async function tearDown (t) {
@@ -51,7 +55,9 @@ async function prepare (t) {
     storeResourceQuery(testCollection, 'test.xq', '"1"')
   ].join(',')
   const { stderr, stdout } = await run('xst', ['run', query], asAdmin)
-  if (stderr) { return t.fail(stderr) }
+  if (stderr) {
+    return t.fail(stderr)
+  }
   t.true(stdout)
 }
 
@@ -61,7 +67,9 @@ function storeResourceQuery (collection, fileName, content) {
 
 test("calling 'xst get --help'", async (t) => {
   const { stderr, stdout } = await run('xst', ['get', '--help'])
-  if (stderr) { return t.fail(stderr) }
+  if (stderr) {
+    return t.fail(stderr)
+  }
 
   t.ok(stdout, stdout)
   t.end()
@@ -72,8 +80,10 @@ test('with test collection', async (t) => {
   t.test(prepare)
 
   t.test(`can get ${testCollection} as admin`, async (st) => {
-    const { stderr, stdout } = await run('xst', ['get', testCollection, '.'], asAdmin)
-    if (stderr) { return st.fail(stderr) }
+    const { stderr, stdout } = await run('xst', ['get', testCollection, '.', '--threads', '2'], asAdmin)
+    if (stderr) {
+      return st.fail(stderr)
+    }
     st.plan(3)
 
     st.notOk(stdout, 'no output')
@@ -84,7 +94,7 @@ test('with test collection', async (t) => {
   })
 
   t.test(`cannot get ${testCollection} as guest`, async (st) => {
-    const { stderr, stdout } = await run('xst', ['get', testCollection, '.'])
+    const { stderr, stdout } = await run('xst', ['get', testCollection, '.', '--threads', '2'])
     if (stdout) {
       await removeLocalDownload()
       return st.ok(stderr)
@@ -98,7 +108,7 @@ test('with test collection', async (t) => {
   })
 
   t.test(`'xst get --verbose ${testCollection}' as admin`, async (st) => {
-    const { stderr, stdout } = await run('xst', ['get', '--verbose', testCollection, '.'], asAdmin)
+    const { stderr, stdout } = await run('xst', ['get', '--verbose', testCollection, '.', '--threads', '2'], asAdmin)
     st.plan(9)
     st.equal(stderr, 'Connecting to https://localhost:8443 as admin\n', stderr)
 
@@ -110,8 +120,16 @@ test('with test collection', async (t) => {
     st.equal(lines[2], 'User: admin')
     st.equal(lines.length, 16, 'all expected lines in verbose output')
     // files are not downloaded in reproducible order
-    st.equal(lines.filter(l => /^✔︎ created directory [/\w]+\/get-test/.exec(l)).length, 3, 'notify 3 directories created')
-    st.equal(lines.filter(l => /^✔︎ downloaded resource [/\w]+\/get-test/.exec(l)).length, 9, 'notify 9 resources downloaded')
+    st.equal(
+      lines.filter((l) => /^✔︎ created directory [/\w]+\/get-test/.exec(l)).length,
+      3,
+      'notify 3 directories created'
+    )
+    st.equal(
+      lines.filter((l) => /^✔︎ downloaded resource [/\w]+\/get-test/.exec(l)).length,
+      9,
+      'notify 9 resources downloaded'
+    )
 
     st.deepEqual(readdirSync(testCollectionName), expectedDirectoryListing, 'all files were downloaded')
     st.deepEqual(readdirSync(testCollectionName + '/subcollection'), ['b'], 'subcollection contents were downloaded')
@@ -123,15 +141,23 @@ test('with test collection', async (t) => {
 
   t.test(`'xst get ${testCollection}/empty-subcollection ${additionalTestDirectory}' as admin`, async (st) => {
     await run('mkdir', [additionalTestDirectory])
-    const { stderr, stdout } = await run('xst', ['get', testCollection + '/empty-subcollection', additionalTestDirectory], asAdmin)
+    const { stderr, stdout } = await run(
+      'xst',
+      ['get', testCollection + '/empty-subcollection', additionalTestDirectory],
+      asAdmin
+    )
     if (stderr) {
       st.fail(stderr)
       return st.end()
     }
     st.plan(2)
     st.notOk(stdout, 'no output')
 
-    st.deepEqual(readdirSync(additionalTestDirectory), ['empty-subcollection'], `empty subcollection was created in ${additionalTestDirectory} folder`)
+    st.deepEqual(
+      readdirSync(additionalTestDirectory),
+      ['empty-subcollection'],
+      `empty subcollection was created in ${additionalTestDirectory} folder`
+    )
     await run('rm', ['-rf', additionalTestDirectory])
   })