refactor: Improve locking mechanism in Semaphore class

- Modify the Semaphore class in Semaphore.ts to include a namespace parameter in the lock method.
- Update the lockTimeout default value from 1000 to 5000 milliseconds.
- Append the namespace to the key when creating a new Mutex instance.
- Add a try-catch block to handle errors when acquiring and releasing the mutex.

Common/Server/Infrastructure/Semaphore.ts

@@ -7,10 +7,11 @@ export default class Semaphore {
  // returns the mutex id
  public static async lock(data: {
  key: string;
+ namespace: string;
  lockTimeout?: number;
  }): Promise<SemaphoreMutex> {
  if (!data.lockTimeout) {
- data.lockTimeout = 1000;
+ data.lockTimeout = 5000;
  }
 
  const { key } = data;
@@ -21,9 +22,13 @@ export default class Semaphore {
  throw new Error("Redis client is not connected");
  }
 
- const mutex: SemaphoreMutex = new Mutex(client, key, {
- lockTimeout: data.lockTimeout,
- });
+ const mutex: SemaphoreMutex = new Mutex(
+ client,
+ data.namespace + "-" + key,
+ {
+ lockTimeout: data.lockTimeout,
+ },
+ );
 
  await mutex.acquire();
 

Common/Server/Utils/Monitor/MonitorResource.ts

@@ -50,11 +50,23 @@ import { TelemetryQuery } from "../../../Types/Telemetry/TelemetryQuery";
 import MonitorIncident from "./MonitorIncident";
 import MonitorAlert from "./MonitorAlert";
 import MonitorStatusTimelineUtil from "./MonitorStatusTimeline";
+import Semaphore, { SemaphoreMutex } from "../../Infrastructure/Semaphore";
 
 export default class MonitorResourceUtil {
  public static async monitorResource(
  dataToProcess: DataToProcess,
  ): Promise<ProbeApiIngestResponse> {
+ let mutex: SemaphoreMutex | null = null;
+
+ try {
+ mutex = await Semaphore.lock({
+ key: dataToProcess.monitorId.toString(),
+ namespace: "MonitorResourceUtil.monitorResource",
+ });
+ } catch (err) {
+ logger.error(err);
+ }
+
  let response: ProbeApiIngestResponse = {
  monitorId: dataToProcess.monitorId,
  criteriaMetId: undefined,
@@ -532,6 +544,14 @@ export default class MonitorResourceUtil {
  }
  }
 
+ if (mutex) {
+ try {
+ await Semaphore.release(mutex);
+ } catch (err) {
+ logger.error(err);
+ }
+ }
+
  return response;
  }
 

Ingestor/API/Monitor.ts

@@ -324,6 +324,7 @@ router.post(
  try {
  mutex = await Semaphore.lock({
  key: probeId.toString(),
+ namespace: "MonitorAPI.monitor-list",
  });
  } catch (err) {
  logger.error(err);

Ingestor/API/Probe.ts

@@ -38,6 +38,7 @@ router.post(
  ): Promise<void> => {
  try {
  const data: JSONObject = req.body;
+
  const statusReport: ProbeStatusReport = JSONFunctions.deserialize(
  (data as JSONObject)["statusReport"] as any,
  ) as any;