refactor: replace better-queue with fastq (#32941)

* refactor: replace better-queue with fastq

* gatsby-transformer-screenshot

* revert the change to recipes

Recipes need another look as tests are failing

* Update yarn.lock

* Remove redundant comment

Author: vladar

packages/gatsby-source-filesystem/package.json

@@ -8,7 +8,7 @@
   },
   "dependencies": {
     "@babel/runtime": "^7.14.8",
-    "better-queue": "^3.8.10",
+    "fastq": "^1.11.1",
     "chokidar": "^3.5.2",
     "file-type": "^16.0.0",
     "fs-extra": "^10.0.0",

packages/gatsby-source-filesystem/src/create-remote-file-node.js

@@ -2,7 +2,7 @@ const fs = require(`fs-extra`)
 const { createContentDigest, fetchRemoteFile } = require(`gatsby-core-utils`)
 const path = require(`path`)
 const { isWebUri } = require(`valid-url`)
-const Queue = require(`better-queue`)
+const Queue = require(`fastq`)
 const { createFileNode } = require(`./create-file-node`)
 const { getRemoteFileExtension, createFilePath } = require(`./utils`)
 
@@ -46,17 +46,7 @@ let showFlagWarning = !!process.env.GATSBY_EXPERIMENTAL_REMOTE_FILE_PLACEHOLDER
  * Queue Management *
  ********************/
 
-/**
- * Queue
- * Use the task's url as the id
- * When pushing a task with a similar id, prefer the original task
- * as it's already in the processing cache
- */
-const queue = new Queue(pushToQueue, {
-  id: `url`,
-  merge: (old, _, cb) => cb(old),
-  concurrent: process.env.GATSBY_CONCURRENT_DOWNLOAD || 200,
-})
+const queue = Queue(pushToQueue, process.env.GATSBY_CONCURRENT_DOWNLOAD || 200)
 
 /**
  * @callback {Queue~queueCallback}
@@ -168,14 +158,13 @@ const processingCache = {}
  */
 const pushTask = task =>
   new Promise((resolve, reject) => {
-    queue
-      .push(task)
-      .on(`finish`, task => {
-        resolve(task)
-      })
-      .on(`failed`, err => {
+    queue.push(task, (err, node) => {
+      if (!err) {
+        resolve(node)
+      } else {
         reject(`failed to process ${task.url}\n${err}`)
-      })
+      }
+    })
   })
 
 /***************

packages/gatsby-transformer-screenshot/package.json

@@ -9,7 +9,7 @@
   "dependencies": {
     "@babel/runtime": "^7.14.8",
     "axios": "^0.21.1",
-    "better-queue": "^3.8.10"
+    "fastq": "^1.11.1"
   },
   "devDependencies": {
     "@babel/cli": "^7.14.8",

packages/gatsby-transformer-screenshot/src/gatsby-node.js

@@ -6,14 +6,19 @@ const SCREENSHOT_ENDPOINT = `https://h7iqvn4842.execute-api.us-east-2.amazonaws.
 const LAMBDA_CONCURRENCY_LIMIT = 50
 const USE_PLACEHOLDER_IMAGE = process.env.GATSBY_SCREENSHOT_PLACEHOLDER
 
-const screenshotQueue = new Queue(
-  (input, cb) => {
-    createScreenshotNode(input)
-      .then(r => cb(null, r))
-      .catch(e => cb(e))
-  },
-  { concurrent: LAMBDA_CONCURRENCY_LIMIT, maxRetries: 3, retryDelay: 1000 }
-)
+const screenshotQueue = Queue.promise(worker, LAMBDA_CONCURRENCY_LIMIT)
+
+async function worker(input) {
+  // maxRetries: 3, retryDelay: 1000
+  for (let i = 0; i < 2; i++) {
+    try {
+      return await createScreenshotNode(input)
+    } catch (e) {
+      await new Promise(resolve => setTimeout(resolve, 1000))
+    }
+  }
+  return await createScreenshotNode(input)
+}
 
 exports.onPreBootstrap = (
   {
@@ -70,10 +75,10 @@ exports.onPreBootstrap = (
     return null
   }
 
-  return new Promise((resolve, reject) => {
-    screenshotQueue.on(`drain`, () => {
+  return new Promise(resolve => {
+    screenshotQueue.drain = () => {
       resolve()
-    })
+    }
   })
 }
 
@@ -100,8 +105,8 @@ exports.onCreateNode = async (
 
   try {
     const screenshotNode = await new Promise((resolve, reject) => {
-      screenshotQueue
-        .push({
+      screenshotQueue.push(
+        {
           url: node.url,
           parent: node.id,
           store,
@@ -111,13 +116,15 @@ exports.onCreateNode = async (
           getCache,
           createContentDigest,
           parentNodeId: node.id,
-        })
-        .on(`finish`, r => {
-          resolve(r)
-        })
-        .on(`failed`, e => {
-          reject(e)
-        })
+        },
+        (err, result) => {
+          if (err) {
+            reject(err)
+          } else {
+            resolve(result)
+          }
+        }
+      )
     })
 
     createParentChildLink({