Merge pull request #13908 from Budibase/fix/app-migrations-better-startup

Better app migration queue startup process

packages/server/src/appMigrations/index.ts

@@ -1,4 +1,4 @@
-import queue from "./queue"
+import { getAppMigrationQueue } from "./queue"
 import { Next } from "koa"
 import { getAppMigrationVersion } from "./appMigrationMetadata"
 import { MIGRATIONS } from "./migrations"
@@ -37,8 +37,10 @@ export async function checkMissingMigrations(
 ) {
   const currentVersion = await getAppMigrationVersion(appId)
   const latestMigration = getLatestEnabledMigrationId()
+  const queue = getAppMigrationQueue()
 
   if (
+    queue &&
     latestMigration &&
     getTimestamp(currentVersion) < getTimestamp(latestMigration)
   ) {

packages/server/src/appMigrations/queue.ts

@@ -4,25 +4,37 @@ import { MIGRATIONS } from "./migrations"
 import { processMigrations } from "./migrationsProcessor"
 import { apiEnabled } from "../features"
 
-const MAX_ATTEMPTS = 3
+const MAX_ATTEMPTS = 1
 
-const appMigrationQueue = queue.createQueue(queue.JobQueue.APP_MIGRATION, {
-  jobOptions: {
-    attempts: MAX_ATTEMPTS,
-    removeOnComplete: true,
-    removeOnFail: true,
-  },
-  maxStalledCount: MAX_ATTEMPTS,
-  removeStalledCb: async (job: Job) => {
-    logging.logAlert(
-      `App migration failed, queue job ID: ${job.id} - reason: ${job.failedReason}`
-    )
-  },
-})
+export type AppMigrationJob = {
+  appId: string
+}
 
-// only run app migrations in main API services
-if (apiEnabled()) {
-  appMigrationQueue.process(processMessage)
+let appMigrationQueue: queue.Queue<AppMigrationJob> | undefined
+
+export function init() {
+  // only run app migrations in main API services
+  if (!apiEnabled()) {
+    return
+  }
+  appMigrationQueue = queue.createQueue<AppMigrationJob>(
+    queue.JobQueue.APP_MIGRATION,
+    {
+      jobOptions: {
+        attempts: MAX_ATTEMPTS,
+        removeOnComplete: true,
+        removeOnFail: true,
+      },
+      maxStalledCount: MAX_ATTEMPTS,
+      removeStalledCb: async (job: Job) => {
+        logging.logAlert(
+          `App migration failed, queue job ID: ${job.id} - reason: ${job.failedReason}`
+        )
+      },
+    }
+  )
+
+  return appMigrationQueue.process(processMessage)
 }
 
 async function processMessage(job: Job) {
@@ -31,4 +43,6 @@ async function processMessage(job: Job) {
   await processMigrations(appId, MIGRATIONS)
 }
 
-export default appMigrationQueue
+export function getAppMigrationQueue() {
+  return appMigrationQueue
+}

packages/server/src/automations/bullboard.ts

@@ -3,6 +3,7 @@ import { KoaAdapter } from "@bull-board/koa"
 import { queue } from "@budibase/backend-core"
 import * as automation from "../threads/automation"
 import { backups } from "@budibase/pro"
+import { getAppMigrationQueue } from "../appMigrations/queue"
 import { createBullBoard } from "@bull-board/api"
 import BullQueue from "bull"
 
@@ -16,10 +17,14 @@ const PATH_PREFIX = "/bulladmin"
 export async function init() {
   // Set up queues for bull board admin
   const backupQueue = backups.getBackupQueue()
+  const appMigrationQueue = getAppMigrationQueue()
   const queues = [automationQueue]
   if (backupQueue) {
     queues.push(backupQueue)
   }
+  if (appMigrationQueue) {
+    queues.push(appMigrationQueue)
+  }
   const adapters = []
   const serverAdapter: any = new KoaAdapter()
   for (let queue of queues) {

packages/server/src/startup/index.ts

@@ -15,6 +15,7 @@ import * as fileSystem from "../utilities/fileSystem"
 import { default as eventEmitter, init as eventInit } from "../events"
 import * as migrations from "../migrations"
 import * as bullboard from "../automations/bullboard"
+import * as appMigrations from "../appMigrations/queue"
 import * as pro from "@budibase/pro"
 import * as api from "../api"
 import sdk from "../sdk"
@@ -114,6 +115,7 @@ export async function startup(
   // configure events to use the pro audit log write
   // can't integrate directly into backend-core due to cyclic issues
   queuePromises.push(events.processors.init(pro.sdk.auditLogs.write))
+  queuePromises.push(appMigrations.init())
   if (automationsEnabled()) {
     queuePromises.push(automations.init())
   }