From 6685b7b98d34a7e12465938bbcf0f5ce4258aea5 Mon Sep 17 00:00:00 2001 From: Eric Doughty-Papassideris Date: Wed, 18 Sep 2024 03:34:18 +0200 Subject: [PATCH] =?UTF-8?q?=E2=99=BB=EF=B8=8F=20oo-connector:=20centralisi?= =?UTF-8?q?ng=20forgotten=20file=20batch=20management=20with=20session=20k?= =?UTF-8?q?ey=20check,=20bit=20of=20health=20refactoring=20(#525)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- .../backend-callbacks.controller.ts | 11 +- .../controllers/health-status.controller.ts | 12 +- .../src/lib/single-processor-lock.ts | 161 ++++++++++++++++++ .../src/services/drive.service.ts | 7 +- .../services/forgotten-processor.service.ts | 63 +++++-- 5 files changed, 220 insertions(+), 34 deletions(-) create mode 100644 tdrive/connectors/onlyoffice-connector/src/lib/single-processor-lock.ts diff --git a/tdrive/connectors/onlyoffice-connector/src/controllers/backend-callbacks.controller.ts b/tdrive/connectors/onlyoffice-connector/src/controllers/backend-callbacks.controller.ts index 6a31afef..82fca7ee 100644 --- a/tdrive/connectors/onlyoffice-connector/src/controllers/backend-callbacks.controller.ts +++ b/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' }); diff --git a/tdrive/connectors/onlyoffice-connector/src/controllers/health-status.controller.ts b/tdrive/connectors/onlyoffice-connector/src/controllers/health-status.controller.ts index a5eee4da..97fa2826 100644 --- a/tdrive/connectors/onlyoffice-connector/src/controllers/health-status.controller.ts +++ b/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 { - 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), ); diff --git a/tdrive/connectors/onlyoffice-connector/src/lib/single-processor-lock.ts b/tdrive/connectors/onlyoffice-connector/src/lib/single-processor-lock.ts new file mode 100644 index 00000000..108f3d6a --- /dev/null +++ b/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 { + public readonly startAtMs = new Date().getTime(); + public readonly promise: Promise; + 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['_resolve']>): void { + this.check(); + this._resolve(...args); + this.finish(); + } + public reject(...args: Parameters['_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 { + /** To distinguish return from {@link LockWaiter} */ + didAcquire: true; + /** This is the same Promise that `resolve` and `reject` will settle, and waiters get */ + promise: LockWaiter['promise']; + /** Release the lock and resolve the promise of all the waiters */ + resolve: ExplodedPromise['resolve']; + /** Release the lock and reject the promise of all the waiters */ + reject: ExplodedPromise['reject']; +} + +/** Use this instance to wait if {@link SingleProcessorLock.acquire} was already processing that key */ +export interface LockWaiter { + /** 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; +} + +/** + * 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() { + const promisesByKey = new Map>(); + + /** + * 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 | LockWaiter => { + 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(); + promisesByKey.set(key, processorPromise); + debugLocks && logger.debug(`LOCK started: ${key}`); + return { + didAcquire: true, + promise: processorPromise.promise, + resolve(...args: Parameters['resolve']>) { + debugLocks && logger.debug(`LOCK resolved: ${key}`); + promisesByKey.delete(key); + processorPromise.resolve(...args); + }, + reject(...args: Parameters['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): Promise { + 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, + }, + ); + }, + }; +} diff --git a/tdrive/connectors/onlyoffice-connector/src/services/drive.service.ts b/tdrive/connectors/onlyoffice-connector/src/services/drive.service.ts index 92b3dbe1..2f8ed3ed 100644 --- a/tdrive/connectors/onlyoffice-connector/src/services/drive.service.ts +++ b/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 } diff --git a/tdrive/connectors/onlyoffice-connector/src/services/forgotten-processor.service.ts b/tdrive/connectors/onlyoffice-connector/src/services/forgotten-processor.service.ts index 3be7713d..c295af3a 100644 --- a/tdrive/connectors/onlyoffice-connector/src/services/forgotten-processor.service.ts +++ b/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; + public readonly forgottenSynchroniser = createSingleProcessorLock(); 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();