Feat: Probing with Queue

package.json

@@ -101,6 +101,7 @@
     "pino-pretty": "10.0.1",
     "piscina": "^4.1.0",
     "prom-client": "13.1.0",
+    "queue": "^6.0.2",
     "redis": "^4.3.0",
     "smtp-server": "^3.8.0",
     "sqlite": "^4.0.21",

src/components/probe/index.ts

@@ -27,6 +27,7 @@ import { differenceInSeconds } from 'date-fns'
 import { getContext } from '../../context'
 import type { Notification } from '@hyperjumptech/monika-notification'
 import type { Probe } from '../../interfaces/probe'
+import Queue from 'queue'
 import {
   getProbeContext,
   getProbeState,
@@ -48,6 +49,13 @@ type doProbeParams = {
  * @param {object} doProbeParams doProbe parameter
  * @returns {Promise<void>} void
  */
+
+const queue = new Queue({
+  concurrency: 1,
+  autostart: true,
+  timeout: 10_000,
+})
+
 export async function doProbe({
   probe,
   notifications,
@@ -77,15 +85,18 @@ export async function doProbe({
       probe.recoveryThreshold || DEFAULT_RECOVERY_THRESHOLD
     )
 
-    await retry(handleAll, {
-      maxAttempts,
-      backoff: new ExponentialBackoff({
-        initialDelay: getContext().flags.retryInitialDelayMs,
-        maxDelay: getContext().flags.retryMaxDelayMs,
-      }),
-    }).execute(({ attempt }) =>
-      Promise.all(probers.map(async (prober) => prober.probe(attempt)))
-    )
+    // Enqueue each prober into the prober queue with retry backoff
+    queue.push(async () => {
+      await retry(handleAll, {
+        maxAttempts,
+        backoff: new ExponentialBackoff({
+          initialDelay: getContext().flags.retryInitialDelayMs,
+          maxDelay: getContext().flags.retryMaxDelayMs,
+        }),
+      }).execute(({ attempt }) =>
+        Promise.all(probers.map(async (prober) => prober.probe(attempt)))
+      )
+    })
 
     setProbeFinish(probe.id)
   }, randomTimeoutMilliseconds)

src/looper/index.ts

@@ -26,10 +26,8 @@ import type { Notification } from '@hyperjumptech/monika-notification'
 
 import { AbortSignal } from 'node-abort-controller'
 import { v4 as uuid } from 'uuid'
-
 import type { Probe, ProbeAlert } from '../interfaces/probe'
 
-import { doProbe } from '../components/probe'
 import { getContext } from '../context'
 import { log } from '../utils/pino'
 import {
@@ -42,6 +40,7 @@ import {
   DEFAULT_INCIDENT_THRESHOLD,
   DEFAULT_RECOVERY_THRESHOLD,
 } from '../components/config/validation/validator/default-values'
+import { doProbe } from '../components/probe'
 
 let checkSTUNinterval: NodeJS.Timeout