feat(server): make a singleton global mutex service (#7900)

Author: forehalo

packages/backend/server/src/core/selfhost/controller.ts

@@ -5,7 +5,7 @@ import {
   ActionForbidden,
   EventEmitter,
   InternalServerError,
-  MutexService,
+  Mutex,
   PasswordRequired,
 } from '../../fundamentals';
 import { AuthService, Public } from '../auth';
@@ -23,7 +23,7 @@ export class CustomSetupController {
     private readonly user: UserService,
     private readonly auth: AuthService,
     private readonly event: EventEmitter,
-    private readonly mutex: MutexService,
+    private readonly mutex: Mutex,
     private readonly server: ServerService
   ) {}
 

packages/backend/server/src/core/workspaces/resolvers/workspace.ts

@@ -20,7 +20,7 @@ import {
   InternalServerError,
   MailService,
   MemberQuotaExceeded,
-  MutexService,
+  RequestMutex,
   Throttle,
   TooManyRequest,
   UserNotFound,
@@ -57,7 +57,7 @@ export class WorkspaceResolver {
     private readonly users: UserService,
     private readonly event: EventEmitter,
     private readonly blobStorage: WorkspaceBlobStorage,
-    private readonly mutex: MutexService
+    private readonly mutex: RequestMutex
   ) {}
 
   @ResolveField(() => Permission, {

packages/backend/server/src/fundamentals/index.ts

@@ -19,7 +19,7 @@ export type { GraphqlContext } from './graphql';
 export { CryptoHelper, URLHelper } from './helpers';
 export { MailService } from './mailer';
 export { CallCounter, CallTimer, metrics } from './metrics';
-export { type ILocker, Lock, Locker, MutexService } from './mutex';
+export { type ILocker, Lock, Locker, Mutex, RequestMutex } from './mutex';
 export {
   GatewayErrorWrapper,
   getOptionalModuleMetadata,

packages/backend/server/src/fundamentals/mutex/index.ts

@@ -1,14 +1,14 @@
 import { Global, Module } from '@nestjs/common';
 
 import { Locker } from './local-lock';
-import { MutexService } from './mutex';
+import { Mutex, RequestMutex } from './mutex';
 
 @Global()
 @Module({
-  providers: [MutexService, Locker],
-  exports: [MutexService],
+  providers: [Mutex, RequestMutex, Locker],
+  exports: [Mutex, RequestMutex],
 })
 export class MutexModule {}
 
-export { Locker, MutexService };
+export { Locker, Mutex, RequestMutex };
 export { type Locker as ILocker, Lock } from './lock';

packages/backend/server/src/fundamentals/mutex/mutex.ts

@@ -11,36 +11,11 @@ import { Locker } from './local-lock';
 export const MUTEX_RETRY = 5;
 export const MUTEX_WAIT = 100;
 
-@Injectable({ scope: Scope.REQUEST })
-export class MutexService {
-  protected logger = new Logger(MutexService.name);
-  private readonly locker: Locker;
-
-  constructor(
-    @Inject(REQUEST) private readonly request: Request | GraphqlContext,
-    private readonly ref: ModuleRef
-  ) {
-    // nestjs will always find and injecting the locker from local module
-    // so the RedisLocker implemented by the plugin mechanism will not be able to overwrite the internal locker
-    // we need to use find and get the locker from the `ModuleRef` manually
-    //
-    // NOTE: when a `constructor` execute in normal service, the Locker module we expect may not have been initialized
-    //       but in the Service with `Scope.REQUEST`, we will create a separate Service instance for each request
-    //       at this time, all modules have been initialized, so we able to get the correct Locker instance in `constructor`
-    this.locker = this.ref.get(Locker, { strict: false });
-  }
+@Injectable()
+export class Mutex {
+  protected logger = new Logger(Mutex.name);
 
-  protected getId() {
-    const req = 'req' in this.request ? this.request.req : this.request;
-    let id = req.headers['x-transaction-id'] as string;
-
-    if (!id) {
-      id = randomUUID();
-      req.headers['x-transaction-id'] = id;
-    }
-
-    return id;
-  }
+  constructor(protected readonly locker: Locker) {}
 
   /**
    * lock an resource and return a lock guard, which will release the lock when disposed
@@ -63,10 +38,10 @@ export class MutexService {
    * @param key resource key
    * @returns LockGuard
    */
-  async lock(key: string) {
+  async lock(key: string, owner: string = 'global') {
     try {
       return await retryable(
-        () => this.locker.lock(this.getId(), key),
+        () => this.locker.lock(owner, key),
         MUTEX_RETRY,
         MUTEX_WAIT
       );
@@ -79,3 +54,36 @@ export class MutexService {
     }
   }
 }
+
+@Injectable({ scope: Scope.REQUEST })
+export class RequestMutex extends Mutex {
+  constructor(
+    @Inject(REQUEST) private readonly request: Request | GraphqlContext,
+    ref: ModuleRef
+  ) {
+    // nestjs will always find and injecting the locker from local module
+    // so the RedisLocker implemented by the plugin mechanism will not be able to overwrite the internal locker
+    // we need to use find and get the locker from the `ModuleRef` manually
+    //
+    // NOTE: when a `constructor` execute in normal service, the Locker module we expect may not have been initialized
+    //       but in the Service with `Scope.REQUEST`, we will create a separate Service instance for each request
+    //       at this time, all modules have been initialized, so we able to get the correct Locker instance in `constructor`
+    super(ref.get(Locker));
+  }
+
+  protected getId() {
+    const req = 'req' in this.request ? this.request.req : this.request;
+    let id = req.headers['x-transaction-id'] as string;
+
+    if (!id) {
+      id = randomUUID();
+      req.headers['x-transaction-id'] = id;
+    }
+
+    return id;
+  }
+
+  override lock(key: string) {
+    return super.lock(key, this.getId());
+  }
+}

packages/backend/server/src/plugins/copilot/resolver.ts

@@ -26,7 +26,7 @@ import { UserType } from '../../core/user';
 import {
   CopilotFailedToCreateMessage,
   FileUpload,
-  MutexService,
+  RequestMutex,
   Throttle,
   TooManyRequest,
 } from '../../fundamentals';
@@ -265,7 +265,7 @@ export class CopilotType {
 export class CopilotResolver {
   constructor(
     private readonly permissions: PermissionService,
-    private readonly mutex: MutexService,
+    private readonly mutex: RequestMutex,
     private readonly chatSession: ChatSessionService,
     private readonly storage: CopilotStorage
   ) {}