refactor(frontend): custom mutex implementation

Lets us track priorities of tasks without cancellation.

4 files changed, 83 insertions, 5 deletions

diff --git a/subprojects/frontend/src/utils/CancelledError.ts b/subprojects/frontend/src/utils/CancelledError.ts
new file mode 100644
index 00000000..8d3e55d8
--- /dev/null
+++ b/subprojects/frontend/src/utils/CancelledError.ts
@@ -0,0 +1,5 @@
+export default class CancelledError extends Error {
+  constructor() {
+    super('Operation cancelled');
+  }
+}
diff --git a/subprojects/frontend/src/utils/PendingTask.ts b/subprojects/frontend/src/utils/PendingTask.ts
index 205c8452..d0b24c1f 100644
--- a/subprojects/frontend/src/utils/PendingTask.ts
+++ b/subprojects/frontend/src/utils/PendingTask.ts
@@ -1,3 +1,4 @@
+import TimeoutError from './TimeoutError';
 import getLogger from './getLogger';
 
 const log = getLogger('utils.PendingTask');
@@ -15,16 +16,14 @@ export default class PendingTask<T> {
     resolveCallback: (value: T) => void,
     rejectCallback: (reason?: unknown) => void,
     timeoutMs: number | undefined,
-    timeoutCallback: () => void | undefined,
+    timeoutCallback?: (() => void) | undefined,
   ) {
     this.resolveCallback = resolveCallback;
     this.rejectCallback = rejectCallback;
     this.timeout = setTimeout(() => {
       if (!this.resolved) {
-        this.reject(new Error('Request timed out'));
-        if (timeoutCallback) {
-          timeoutCallback();
-        }
+        this.reject(new TimeoutError());
+        timeoutCallback?.();
       }
     }, timeoutMs);
   }
diff --git a/subprojects/frontend/src/utils/PriorityMutex.ts b/subprojects/frontend/src/utils/PriorityMutex.ts
new file mode 100644
index 00000000..78736141
--- /dev/null
+++ b/subprojects/frontend/src/utils/PriorityMutex.ts
@@ -0,0 +1,69 @@
+import CancelledError from './CancelledError';
+import PendingTask from './PendingTask';
+import getLogger from './getLogger';
+
+const log = getLogger('utils.PriorityMutex');
+
+export default class PriorityMutex {
+  private readonly lowPriorityQueue: PendingTask<void>[] = [];
+
+  private readonly highPriorityQueue: PendingTask<void>[] = [];
+
+  private _locked = false;
+
+  constructor(private readonly timeout: number) {}
+
+  get locked(): boolean {
+    return this._locked;
+  }
+
+  async runExclusive<T>(
+    callback: () => Promise<T>,
+    highPriority = false,
+  ): Promise<T> {
+    await this.acquire(highPriority);
+    try {
+      return await callback();
+    } finally {
+      this.release();
+    }
+  }
+
+  cancelAllWaiting(): void {
+    [this.highPriorityQueue, this.lowPriorityQueue].forEach((queue) =>
+      queue.forEach((task) => task.reject(new CancelledError())),
+    );
+  }
+
+  private acquire(highPriority: boolean): Promise<void> {
+    if (!this.locked) {
+      this._locked = true;
+      return Promise.resolve();
+    }
+    const queue = highPriority ? this.highPriorityQueue : this.lowPriorityQueue;
+    return new Promise((resolve, reject) => {
+      const task = new PendingTask(resolve, reject, this.timeout, () => {
+        const index = queue.indexOf(task);
+        if (index < 0) {
+          log.error('Timed out task already removed from queue');
+          return;
+        }
+        queue.splice(index, 1);
+      });
+      queue.push(task);
+    });
+  }
+
+  private release(): void {
+    if (!this.locked) {
+      throw new Error('Trying to release already released mutext');
+    }
+    const task =
+      this.highPriorityQueue.shift() ?? this.lowPriorityQueue.shift();
+    if (task === undefined) {
+      this._locked = false;
+      return;
+    }
+    task.resolve();
+  }
+}
diff --git a/subprojects/frontend/src/utils/TimeoutError.ts b/subprojects/frontend/src/utils/TimeoutError.ts
new file mode 100644
index 00000000..eb800f40
--- /dev/null
+++ b/subprojects/frontend/src/utils/TimeoutError.ts
@@ -0,0 +1,5 @@
+export default class TimeoutError extends Error {
+  constructor() {
+    super('Operation timed out');
+  }
+}