feat(gatsby): enable Jobs API v2 (#19858)

Added a new action called createJobV2 to support the new api. createJob is still available to keep backward-compatibility. I'll add a deprecation message when creatJobV2 is fully operational.

The createJobV2 needs a job object that contains 4 arguments name, inputPaths, outputDir & args. These args are used to create a unique content digest to make sure a job is deterministic.

InputPaths are converted into relative paths before sending it to the worker as they need to be filesystem agnostic.

More info on why can be found in #19831

Author: wardpeet

e2e-tests/production-runtime/gatsby-config.js

@@ -19,5 +19,6 @@ module.exports = {
         icon: `src/images/gatsby-icon.png`, // This path is relative to the root of the site.
       },
     },
+    `gatsby-plugin-local-worker`,
   ].concat(process.env.TEST_PLUGIN_OFFLINE ? [`gatsby-plugin-offline`] : []),
 }

e2e-tests/production-runtime/plugins/gatsby-plugin-local-worker/gatsby-node.js

@@ -0,0 +1,16 @@
+const path = require(`path`)
+
+exports.onPostBootstrap = async ({ actions, store }) => {
+  const result = await actions.createJobV2({
+    name: `TEST_JOB`,
+    inputPaths: [],
+    outputDir: path.join(store.getState().program.directory, `./public`),
+    args: {
+      result: `hello`,
+    },
+  })
+
+  if (result.result !== `hello`) {
+    throw new Error(`the result of our worker is wrong`)
+  }
+}

e2e-tests/production-runtime/plugins/gatsby-plugin-local-worker/gatsby-worker.js

@@ -0,0 +1,9 @@
+const sleep = timeout => new Promise(resolve => setTimeout(resolve, timeout))
+
+exports.TEST_JOB = async ({ args }) => {
+  await sleep(5000)
+
+  return {
+    result: args.result,
+  }
+}

e2e-tests/production-runtime/plugins/gatsby-plugin-local-worker/index.js

@@ -0,0 +1 @@
+// noop

e2e-tests/production-runtime/plugins/gatsby-plugin-local-worker/package.json

@@ -0,0 +1,4 @@
+{
+  "name": "gatsby-plugin-local-worker",
+  "version": "1.0.0"
+}

packages/gatsby/index.d.ts

@@ -804,7 +804,12 @@ export interface Actions {
 
   /** @see https://www.gatsbyjs.org/docs/actions/#createPage */
   createPage<TContext = Record<string, unknown>>(
-    args: { path: string; matchPath?: string; component: string; context: TContext },
+    args: {
+      path: string
+      matchPath?: string
+      component: string
+      context: TContext
+    },
     plugin?: ActionPlugin,
     option?: ActionOptions
   ): void
@@ -878,6 +883,17 @@ export interface Actions {
     plugin?: ActionPlugin
   ): void
 
+  /** @see https://www.gatsbyjs.org/docs/actions/#createJobV2 */
+  createJobV2(
+    job: {
+      name: string
+      inputPaths: string[]
+      outputDir: string
+      args: Record<string, unknown>
+    },
+    plugin?: ActionPlugin
+  ): Promise<unknown>
+
   /** @see https://www.gatsbyjs.org/docs/actions/#setJob */
   setJob(
     job: Record<string, unknown> & { id: string },

packages/gatsby/package.json

@@ -81,6 +81,7 @@
     "graphql": "^14.5.8",
     "graphql-compose": "^6.3.7",
     "graphql-playground-middleware-express": "^1.7.12",
+    "hasha": "^5.1.0",
     "invariant": "^2.2.4",
     "is-relative": "^1.0.0",
     "is-relative-url": "^3.0.0",
@@ -104,6 +105,7 @@
     "opentracing": "^0.14.4",
     "optimize-css-assets-webpack-plugin": "^5.0.3",
     "parseurl": "^1.3.3",
+    "p-defer": "^3.0.0",
     "physical-cpu-count": "^2.0.0",
     "pnp-webpack-plugin": "^1.5.0",
     "postcss-flexbugs-fixes": "^3.3.1",

packages/gatsby/src/bootstrap/__tests__/__snapshots__/remove-stale-jobs.js.snap

@@ -0,0 +1,33 @@
+// Jest Snapshot v1, https://goo.gl/fbAQLP
+
+exports[`remove-stale-jobs should enqueue pending jobs 1`] = `
+Array [
+  undefined,
+]
+`;
+
+exports[`remove-stale-jobs should remove stale jobs from complete cache 1`] = `
+Array [
+  Object {
+    "payload": Object {
+      "contentDigest": "1234",
+    },
+    "plugin": undefined,
+    "traceId": undefined,
+    "type": "REMOVE_STALE_JOB_V2",
+  },
+]
+`;
+
+exports[`remove-stale-jobs should remove stale jobs from pending cache 1`] = `
+Array [
+  Object {
+    "payload": Object {
+      "contentDigest": "1234",
+    },
+    "plugin": undefined,
+    "traceId": undefined,
+    "type": "REMOVE_STALE_JOB_V2",
+  },
+]
+`;

packages/gatsby/src/bootstrap/__tests__/remove-stale-jobs.js

@@ -0,0 +1,86 @@
+jest.mock(`../../utils/jobs-manager`)
+
+const { isJobStale } = require(`../../utils/jobs-manager`)
+const { internalActions, publicActions } = require(`../../redux/actions`)
+
+jest.spyOn(internalActions, `removeStaleJob`)
+
+const removeStaleJobs = require(`../remove-stale-jobs`)
+
+describe(`remove-stale-jobs`, () => {
+  let state
+
+  beforeEach(() => {
+    state = {
+      jobsV2: {
+        complete: new Map(),
+        incomplete: new Map(),
+      },
+    }
+
+    publicActions.createJobV2 = jest.fn()
+    internalActions.removeStaleJob.mockClear()
+  })
+
+  it(`should remove stale jobs from complete cache`, () => {
+    const job = {
+      inputPaths: [`/src/myfile.js`],
+    }
+
+    state.jobsV2.complete.set(`1234`, job)
+
+    isJobStale.mockReturnValue(true)
+
+    expect(removeStaleJobs(state)).toMatchSnapshot()
+    expect(internalActions.removeStaleJob).toHaveBeenCalledTimes(1)
+    expect(internalActions.removeStaleJob).toHaveBeenCalledWith(`1234`)
+    expect(publicActions.createJobV2).not.toHaveBeenCalled()
+  })
+
+  it(`should remove stale jobs from pending cache`, () => {
+    const data = {
+      job: {
+        inputPaths: [`/src/myfile.js`],
+        contentDigest: `1234`,
+      },
+      plugin: {
+        name: `test`,
+        version: `1.0.0`,
+      },
+    }
+
+    state.jobsV2.incomplete.set(`1234`, data)
+
+    isJobStale.mockReturnValue(true)
+
+    expect(removeStaleJobs(state)).toMatchSnapshot()
+    expect(internalActions.removeStaleJob).toHaveBeenCalledTimes(1)
+    expect(internalActions.removeStaleJob).toHaveBeenCalledWith(`1234`)
+    expect(publicActions.createJobV2).not.toHaveBeenCalled()
+  })
+
+  it(`should enqueue pending jobs`, () => {
+    const data = {
+      job: {
+        inputPaths: [`/src/myfile.js`],
+        contentDigest: `1234`,
+      },
+      plugin: {
+        name: `test`,
+        version: `1.0.0`,
+      },
+    }
+
+    state.jobsV2.incomplete.set(`1234`, data)
+
+    isJobStale.mockReturnValue(false)
+
+    expect(removeStaleJobs(state)).toMatchSnapshot()
+    expect(internalActions.removeStaleJob).toHaveBeenCalledTimes(0)
+    expect(publicActions.createJobV2).toHaveBeenCalledTimes(1)
+    expect(publicActions.createJobV2).toHaveBeenCalledWith(
+      data.job,
+      data.plugin
+    )
+  })
+})

packages/gatsby/src/bootstrap/index.js

@@ -19,6 +19,7 @@ const report = require(`gatsby-cli/lib/reporter`)
 const getConfigFile = require(`./get-config-file`)
 const tracer = require(`opentracing`).globalTracer()
 const preferDefault = require(`./prefer-default`)
+const removeStaleJobs = require(`./remove-stale-jobs`)
 // Add `util.promisify` polyfill for old node versions
 require(`util.promisify/shim`)()
 
@@ -160,6 +161,9 @@ module.exports = async (args: BootstrapArgs) => {
 
   activity.end()
 
+  // run stale jobs
+  store.dispatch(removeStaleJobs(store.getState()))
+
   activity = report.activityTimer(`load plugins`, { parentSpan: bootstrapSpan })
   activity.start()
   const flattenedPlugins = await loadPlugins(config, program.directory)
@@ -229,7 +233,7 @@ module.exports = async (args: BootstrapArgs) => {
     .createHash(`md5`)
     .update(JSON.stringify(pluginVersions.concat(hashes)))
     .digest(`hex`)
-  let state = store.getState()
+  const state = store.getState()
   const oldPluginsHash = state && state.status ? state.status.PLUGINS_HASH : ``
 
   // Check if anything has changed. If it has, delete the site's .cache

packages/gatsby/src/bootstrap/remove-stale-jobs.js

@@ -0,0 +1,25 @@
+const { isJobStale } = require(`../utils/jobs-manager`)
+const { publicActions, internalActions } = require(`../redux/actions`)
+
+module.exports = state => {
+  const actions = []
+
+  // If any of our finished jobs are stale we remove them to keep our cache small
+  state.jobsV2.complete.forEach((job, contentDigest) => {
+    if (isJobStale(job)) {
+      actions.push(internalActions.removeStaleJob(contentDigest))
+    }
+  })
+
+  // If any of our pending jobs do not have an existing inputPath or the inputPath changed
+  // we remove it from the queue as they would fail anyway
+  state.jobsV2.incomplete.forEach(({ job, plugin }) => {
+    if (isJobStale(job)) {
+      actions.push(internalActions.removeStaleJob(job.contentDigest))
+    } else {
+      actions.push(publicActions.createJobV2(job, plugin))
+    }
+  })
+
+  return actions
+}

packages/gatsby/src/commands/build.js

@@ -16,6 +16,9 @@ const queryUtil = require(`../query`)
 const appDataUtil = require(`../utils/app-data`)
 const WorkerPool = require(`../utils/worker/pool`)
 const { structureWebpackErrors } = require(`../utils/webpack-error-utils`)
+const {
+  waitUntilAllJobsComplete: waitUntilAllJobsV2Complete,
+} = require(`../utils/jobs-manager`)
 
 type BuildArgs = {
   directory: string,
@@ -25,8 +28,8 @@ type BuildArgs = {
   openTracingConfigFile: string,
 }
 
-const waitJobsFinished = () =>
-  new Promise((resolve, reject) => {
+const waitUntilAllJobsComplete = () => {
+  const jobsV1Promise = new Promise(resolve => {
     const onEndJob = () => {
       if (store.getState().jobs.active.length === 0) {
         resolve()
@@ -37,6 +40,12 @@ const waitJobsFinished = () =>
     onEndJob()
   })
 
+  return Promise.all([
+    jobsV1Promise,
+    waitUntilAllJobsV2Complete(),
+  ]).then(() => {})
+}
+
 module.exports = async function build(program: BuildArgs) {
   const publicDir = path.join(program.directory, `public`)
   initTracer(program.openTracingConfigFile)
@@ -128,9 +137,13 @@ module.exports = async function build(program: BuildArgs) {
     `BOOTSTRAP_QUERY_RUNNING_FINISHED`
   )
 
-  await waitJobsFinished()
+  await db.saveState()
+
+  await waitUntilAllJobsComplete()
 
+  // we need to save it again to make sure our latest state has been saved
   await db.saveState()
+
   const pagePaths = [...store.getState().pages.keys()]
   activity = report.createProgress(
     `Building static HTML for pages`,
@@ -176,6 +189,9 @@ module.exports = async function build(program: BuildArgs) {
     parentSpan: buildSpan,
   })
 
+  // Make sure we saved the latest state so we have all jobs cached
+  await db.saveState()
+
   report.info(`Done building in ${process.uptime()} sec`)
 
   buildSpan.finish()

packages/gatsby/src/commands/develop.ts

@@ -54,6 +54,8 @@ import {
   structureWebpackErrors,
 } from "../utils/webpack-error-utils"
 
+import { waitUntilAllJobsComplete as waitUntilAllJobsV2Complete } from "../utils/jobs-manager"
+
 interface ICert {
   keyPath: string
   certPath: string
@@ -75,6 +77,24 @@ interface IProgram {
   ssl?: ICert
 }
 
+const waitUntilAllJobsComplete = (): Promise<void> => {
+  const jobsV1Promise = new Promise(resolve => {
+    const onEndJob = (): void => {
+      if (store.getState().jobs.active.length === 0) {
+        resolve()
+        emitter.off(`END_JOB`, onEndJob)
+      }
+    }
+    emitter.on(`END_JOB`, onEndJob)
+    onEndJob()
+  })
+
+  return Promise.all([
+    jobsV1Promise,
+    waitUntilAllJobsV2Complete(),
+  ]).then(() => {})
+}
+
 // const isInteractive = process.stdout.isTTY
 
 // Watch the static directory and copy files to public as they're added or
@@ -88,18 +108,6 @@ onExit(() => {
   telemetry.trackCli(`DEVELOP_STOP`)
 })
 
-const waitJobsFinished = (): Promise<void> =>
-  new Promise(resolve => {
-    const onEndJob = (): void => {
-      if (store.getState().jobs.active.length === 0) {
-        resolve()
-        emitter.off(`END_JOB`, onEndJob)
-      }
-    }
-    emitter.on(`END_JOB`, onEndJob)
-    onEndJob()
-  })
-
 type ActivityTracker = any // TODO: Replace this with proper type once reporter is typed
 
 interface IServer {
@@ -421,7 +429,9 @@ module.exports = async (program: IProgram): Promise<void> => {
     `BOOTSTRAP_QUERY_RUNNING_FINISHED`
   )
 
-  await waitJobsFinished()
+  await db.saveState()
+
+  await waitUntilAllJobsComplete()
   requiresWriter.startListener()
   db.startAutosave()
   queryUtil.startListeningToDevelopQueue()
@@ -457,9 +467,9 @@ module.exports = async (program: IProgram): Promise<void> => {
       })
 
     const isUnspecifiedHost = host === `0.0.0.0` || host === `::`
-    let prettyHost = host,
-      lanUrlForConfig,
-      lanUrlForTerminal
+    let prettyHost = host
+    let lanUrlForConfig
+    let lanUrlForTerminal
     if (isUnspecifiedHost) {
       prettyHost = `localhost`