Adding worker controls to both automations and queries.

packages/builder/src/components/integration/QueryViewer.svelte

@@ -157,6 +157,7 @@
 
   onMount(async () => {
     if (!query || !query._id) {
+      roleId = Roles.BASIC
       return
     }
     try {

packages/server/src/api/controllers/query.js

@@ -3,7 +3,9 @@ const CouchDB = require("../../db")
 const { generateQueryID, getQueryParams } = require("../../db/utils")
 const { BaseQueryVerbs } = require("../../constants")
 const env = require("../../environment")
-const queryRunner = require("../../utilities/queryRunner")
+const { Thread, ThreadType } = require("../../threads")
+
+const Runner = new Thread(ThreadType.QUERY, { timeoutMs: 10000 })
 
 // simple function to append "readable" to all read queries
 function enrichQueries(input) {
@@ -104,12 +106,12 @@ exports.preview = async function (ctx) {
   const { fields, parameters, queryVerb, transformer } = ctx.request.body
   const enrichedQuery = await enrichQueryFields(fields, parameters)
 
-  const { rows, keys } = await queryRunner(
+  const { rows, keys } = await Runner.run({
     datasource,
     queryVerb,
-    enrichedQuery,
-    transformer
-  )
+    query: enrichedQuery,
+    transformer,
+  })
 
   ctx.body = {
     rows,
@@ -129,12 +131,12 @@ exports.execute = async function (ctx) {
   )
 
   // call the relevant CRUD method on the integration class
-  const { rows } = await queryRunner(
+  const { rows } = await Runner.run({
     datasource,
-    query.queryVerb,
-    enrichedQuery,
-    query.transformer
-  )
+    queryVerb: query.queryVerb,
+    query: enrichedQuery,
+    transformer: query.transformer,
+  })
   ctx.body = rows
 }
 

packages/server/src/automations/tests/automation.spec.js

@@ -11,8 +11,7 @@ jest.spyOn(global.console, "error")
 
 require("../../environment")
 const automation = require("../index")
-const usageQuota = require("../../utilities/usageQuota")
-const thread = require("../thread")
+const thread = require("../../threads/automation")
 const triggers = require("../triggers")
 const { basicAutomation } = require("../../tests/utilities/structures")
 const { wait } = require("../../utilities")
@@ -54,7 +53,7 @@ describe("Run through some parts of the automations system", () => {
       }
     })
     await wait(100)
-    expect(thread).toHaveBeenCalledWith(makePartial({
+    expect().toHaveBeenCalledWith(makePartial({
       data: {
         event: {
           fields: {

packages/server/src/automations/utils.js

@@ -1,4 +1,4 @@
-const runner = require("./thread")
+const { Thread, ThreadType } = require("../threads")
 const { definitions } = require("./triggerInfo")
 const webhooks = require("../api/controllers/webhook")
 const CouchDB = require("../db")
@@ -10,11 +10,12 @@ const { getDeployedAppID } = require("@budibase/auth/db")
 
 const WH_STEP_ID = definitions.WEBHOOK.stepId
 const CRON_STEP_ID = definitions.CRON.stepId
+const Runner = new Thread(ThreadType.AUTOMATION)
 
 exports.processEvent = async job => {
   try {
     // need to actually await these so that an error can be captured properly
-    return await runner(job)
+    return await Runner.run(job)
   } catch (err) {
     console.error(
       `${job.data.automation.appId} automation ${job.data.automation._id} was unable to run - ${err}`

packages/server/src/threads/automation.js

@@ -1,5 +1,5 @@
-const actions = require("./actions")
-const automationUtils = require("./automationUtils")
+const actions = require("../automations/actions")
+const automationUtils = require("../automations/automationUtils")
 const AutomationEmitter = require("../events/AutomationEmitter")
 const { processObject } = require("@budibase/string-templates")
 const { DEFAULT_TENANT_ID } = require("@budibase/auth").constants
@@ -119,10 +119,17 @@ class Orchestrator {
   }
 }
 
-module.exports = async job => {
+module.exports = (input, callback) => {
   const automationOrchestrator = new Orchestrator(
-    job.data.automation,
-    job.data.event
+    input.data.automation,
+    input.data.event
   )
-  return automationOrchestrator.execute()
+  automationOrchestrator
+    .execute()
+    .then(response => {
+      callback(null, response)
+    })
+    .catch(err => {
+      callback(err)
+    })
 }

packages/server/src/threads/index.js

@@ -0,0 +1,49 @@
+const workerFarm = require("worker-farm")
+
+const ThreadType = {
+  QUERY: "query",
+  AUTOMATION: "automation",
+}
+
+function typeToFile(type) {
+  let filename = null
+  switch (type) {
+    case ThreadType.QUERY:
+      filename = "./query"
+      break
+    case ThreadType.AUTOMATION:
+      filename = "./automation"
+      break
+    default:
+      throw "Unknown thread type"
+  }
+  return require.resolve(filename)
+}
+
+class Thread {
+  constructor(type, opts = { timeoutMs: null, count: 1 }) {
+    const workerOpts = {
+      autoStart: true,
+      maxConcurrentWorkers: opts.count ? opts.count : 1,
+    }
+    if (opts.timeoutMs) {
+      workerOpts.maxCallTime = opts.timeoutMs
+    }
+    this.workers = workerFarm(workerOpts, typeToFile(type))
+  }
+
+  run(data) {
+    return new Promise((resolve, reject) => {
+      this.workers(data, (err, response) => {
+        if (err) {
+          reject(err)
+        } else {
+          resolve(response)
+        }
+      })
+    })
+  }
+}
+
+module.exports.Thread = Thread
+module.exports.ThreadType = ThreadType

packages/server/src/threads/query.js

@@ -1,5 +1,5 @@
-const ScriptRunner = require("../scriptRunner")
-const { integrations } = require("../../integrations")
+const ScriptRunner = require("../utilities/scriptRunner")
+const { integrations } = require("../integrations")
 
 function formatResponse(resp) {
   if (typeof resp === "string") {

packages/server/src/utilities/queryRunner/index.js

@@ -1,31 +0,0 @@
-const workerFarm = require("worker-farm")
-const MAX_WORKER_TIME_MS = 10000
-const workers = workerFarm(
-  {
-    autoStart: true,
-    maxConcurrentWorkers: 1,
-    maxCallTime: MAX_WORKER_TIME_MS,
-  },
-  require.resolve("./runner")
-)
-
-function runService(data) {
-  return new Promise((resolve, reject) => {
-    workers(data, (err, response) => {
-      if (err) {
-        reject(err)
-      } else {
-        resolve(response)
-      }
-    })
-  })
-}
-
-module.exports = async (datasource, queryVerb, query, transformer) => {
-  return runService({
-    datasource,
-    queryVerb,
-    query,
-    transformer,
-  })
-}

packages/server/src/utilities/scriptRunner.js

@@ -1,10 +1,13 @@
 const fetch = require("node-fetch")
 const { VM, VMScript } = require("vm2")
+const JS_TIMEOUT_MS = 1000
 
 class ScriptRunner {
   constructor(script, context) {
     const code = `let fn = () => {\n${script}\n}; results.out = fn();`
-    this.vm = new VM()
+    this.vm = new VM({
+      timeout: JS_TIMEOUT_MS,
+    })
     this.results = { out: "" }
     this.vm.setGlobals(context)
     this.vm.setGlobal("fetch", fetch)