feat(worker): check all targets in parellel, reduces ~15% cpu time usage in test, also try to a fix rare race condition

worker/package-lock.json

@@ -7,6 +7,9 @@
     "": {
       "name": "uptimeworker",
       "version": "0.0.0",
+      "dependencies": {
+        "p-limit": "^7.2.0"
+      },
       "devDependencies": {
         "@cloudflare/workers-types": "^4.20250410.0",
         "typescript": "^5.0.4",
@@ -1299,6 +1302,21 @@
         "node": ">=18.0.0"
       }
     },
+    "node_modules/p-limit": {
+      "version": "7.2.0",
+      "resolved": "https://registry.npmjs.org/p-limit/-/p-limit-7.2.0.tgz",
+      "integrity": "sha512-ATHLtwoTNDloHRFFxFJdHnG6n2WUeFjaR8XQMFdKIv0xkXjrER8/iG9iu265jOM95zXHAfv9oTkqhrfbIzosrQ==",
+      "license": "MIT",
+      "dependencies": {
+        "yocto-queue": "^1.2.1"
+      },
+      "engines": {
+        "node": ">=20"
+      },
+      "funding": {
+        "url": "https://github.com/sponsors/sindresorhus"
+      }
+    },
     "node_modules/path-to-regexp": {
       "version": "6.3.0",
       "resolved": "https://registry.npmjs.org/path-to-regexp/-/path-to-regexp-6.3.0.tgz",
@@ -1519,6 +1537,18 @@
         }
       }
     },
+    "node_modules/yocto-queue": {
+      "version": "1.2.2",
+      "resolved": "https://registry.npmjs.org/yocto-queue/-/yocto-queue-1.2.2.tgz",
+      "integrity": "sha512-4LCcse/U2MHZ63HAJVE+v71o7yOdIe4cZ70Wpf8D/IyjDKYQLV5GD46B+hSTjJsvV5PztjvHoU580EftxjDZFQ==",
+      "license": "MIT",
+      "engines": {
+        "node": ">=12.20"
+      },
+      "funding": {
+        "url": "https://github.com/sponsors/sindresorhus"
+      }
+    },
     "node_modules/youch": {
       "version": "4.1.0-beta.10",
       "resolved": "https://registry.npmjs.org/youch/-/youch-4.1.0-beta.10.tgz",
@@ -2196,6 +2226,14 @@
         "zod": "3.22.3"
       }
     },
+    "p-limit": {
+      "version": "7.2.0",
+      "resolved": "https://registry.npmjs.org/p-limit/-/p-limit-7.2.0.tgz",
+      "integrity": "sha512-ATHLtwoTNDloHRFFxFJdHnG6n2WUeFjaR8XQMFdKIv0xkXjrER8/iG9iu265jOM95zXHAfv9oTkqhrfbIzosrQ==",
+      "requires": {
+        "yocto-queue": "^1.2.1"
+      }
+    },
     "path-to-regexp": {
       "version": "6.3.0",
       "resolved": "https://registry.npmjs.org/path-to-regexp/-/path-to-regexp-6.3.0.tgz",
@@ -2330,6 +2368,11 @@
       "dev": true,
       "requires": {}
     },
+    "yocto-queue": {
+      "version": "1.2.2",
+      "resolved": "https://registry.npmjs.org/yocto-queue/-/yocto-queue-1.2.2.tgz",
+      "integrity": "sha512-4LCcse/U2MHZ63HAJVE+v71o7yOdIe4cZ70Wpf8D/IyjDKYQLV5GD46B+hSTjJsvV5PztjvHoU580EftxjDZFQ=="
+    },
     "youch": {
       "version": "4.1.0-beta.10",
       "resolved": "https://registry.npmjs.org/youch/-/youch-4.1.0-beta.10.tgz",

worker/package.json

@@ -11,5 +11,8 @@
     "@cloudflare/workers-types": "^4.20250410.0",
     "typescript": "^5.0.4",
     "wrangler": "^4.54.0"
+  },
+  "dependencies": {
+    "p-limit": "^7.2.0"
   }
 }

worker/src/index.ts

@@ -1,12 +1,12 @@
 import { DurableObject } from 'cloudflare:workers'
 import { MonitorTarget } from '../../types/config'
 import { workerConfig } from '../../uptime.config'
-import { getStatus, getStatusWithGlobalPing } from './monitor'
+import { doMonitor, getStatus } from './monitor'
 import { formatAndNotify, getWorkerLocation } from './util'
 import { CompactedMonitorStateWrapper, getFromStore, setToStore } from './store'
+import pLimit from 'p-limit'
 
 export interface Env {
-  UPTIMEFLARE_STATE: KVNamespace
   REMOTE_CHECKER_DO: DurableObjectNamespace<RemoteChecker>
   UPTIMEFLARE_D1: D1Database
 }
@@ -24,62 +24,25 @@ const Worker = {
     let statusChanged = false
     const currentTimeSecond = Math.round(Date.now() / 1000)
 
-    // Check each monitor
+    // Parallel check multiple monitors
-    // TODO: concurrent status check
+    // Max concurrent connection is 6 limited by Cloudflare Workers, we use 5 here to be safe
+    type CheckResult = { id: string; location: string; status: { ping: number; up: boolean; err: string } }
+    let checkQueue: Promise<CheckResult>[] = []
+    let checkResult: Record<string, CheckResult> = {};
+    const limit = pLimit(5);
     for (const monitor of workerConfig.monitors) {
-      console.log(`[${workerLocation}] Checking ${monitor.name}...`)
+      checkQueue.push(limit(() => doMonitor(monitor, workerLocation, env)))
+    }
+    for (const result of await Promise.all(checkQueue)) {
+      checkResult[result.id] = result
+    }
+
+    // Update each monitor's state based on check results
+    for (const monitor of workerConfig.monitors) {
+      console.log(`Processing monitor result: ${monitor.name} (${monitor.id})`)
 
       let monitorStatusChanged = false
-      let checkLocation = workerLocation
+      const { location: checkLocation, status } = checkResult[monitor.id]
-      let status
-
-      if (monitor.checkProxy) {
-        // Initiate a check using proxy (Geo-specific monitoring)
-        try {
-          console.log('Calling check proxy: ' + monitor.checkProxy)
-          let resp
-          if (monitor.checkProxy.startsWith('worker://')) {
-            const doLoc = monitor.checkProxy.replace('worker://', '')
-            const doId = env.REMOTE_CHECKER_DO.idFromName(doLoc)
-            const doStub = env.REMOTE_CHECKER_DO.get(doId, {
-              locationHint: doLoc as DurableObjectLocationHint,
-            })
-            resp = await doStub.getLocationAndStatus(monitor)
-            try {
-              // Kill the DO instance after use, to avoid extra resource usage
-              await doStub.kill()
-            } catch (err) {
-              // An error here is expected, ignore it
-            }
-          } else if (monitor.checkProxy.startsWith('globalping://')) {
-            resp = await getStatusWithGlobalPing(monitor)
-          } else {
-            resp = await (
-              await fetch(monitor.checkProxy, {
-                method: 'POST',
-                headers: { 'Content-Type': 'application/json' },
-                body: JSON.stringify(monitor),
-              })
-            ).json<{ location: string; status: { ping: number; up: boolean; err: string } }>()
-          }
-          checkLocation = resp.location
-          status = resp.status
-        } catch (err) {
-          console.log('Error calling proxy: ' + err)
-          if (monitor.checkProxyFallback) {
-            console.log('Falling back to local check...')
-            status = await getStatus(monitor)
-          } else {
-            // TODO: more consistent error handling (throw or return?)
-            status = { ping: 0, up: false, err: 'Unknown check proxy error' }
-          }
-        }
-      } else {
-        // Initiate a check from the current location
-        status = await getStatus(monitor)
-      }
-
-      const currentTimeSecond = Math.round(Date.now() / 1000)
 
       // Update counters
       status.up ? state.data.overallUp++ : state.data.overallDown++

worker/src/monitor.ts

@@ -1,3 +1,4 @@
+import { Env } from '.'
 import { MonitorTarget } from '../../types/config'
 import { withTimeout, fetchTimeout } from './util'
 
@@ -352,3 +353,62 @@ export async function getStatus(
 
   return status
 }
+
+export async function doMonitor(monitor: MonitorTarget, defaultLocation: string, env: Env) {
+  let checkLocation = defaultLocation
+  let status
+
+  if (monitor.checkProxy) {
+    // Initiate a check using proxy (Geo-specific monitoring)
+    try {
+      console.log(`[${monitor.id}] Calling check proxy: ${monitor.checkProxy}`)
+      let resp
+      if (monitor.checkProxy.startsWith('worker://')) {
+        const doLoc = monitor.checkProxy.replace('worker://', '')
+        const doId = env.REMOTE_CHECKER_DO.idFromName(monitor.id)
+        const doStub = env.REMOTE_CHECKER_DO.get(doId, {
+          locationHint: doLoc as DurableObjectLocationHint,
+        })
+        resp = await doStub.getLocationAndStatus(monitor)
+        try {
+          // Kill the DO instance after use, to avoid extra resource usage
+          await doStub.kill()
+        } catch (err) {
+          // An error here is expected, ignore it
+        }
+      } else if (monitor.checkProxy.startsWith('globalping://')) {
+        resp = await getStatusWithGlobalPing(monitor)
+      } else {
+        resp = await (
+          await fetch(monitor.checkProxy, {
+            method: 'POST',
+            headers: { 'Content-Type': 'application/json' },
+            body: JSON.stringify(monitor),
+          })
+        ).json<{ location: string; status: { ping: number; up: boolean; err: string } }>()
+      }
+      checkLocation = resp.location
+      status = resp.status
+    } catch (err) {
+      console.log(`[${monitor.id}] Error calling proxy: ${err}`)
+      if (monitor.checkProxyFallback) {
+        console.log('Falling back to local check...')
+        status = await getStatus(monitor)
+      } else {
+        // TODO: more consistent error handling (throw or return?)
+        status = { ping: 0, up: false, err: 'Unknown check proxy error' }
+      }
+    }
+  } else {
+    // Initiate a check from the current location
+    status = await getStatus(monitor)
+  }
+
+  console.log(`[${monitor.id}] Check result from ${checkLocation}: up=${status.up}, ping=${status.ping}, err=${status.err}`)
+
+  return {
+    location: checkLocation,
+    status,
+    id: monitor.id,
+  }
+}