diff --git a/tdrive/backend/node/src/core/platform/services/database/services/orm/repository/repository.ts b/tdrive/backend/node/src/core/platform/services/database/services/orm/repository/repository.ts index 25038583..e79df829 100644 --- a/tdrive/backend/node/src/core/platform/services/database/services/orm/repository/repository.ts +++ b/tdrive/backend/node/src/core/platform/services/database/services/orm/repository/repository.ts @@ -43,6 +43,11 @@ export type FindOptions = { sort?: SortOption; }; +export type AtomicCompareAndSetResult = { + didSet: boolean; + currentValue: FieldValueType | null; +}; + /** * Repository to work with entities. Each entity type has its own repository instance. */ @@ -129,10 +134,7 @@ export default class Repository { fieldName: keyof EntityType, previousValue: FieldValueType | null, newValue: FieldValueType | null, - ): Promise<{ - didSet: boolean; - currentValue: FieldValueType | null; - }> { + ): Promise> { if (previousValue === newValue) throw new Error(`Previous and new values are identical: ${JSON.stringify(previousValue)}`); return this.connector.atomicCompareAndSet(entity, fieldName, previousValue, newValue); diff --git a/tdrive/backend/node/src/services/applications-api/index.ts b/tdrive/backend/node/src/services/applications-api/index.ts index d817e62a..c0aee25d 100644 --- a/tdrive/backend/node/src/services/applications-api/index.ts +++ b/tdrive/backend/node/src/services/applications-api/index.ts @@ -9,6 +9,17 @@ import { logger } from "../../core/platform/framework/logger"; import { EditingSessionKeyFormat } from "../documents/entities/drive-file"; import jwt from "jsonwebtoken"; +export enum ApplicationEditingKeyStatus { + /** the key isn't known and maybe used for a new session */ + unknown = "unknown", + /** the key needed updating but is now invalid */ + updated = "updated", + /** the key was already used in a finished session and can't be used again */ + expired = "expired", + /** the key is valid and current and should be used again for the same file */ + live = "live", +} + @Prefix("/api") export default class ApplicationsApiService extends TdriveService { version = "1"; @@ -128,17 +139,22 @@ export default class ApplicationsApiService extends TdriveService { /** * Check status of `editing_session_key` in the corresponding application. * @param editingSessionKey {@see DriveFile.editing_session_key} to check - * @returns a URL string if there is a pending version to add, `null` - * if the key is unknown. + * @returns status of the provided key as far as the application knows */ - async checkPendingEditingStatus(editingSessionKey: string): Promise { + async checkPendingEditingStatus(editingSessionKey: string): Promise { const parsedKey = EditingSessionKeyFormat.parse(editingSessionKey); const response = await this.requestFromApplication( "POST", "tdriveApi/1/session/" + encodeURIComponent(editingSessionKey) + "/check", parsedKey.applicationId, ); - return (response.data.url as string) || null; + if (response.status != 200 || response.data.error) + throw new Error( + `Application check key ${editingSessionKey} failed with HTTP ${ + response.status + }: ${JSON.stringify(response.data)}`, + ); + return (response.data.status as ApplicationEditingKeyStatus) || null; } /** diff --git a/tdrive/backend/node/src/services/documents/services/index.ts b/tdrive/backend/node/src/services/documents/services/index.ts index 8c70a5eb..4b9e431c 100644 --- a/tdrive/backend/node/src/services/documents/services/index.ts +++ b/tdrive/backend/node/src/services/documents/services/index.ts @@ -6,6 +6,7 @@ import { Pagination, } from "../../../core/platform/framework/api/crud-service"; import Repository, { + AtomicCompareAndSetResult, comparisonType, inType, } from "../../../core/platform/services/database/services/orm/repository/repository"; @@ -61,7 +62,7 @@ import config from "config"; import { MultipartFile } from "@fastify/multipart"; import { UploadOptions } from "src/services/files/types"; import { SortType } from "src/core/platform/services/search/api"; -import ApplicationsApiService from "../../applications-api"; +import ApplicationsApiService, { ApplicationEditingKeyStatus } from "../../applications-api"; export class DocumentsService { version: "1"; @@ -987,18 +988,45 @@ export class DocumentsService { new CrudException("Missing or invalid application ID", 400), ); } - let newKey: string; - try { - newKey = EditingSessionKeyFormat.generate( - editorApplicationId, - appInstanceId, - context.company.id, - context.user.id, - ); - } catch (e) { - logger.error(`Error generating new editing_session_key: ${e}`, { error: e }); - CrudException.throwMe(e, new CrudException("Error generating new editing_session_key", 500)); - } + const spinLoopUntilEditable = async ( + provider: { + generateKey: () => string; + atomicSet: ( + key: string | null, + previous: string | null, + ) => Promise>; + getPluginKeyStatus: (key: string) => Promise; + }, + attemptCount = 8, + tarpitS = 1, + tarpitWorsenCoeff = 1.2, + ) => { + while (attemptCount-- > 0) { + const newKey = provider.generateKey(); + const swapResult = await provider.atomicSet(newKey, null); + logger.debug(`Begin edit try ${newKey}, got: ${JSON.stringify(swapResult)}`); + if (swapResult.didSet) return newKey; + if (!swapResult.currentValue) continue; // glitch in the matrix but ok because atomicCompareAndSet is not actually completely atomic + const existingStatus = await provider.getPluginKeyStatus(swapResult.currentValue); + logger.debug(`Begin edit get status of ${newKey}: ${JSON.stringify(existingStatus)}`); + switch (existingStatus) { + case ApplicationEditingKeyStatus.unknown: + case ApplicationEditingKeyStatus.live: + return swapResult.currentValue; + case ApplicationEditingKeyStatus.updated: + case ApplicationEditingKeyStatus.expired: + logger.debug(`Begin edit emptying previous ${swapResult.currentValue}`); + await provider.atomicSet(null, swapResult.currentValue); + break; + default: + throw new Error( + `Unexpected ApplicationEditingKeyStatus: ${JSON.stringify(existingStatus)}`, + ); + } + await new Promise(resolve => setTimeout(resolve, tarpitS * 1000)); + tarpitS *= tarpitWorsenCoeff; + } + }; const hasAccess = await checkAccess(id, null, "write", this.repository, context); if (!hasAccess) { @@ -1018,13 +1046,20 @@ export class DocumentsService { {}, context, ); - const result = await this.repository.atomicCompareAndSet( - driveFile, - "editing_session_key", - null, - newKey, - ); - return { editingSessionKey: result.currentValue }; + const editingSessionKey = await spinLoopUntilEditable({ + atomicSet: (key, previous) => + this.repository.atomicCompareAndSet(driveFile, "editing_session_key", previous, key), + generateKey: () => + EditingSessionKeyFormat.generate( + editorApplicationId, + appInstanceId, + context.company.id, + context.user.id, + ), + getPluginKeyStatus: key => + ApplicationsApiService.getDefault().checkPendingEditingStatus(key), + }); + return { editingSessionKey }; } catch (error) { logger.error({ error: `${error}` }, "Failed to begin editing Drive item"); CrudException.throwMe(error, new CrudException("Failed to begin editing Drive item", 500)); 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 82fca7ee..5475a80b 100644 --- a/tdrive/connectors/onlyoffice-connector/src/controllers/backend-callbacks.controller.ts +++ b/tdrive/connectors/onlyoffice-connector/src/controllers/backend-callbacks.controller.ts @@ -1,12 +1,21 @@ import { Request, Response } from 'express'; import logger from '@/lib/logger'; +import { createSingleProcessorLock } from '@/lib/single-processor-lock'; import onlyofficeService, { Callback, CommandError, ErrorCode } from '@/services/onlyoffice.service'; import driveService from '@/services/drive.service'; import forgottenProcessorService from '@/services/forgotten-processor.service'; +import { IHealthProvider, registerHealthProvider } from '@/services/health-providers.service'; interface RequestQuery { editing_session_key: string; } +const keyCheckLock = createSingleProcessorLock<[status: number, body: unknown]>(); + +registerHealthProvider({ + async getHealthData() { + return { checks: { locks: keyCheckLock.getWorstStats() } }; + }, +}); /** * These routes are called by Twake Drive backend, for ex. before editing or retreiving a file, @@ -26,51 +35,54 @@ export default class TwakeDriveBackendCallbackController { * - `{ error: number }`: there was an error retreiving the status of the key, http status `!= 200` */ public async checkSessionStatus(req: Request, res: Response): Promise { - try { - const forgottenURL = await onlyofficeService.getForgotten(req.params.editing_session_key); + const [status, body] = await keyCheckLock.runWithLock(req.params.editing_session_key, async () => { try { - await forgottenProcessorService.processForgottenFile(req.params.editing_session_key, forgottenURL); - } catch (error) { - logger.error(`processForgottenFile failed`, { error }); - return void res.status(502).send({ error: -57650 }); + const forgottenURL = await onlyofficeService.getForgotten(req.params.editing_session_key); + try { + await forgottenProcessorService.processForgottenFile(req.params.editing_session_key, forgottenURL); + } catch (error) { + logger.error(`processForgottenFile failed`, { error }); + return [502, { error: -57650 }]; + } + return [200, { status: 'updated' }]; + } catch (e) { + if (!(e instanceof CommandError && e.errorCode == ErrorCode.KEY_MISSING_OR_DOC_NOT_FOUND)) { + logger.error(`getForgotten failed`, { error: e }); + return [e instanceof CommandError ? 502 : 500, { error: -57651 }]; + } } - return void res.send({ status: 'updated' }); - } catch (e) { - if (!(e instanceof CommandError && e.errorCode == ErrorCode.KEY_MISSING_OR_DOC_NOT_FOUND)) { - logger.error(`getForgotten failed`, { error: e }); - return void res.status(e instanceof CommandError ? 502 : 500).send({ error: -57651 }); + const info = await onlyofficeService.getInfoAndWaitForCallbackUnsafe(req.params.editing_session_key); + if (info.error === ErrorCode.KEY_MISSING_OR_DOC_NOT_FOUND) { + // just start using it + return [200, { status: 'unknown' }]; } - } - const info = await onlyofficeService.getInfoAndWaitForCallbackUnsafe(req.params.editing_session_key); - if (info.error === ErrorCode.KEY_MISSING_OR_DOC_NOT_FOUND) { - // just start using it - return void res.send({ status: 'unknown' }); - } - if (info.error !== undefined) { - logger.error(`getInfo failed`, { error: info }); - return void res.status(502).send({ error: -57652 }); - } - switch (info.result.status) { - case Callback.Status.BEING_EDITED: - case Callback.Status.BEING_EDITED_BUT_IS_SAVED: - // use it as is - return void res.send({ status: 'live' }); + if (info.error !== undefined) { + logger.error(`getInfo failed`, { error: info }); + return [502, { error: -57652 }]; + } + switch (info.result.status) { + case Callback.Status.BEING_EDITED: + case Callback.Status.BEING_EDITED_BUT_IS_SAVED: + // use it as is + return [200, { status: 'live' }]; - case Callback.Status.CLOSED_WITHOUT_CHANGES: - // just cancel it - return void res.send({ status: 'expired' }); + case Callback.Status.CLOSED_WITHOUT_CHANGES: + // just cancel it + return [200, { status: 'expired' }]; - case Callback.Status.ERROR_FORCE_SAVING: - case Callback.Status.ERROR_SAVING: - return void res.status(502).send({ error: info.result.status }); + case Callback.Status.ERROR_FORCE_SAVING: + case Callback.Status.ERROR_SAVING: + return [502, { error: info.result.status }]; - case Callback.Status.READY_FOR_SAVING: - // upload it, have to do it here for correct user stored in url in OO - await driveService.endEditing(req.params.editing_session_key, info.result.url); - return void res.send({ status: 'updated' }); + case Callback.Status.READY_FOR_SAVING: + // upload it, have to do it here for correct user stored in url in OO + await driveService.endEditing(req.params.editing_session_key, info.result.url); + return [200, { status: 'updated' }]; - default: - throw new Error(`Unexpected callback status: ${JSON.stringify(info.result)}`); - } + default: + throw new Error(`Unexpected callback status: ${JSON.stringify(info.result)}`); + } + }); + await res.status(status).send(body); } } diff --git a/tdrive/connectors/onlyoffice-connector/src/controllers/onlyoffice.controller.ts b/tdrive/connectors/onlyoffice-connector/src/controllers/onlyoffice.controller.ts index 112e25be..f472d34a 100644 --- a/tdrive/connectors/onlyoffice-connector/src/controllers/onlyoffice.controller.ts +++ b/tdrive/connectors/onlyoffice-connector/src/controllers/onlyoffice.controller.ts @@ -73,13 +73,6 @@ class OnlyOfficeController { const officeTokenPayload = jwt.verify(token, CREDENTIALS_SECRET) as OfficeToken; const { preview, /* company_id, file_id, user_id, drive_file_id, */ in_page_token /* editing_session_key */ } = officeTokenPayload; - // Ignore errors generated by pending request - // try-catch not needed because it is async - // there may be later reasons to wait for callbacks - // to process and eventually respond accordingly to - // OO an error for certain statuses - void OnlyOffice.default.ooCallbackCalled(req.body); // has to be single thread per key - // check token is an in_page_token and allow save if (!in_page_token) throw new Error('OO Callback invalid token, must be a in_page_token'); if (preview) throw new Error('OO Callback invalid token, must not be a preview token for save operation'); @@ -125,6 +118,14 @@ class OnlyOfficeController { `OO Callback unexpected status field: ${OnlyOffice.Callback.StatusToString(req.body.status)} in ${JSON.stringify(req.body)}`, ); } + + // Ignore errors generated by pending request + // try-catch not needed because it is async + // there may be later reasons to wait for callbacks + // to process and eventually respond accordingly to + // OO an error for certain statuses + void OnlyOffice.default.ooCallbackCalled(req.body); // has to be single thread per key + return respondToOO(0); } catch (error) { logger.error(`OO Callback root error`, { error });