♻️ oo-connector: centralising forgotten file batch management with session key check, bit of health refactoring (#525)

tdrive/connectors/onlyoffice-connector/src/controllers/backend-callbacks.controller.ts

@@ -2,6 +2,7 @@ import { Request, Response } from 'express';
 import logger from '@/lib/logger';
 import onlyofficeService, { Callback, CommandError, ErrorCode } from '@/services/onlyoffice.service';
 import driveService from '@/services/drive.service';
+import forgottenProcessorService from '@/services/forgotten-processor.service';
 
 interface RequestQuery {
   editing_session_key: string;
@@ -28,15 +29,9 @@ export default class TwakeDriveBackendCallbackController {
     try {
       const forgottenURL = await onlyofficeService.getForgotten(req.params.editing_session_key);
       try {
-        await driveService.endEditing(req.params.editing_session_key, forgottenURL);
+        await forgottenProcessorService.processForgottenFile(req.params.editing_session_key, forgottenURL);
       } catch (error) {
-        logger.error(`endEditing failed`, { error });
-        return void res.status(502).send({ error: -57649 });
-      }
-      try {
-        await onlyofficeService.deleteForgotten(req.params.editing_session_key);
-      } catch (error) {
-        logger.error(`deleteForgotten failed`, { error });
+        logger.error(`processForgottenFile failed`, { error });
         return void res.status(502).send({ error: -57650 });
       }
       return void res.send({ status: 'updated' });

tdrive/connectors/onlyoffice-connector/src/controllers/health-status.controller.ts

@@ -9,19 +9,13 @@ import forgottenProcessorService from '@/services/forgotten-processor.service';
  */
 export const HealthStatusController = {
   async get(req: Request<{}, {}, {}, {}>, res: Response): Promise<void> {
-    Promise.all([
-      onlyofficeService.getLatestLicence(),
-      apiService.hasToken(),
-      onlyofficeService.getForgottenList(),
-      forgottenProcessorService.getLastStartTimeAgoS(),
-    ]).then(
-      ([onlyOfficeLicense, twakeDriveToken, forgottenKeys, forgottenLastProcessAgoS]) =>
+    Promise.all([onlyofficeService.getLatestLicence(), apiService.hasToken(), forgottenProcessorService.getHealthData()]).then(
+      ([onlyOfficeLicense, twakeDriveToken, forgottenProcessorHealth]) =>
         res.status(onlyOfficeLicense && twakeDriveToken ? 200 : 500).send({
           uptimeS: Math.floor(process.uptime()),
           onlyOfficeLicense,
           twakeDriveToken,
-          forgottenCount: forgottenKeys?.length ?? 0,
-          forgottenLastProcessAgoS,
+          ...forgottenProcessorHealth,
         }),
       err => res.status(500).send(err),
     );

tdrive/connectors/onlyoffice-connector/src/lib/single-processor-lock.ts

@@ -0,0 +1,161 @@
+import logger from './logger';
+
+export class CannotSettleAlreadyReleasedLockError extends Error {}
+
+const debugLocks = false;
+
+/** Simplifies use of the Promise constructor function out of its scope */
+class ExplodedPromise<T> {
+  public readonly startAtMs = new Date().getTime();
+  public readonly promise: Promise<T>;
+  private _waiters = 0;
+  private _resolve: (value: T) => void;
+  private _reject: (reason?: any) => void;
+
+  constructor() {
+    this.promise = new Promise((resolve, reject) => {
+      // Works because this is garanteed to be called synchroneously
+      this._resolve = resolve;
+      this._reject = reject;
+    });
+  }
+  private check() {
+    if (!this._resolve) throw new CannotSettleAlreadyReleasedLockError(`Promise is already settled`);
+  }
+  private finish() {
+    this._resolve = this._reject = null;
+  }
+  /** Increase waiter count and return new count */
+  public addWaiter() {
+    return ++this._waiters;
+  }
+  get waiting() {
+    return this._waiters;
+  }
+  public resolve(...args: Parameters<ExplodedPromise<T>['_resolve']>): void {
+    this.check();
+    this._resolve(...args);
+    this.finish();
+  }
+  public reject(...args: Parameters<ExplodedPromise<T>['_reject']>): void {
+    this.check();
+    this._reject(...args);
+    this.finish();
+  }
+}
+
+/**
+ * Use this instance to release the lock acquired by {@link SingleProcessorLock.acquire}.
+ * Only call either `resolve` or `reject` once or they will
+ * throw a {@link CannotSettleAlreadyReleasedLockError}.
+ */
+export interface LockReleaser<T> {
+  /** To distinguish return from {@link LockWaiter} */
+  didAcquire: true;
+  /** This is the same Promise that `resolve` and `reject` will settle, and waiters get */
+  promise: LockWaiter<T>['promise'];
+  /** Release the lock and resolve the promise of all the waiters */
+  resolve: ExplodedPromise<T>['resolve'];
+  /** Release the lock and reject the promise of all the waiters */
+  reject: ExplodedPromise<T>['reject'];
+}
+
+/** Use this instance to wait if {@link SingleProcessorLock.acquire} was already processing that key */
+export interface LockWaiter<T> {
+  /** To distinguish return from {@link LockReleaser} */
+  didAcquire: false;
+  /** Number of waiters in the queue at the start of this one, including this one (ie. first to wait is `1`) */
+  numberInQueue: number;
+  /** Promise that must be waited for after the acquire failed. Will settle as per called by the {@see LockReleaser} */
+  promise: Promise<T>;
+}
+
+/**
+ * Create an synchronisation primitive that permits a single caller
+ * to process for a given `key`. Other callers with the same `key`
+ * will receive an `Promise` that should be waited for. When the
+ * processor caller releases the lock, the promise is settled accordingly.
+ */
+export function createSingleProcessorLock<T>() {
+  const promisesByKey = new Map<string, ExplodedPromise<T>>();
+
+  /**
+   * The first caller for the given `key` is the processor that must settle the lock. The
+   * return value will be an instance of {@link LockReleaser} that can be tested because
+   * {@link LockReleaser.didAcquire} will be `true`. This caller has the responsibility to
+   * ensure a call to either {@link LockReleaser.resolve} or {@link LockReleaser.reject}.
+   * There is no other cleanup mecanism.
+   *
+   * All callers following them for the same `key` will get an instance of {@link LockWaiter},
+   * that can be tested because {@link LockWaiter.didAcquire} will be `false`. They must
+   * `await` {@link LockWaiter.promise}. It will be settled by the processor when they call
+   * the methods of {@link LockReleaser}. must be ensured, there is no cleanup mecanism.
+   *
+   * @param key Unique key to lock on
+   */
+  const acquire = (key: string): LockReleaser<T> | LockWaiter<T> => {
+    const existing = promisesByKey.get(key);
+    if (existing) {
+      const numberInQueue = existing.addWaiter();
+      debugLocks && logger.debug(`LOCK blocked ${numberInQueue}: ${key}`);
+      return {
+        didAcquire: false,
+        numberInQueue,
+        promise: existing.promise,
+      };
+    }
+    const processorPromise = new ExplodedPromise<T>();
+    promisesByKey.set(key, processorPromise);
+    debugLocks && logger.debug(`LOCK started: ${key}`);
+    return {
+      didAcquire: true,
+      promise: processorPromise.promise,
+      resolve(...args: Parameters<LockReleaser<T>['resolve']>) {
+        debugLocks && logger.debug(`LOCK resolved: ${key}`);
+        promisesByKey.delete(key);
+        processorPromise.resolve(...args);
+      },
+      reject(...args: Parameters<LockReleaser<T>['reject']>) {
+        debugLocks && logger.debug(`LOCK rejected: ${key}`);
+        promisesByKey.delete(key);
+        processorPromise.reject(...args);
+      },
+    };
+  };
+  return {
+    acquire,
+
+    /**
+     * Attempt to acquire a lock. If succesful, settle the promise with the
+     * return value of `processor`. In both cases return the Promise.
+     */
+    async runWithLock(key: string, processor: () => Promise<T>): Promise<T> {
+      const lock = acquire(key);
+      if (lock.didAcquire)
+        try {
+          lock.resolve(await processor());
+        } catch (e) {
+          lock.reject(e);
+        }
+      return lock.promise;
+    },
+
+    /** Get statistics about pending locks */
+    getWorstStats() {
+      const all = [...promisesByKey.values()];
+      const now = new Date().getTime();
+      return all.reduce(
+        ({ oldestMs, waiting, total }, cur) => ({
+          oldestMs: Math.max(oldestMs, now - cur.startAtMs),
+          waiting: Math.max(waiting, cur.waiting),
+          total: total + 1,
+        }),
+        {
+          oldestMs: 0,
+          waiting: 0,
+          total: 0,
+        },
+      );
+    },
+  };
+}

tdrive/connectors/onlyoffice-connector/src/services/drive.service.ts

@@ -5,6 +5,9 @@ import { Stream } from 'stream';
 import FormData from 'form-data';
 import { INSTANCE_ID } from '@/config';
 
+/** Thrown when Twake Drive returns a 404 for a key */
+export class UnknownKeyInDriveError extends Error {}
+
 /**
  * Client for Twake Drive's APIs dealing with `DriveItem`s, using {@see apiService}
  * to handle authorization
@@ -107,8 +110,8 @@ class DriveService implements IDriveService {
         //   the key anymore, there's not much we can do, and we should get OO
         //   to clean up and stop trying to upload it.
         //   but for today:
-        logger.fatal("Losing OO document because Twake Drive doesn't know that Key.", { editing_session_key, url });
-        throw new Error(`Unknown key ${JSON.stringify(editing_session_key)}`);
+        logger.error("Forgotten OO document that Twake Drive doesn't know the key of.", { editing_session_key, url });
+        throw new UnknownKeyInDriveError(`Unknown key ${JSON.stringify(editing_session_key)}`);
         // TODO: Distinguish this case from a long disconnected browser tab waking up
       }
 

tdrive/connectors/onlyoffice-connector/src/services/forgotten-processor.service.ts

@@ -1,10 +1,11 @@
+import logger from '@/lib/logger';
 import { onlyOfficeForgottenFilesCheckPeriodMS } from '@/config';
 import { PolledThingieValue } from '@/lib/polled-thingie-value';
+import { createSingleProcessorLock } from '@/lib/single-processor-lock';
 
 import apiService from './api.service';
 import onlyofficeService from './onlyoffice.service';
-import driveService from './drive.service';
-import logger from '@/lib/logger';
+import driveService, { UnknownKeyInDriveError } from './drive.service';
 
 /**
  * Periodically poll the Only Office document server for forgotten
@@ -12,6 +13,7 @@ import logger from '@/lib/logger';
  */
 class ForgottenProcessor {
   private readonly forgottenFilesPoller: PolledThingieValue<number>;
+  public readonly forgottenSynchroniser = createSingleProcessorLock<boolean>();
   private lastStart = 0;
 
   constructor() {
@@ -23,34 +25,65 @@ class ForgottenProcessor {
     );
   }
 
-  /** Get the number of seconds since the last time this process started */
-  public getLastStartTimeAgoS() {
-    return ~~((new Date().getTime() - this.lastStart) / 1000);
+  public async getHealthData() {
+    const keys = await onlyofficeService.getForgottenList();
+    return {
+      forgotten: {
+        timeSinceLastStartS: ~~((new Date().getTime() - this.lastStart) / 1000),
+        count: keys?.length ?? 0,
+        locks: this.forgottenSynchroniser.getWorstStats(),
+      },
+    };
   }
 
+  /**
+   * The point of this is to ensure this file is imported,
+   * which is needed for side-effect of starting this timer.
+   * The only other use being the health stuff it could easily
+   * be refactored out as unused.
+   */
   public makeSureItsLoaded() {
-    // The point of this is to ensure this file is imported,
-    // which is needed for side-effect of starting this timer
-    return true;
+    return 'yup this module is loaded !';
   }
 
-  private async processForgottenFiles() {
-    this.lastStart = new Date().getTime();
-    if (!(await apiService.hasToken())) return -1;
-    return await onlyofficeService.processForgotten(async (key, url) => {
+  /**
+   * Try to upload the forgotten file, optionally delete it from OO, will return if it was
+   * (or should be) deleted. Does not throw unless the OO deletion itself threw.
+   */
+  private async safeEndEditing(key: string, url: string, deleteForgotten: boolean) {
+    return this.forgottenSynchroniser.runWithLock(key, async () => {
+      let succeded = false;
       try {
         await driveService.endEditing(key, url);
-        return true;
+        succeded = true;
       } catch (error) {
-        logger.error(`Error processing forgotten file by key ${JSON.stringify(key)}: ${error}`, { key, url, error });
+        if (!(error instanceof UnknownKeyInDriveError))
+          logger.error(`Error processing forgotten file by key ${JSON.stringify(key)}: ${error}`, { key, url, error });
         // Can't do much about it here, hope it goes in retry, but don't
         // throw to keep processing
         //TODO: Maybe make a date string, compare to key, if old enough, delete...
         // this logic should probably in Twake Drive backend though
       }
-      return false;
+      if (succeded && deleteForgotten) await onlyofficeService.deleteForgotten(key);
+      return succeded;
     });
   }
+
+  private async processForgottenFiles() {
+    this.lastStart = new Date().getTime();
+    if (!(await apiService.hasToken())) return -1;
+    return await onlyofficeService.processForgotten((key, url) =>
+      this.safeEndEditing(key, url, false /* `onlyofficeService.processForgotten` will do it */),
+    );
+  }
+
+  /**
+   * Attempt to upload the forgotten OO file, only throws if deleting it failed,
+   * returns wether it was deleted.
+   */
+  public async processForgottenFile(key: string, url: string) {
+    return this.safeEndEditing(key, url, true);
+  }
 }
 
 export default new ForgottenProcessor();