Fixing open handle issue - now that the worker has access to queues needs to shut them down.

packages/backend-core/src/queue/queue.ts

@@ -40,8 +40,10 @@ export function createQueue<T>(
 }
 
 export async function shutdown() {
-  if (QUEUES.length) {
+  if (cleanupInterval) {
     clearInterval(cleanupInterval)
+  }
+  if (QUEUES.length) {
     for (let queue of QUEUES) {
       await queue.close()
     }

packages/worker/src/api/routes/global/tests/auditLogs.spec.ts

@@ -1,111 +1,111 @@
-// import { mocks, structures } from "@budibase/backend-core/tests"
+import { mocks, structures } from "@budibase/backend-core/tests"
-// import { context, events } from "@budibase/backend-core"
+import { context, events } from "@budibase/backend-core"
-// import { Event, IdentityType } from "@budibase/types"
+import { Event, IdentityType } from "@budibase/types"
-// import { TestConfiguration } from "../../../../tests"
+import { TestConfiguration } from "../../../../tests"
-//
+
-// mocks.licenses.useAuditLogs()
+mocks.licenses.useAuditLogs()
-//
+
-// const BASE_IDENTITY = {
+const BASE_IDENTITY = {
-//   account: undefined,
+  account: undefined,
-//   type: IdentityType.USER,
+  type: IdentityType.USER,
-// }
+}
-// const USER_AUDIT_LOG_COUNT = 3
+const USER_AUDIT_LOG_COUNT = 3
-// const APP_ID = "app_1"
+const APP_ID = "app_1"
-//
+
-// describe("/api/global/auditlogs", () => {
+describe("/api/global/auditlogs", () => {
-//   const config = new TestConfiguration()
+  const config = new TestConfiguration()
-//
+
-//   beforeAll(async () => {
+  beforeAll(async () => {
-//     await config.beforeAll()
+    await config.beforeAll()
-//   })
+  })
-//
+
-//   afterAll(async () => {
+  afterAll(async () => {
-//     await config.afterAll()
+    await config.afterAll()
-//   })
+  })
-//
+
-//   describe("POST /api/global/auditlogs/search", () => {
+  describe("POST /api/global/auditlogs/search", () => {
-//     it("should be able to fire some events (create audit logs)", async () => {
+    it("should be able to fire some events (create audit logs)", async () => {
-//       await context.doInTenant(config.tenantId, async () => {
+      await context.doInTenant(config.tenantId, async () => {
-//         const userId = config.user!._id!
+        const userId = config.user!._id!
-//         const identity = {
+        const identity = {
-//           ...BASE_IDENTITY,
+          ...BASE_IDENTITY,
-//           _id: userId,
+          _id: userId,
-//           tenantId: config.tenantId,
+          tenantId: config.tenantId,
-//         }
+        }
-//         await context.doInIdentityContext(identity, async () => {
+        await context.doInIdentityContext(identity, async () => {
-//           for (let i = 0; i < USER_AUDIT_LOG_COUNT; i++) {
+          for (let i = 0; i < USER_AUDIT_LOG_COUNT; i++) {
-//             await events.user.created(structures.users.user())
+            await events.user.created(structures.users.user())
-//           }
+          }
-//           await context.doInAppContext(APP_ID, async () => {
+          await context.doInAppContext(APP_ID, async () => {
-//             await events.app.created(structures.apps.app(APP_ID))
+            await events.app.created(structures.apps.app(APP_ID))
-//           })
+          })
-//           // fetch the user created events
+          // fetch the user created events
-//           const response = await config.api.auditLogs.search({
+          const response = await config.api.auditLogs.search({
-//             events: [Event.USER_CREATED],
+            events: [Event.USER_CREATED],
-//           })
+          })
-//           expect(response.data).toBeDefined()
+          expect(response.data).toBeDefined()
-//           // there will be an initial event which comes from the default user creation
+          // there will be an initial event which comes from the default user creation
-//           expect(response.data.length).toBe(USER_AUDIT_LOG_COUNT + 1)
+          expect(response.data.length).toBe(USER_AUDIT_LOG_COUNT + 1)
-//         })
+        })
-//       })
+      })
-//     })
+    })
-//
+
-//     it("should be able to search by event", async () => {
+    it("should be able to search by event", async () => {
-//       const response = await config.api.auditLogs.search({
+      const response = await config.api.auditLogs.search({
-//         events: [Event.USER_CREATED],
+        events: [Event.USER_CREATED],
-//       })
+      })
-//       expect(response.data.length).toBeGreaterThan(0)
+      expect(response.data.length).toBeGreaterThan(0)
-//       for (let log of response.data) {
+      for (let log of response.data) {
-//         expect(log.event).toBe(Event.USER_CREATED)
+        expect(log.event).toBe(Event.USER_CREATED)
-//       }
+      }
-//     })
+    })
-//
+
-//     it("should be able to search by time range (frozen)", async () => {
+    it("should be able to search by time range (frozen)", async () => {
-//       // this is frozen, only need to add 1 and minus 1
+      // this is frozen, only need to add 1 and minus 1
-//       const now = new Date()
+      const now = new Date()
-//       const start = new Date()
+      const start = new Date()
-//       start.setSeconds(now.getSeconds() - 1)
+      start.setSeconds(now.getSeconds() - 1)
-//       const end = new Date()
+      const end = new Date()
-//       end.setSeconds(now.getSeconds() + 1)
+      end.setSeconds(now.getSeconds() + 1)
-//       const response = await config.api.auditLogs.search({
+      const response = await config.api.auditLogs.search({
-//         startDate: start.toISOString(),
+        startDate: start.toISOString(),
-//         endDate: end.toISOString(),
+        endDate: end.toISOString(),
-//       })
+      })
-//       expect(response.data.length).toBeGreaterThan(0)
+      expect(response.data.length).toBeGreaterThan(0)
-//       for (let log of response.data) {
+      for (let log of response.data) {
-//         expect(log.timestamp).toBe(now.toISOString())
+        expect(log.timestamp).toBe(now.toISOString())
-//       }
+      }
-//     })
+    })
-//
+
-//     it("should be able to search by user ID", async () => {
+    it("should be able to search by user ID", async () => {
-//       const userId = config.user!._id!
+      const userId = config.user!._id!
-//       const response = await config.api.auditLogs.search({
+      const response = await config.api.auditLogs.search({
-//         userIds: [userId],
+        userIds: [userId],
-//       })
+      })
-//       expect(response.data.length).toBeGreaterThan(0)
+      expect(response.data.length).toBeGreaterThan(0)
-//       for (let log of response.data) {
+      for (let log of response.data) {
-//         expect(log.user._id).toBe(userId)
+        expect(log.user._id).toBe(userId)
-//       }
+      }
-//     })
+    })
-//
+
-//     it("should be able to search by app ID", async () => {
+    it("should be able to search by app ID", async () => {
-//       const response = await config.api.auditLogs.search({
+      const response = await config.api.auditLogs.search({
-//         appIds: [APP_ID],
+        appIds: [APP_ID],
-//       })
+      })
-//       expect(response.data.length).toBeGreaterThan(0)
+      expect(response.data.length).toBeGreaterThan(0)
-//       for (let log of response.data) {
+      for (let log of response.data) {
-//         expect(log.app?._id).toBe(APP_ID)
+        expect(log.app?._id).toBe(APP_ID)
-//       }
+      }
-//     })
+    })
-//
+
-//     it("should be able to search by full string", async () => {
+    it("should be able to search by full string", async () => {
-//       const response = await config.api.auditLogs.search({
+      const response = await config.api.auditLogs.search({
-//         fullSearch: "User",
+        fullSearch: "User",
-//       })
+      })
-//       expect(response.data.length).toBeGreaterThan(0)
+      expect(response.data.length).toBeGreaterThan(0)
-//       for (let log of response.data) {
+      for (let log of response.data) {
-//         expect(log.name.includes("User")).toBe(true)
+        expect(log.name.includes("User")).toBe(true)
-//       }
+      }
-//     })
+    })
-//   })
+  })
-// })
+})

packages/worker/src/index.ts

@@ -13,8 +13,8 @@ import { Event } from "@sentry/types/dist/event"
 import Application from "koa"
 import { bootstrap } from "global-agent"
 import * as db from "./db"
-import { auth, logging, events, middleware } from "@budibase/backend-core"
+import { auth, logging, events, middleware, queue } from "@budibase/backend-core"
-import { sdk as proSdk, sdk } from "@budibase/pro"
+import { sdk as proSdk } from "@budibase/pro"
 db.init()
 import Koa from "koa"
 import koaBody from "koa-body"
@@ -86,6 +86,7 @@ server.on("close", async () => {
   console.log("Server Closed")
   await redis.shutdown()
   await events.shutdown()
+  await queue.shutdown()
   if (!env.isTest()) {
     process.exit(errCode)
   }