fix: address livechat race condition in agent assignment with locking

.changeset/cozy-melons-march.md

@@ -0,0 +1,8 @@
+---
+'@rocket.chat/model-typings': patch
+'@rocket.chat/core-typings': patch
+'@rocket.chat/models': patch
+'@rocket.chat/meteor': patch
+---
+
+Prevents over-assignment of omnichannel agents beyond their max chats limit in microservices deployments by serializing agent assignment with explicit user-level locking.

apps/meteor/app/livechat/server/lib/RoutingManager.ts

@@ -28,6 +28,7 @@ import {
 	updateChatDepartment,
 	allowAgentSkipQueue,
 } from './Helper';
+import { conditionalLockAgent } from './conditionalLockAgent';
 import { afterTakeInquiry, beforeDelegateAgent } from './hooks';
 import { callbacks } from '../../../../server/lib/callbacks';
 import { notifyOnLivechatInquiryChangedById, notifyOnLivechatInquiryChanged } from '../../../lib/server/lib/notifyListener';
@@ -257,19 +258,35 @@ export const RoutingManager: Routing = {
 			return room;
 		}
 
-		try {
-			await callbacks.run('livechat.checkAgentBeforeTakeInquiry', {
-				agent,
-				inquiry,
-				options,
+		const lock = await conditionalLockAgent(agent.agentId);
+		if (!lock.acquired && lock.required) {
+			logger.debug({
+				msg: 'Cannot take inquiry because agent is currently locked by another process',
+				agentId: agent.agentId,
+				inquiryId: _id,
 			});
-		} catch (e) {
 			if (options.clientAction && !options.forwardingToDepartment) {
-				throw e;
+				throw new Error('error-agent-is-locked');
 			}
 			agent = null;
 		}
 
+		if (agent) {
+			try {
+				await callbacks.run('livechat.checkAgentBeforeTakeInquiry', {
+					agent,
+					inquiry,
+					options,
+				});
+			} catch (e) {
+				await lock.unlock();
+				if (options.clientAction && !options.forwardingToDepartment) {
+					throw e;
+				}
+				agent = null;
+			}
+		}
+
 		if (!agent) {
 			logger.debug({ msg: 'Cannot take inquiry. Precondition failed for agent', inquiryId: inquiry._id });
 			const cbRoom = await callbacks.run<'livechat.onAgentAssignmentFailed'>('livechat.onAgentAssignmentFailed', room, {
@@ -279,35 +296,39 @@ export const RoutingManager: Routing = {
 			return cbRoom;
 		}
 
-		const result = await LivechatInquiry.takeInquiry(_id, inquiry.lockedAt);
-		if (result.modifiedCount === 0) {
-			logger.error({ msg: 'Failed to take inquiry, could not match lockedAt', inquiryId: _id, lockedAt: inquiry.lockedAt });
-			throw new Error('error-taking-inquiry-lockedAt-mismatch');
-		}
+		try {
+			const result = await LivechatInquiry.takeInquiry(_id, inquiry.lockedAt);
+			if (result.modifiedCount === 0) {
+				logger.error({ msg: 'Failed to take inquiry because lockedAt did not match', inquiryId: _id, lockedAt: inquiry.lockedAt });
+				throw new Error('error-taking-inquiry-lockedAt-mismatch');
+			}
 
-		logger.info({ msg: 'Inquiry taken by agent', inquiryId: inquiry._id, agentId: agent.agentId });
+			logger.info({ msg: 'Inquiry taken', inquiryId: _id, agentId: agent.agentId });
 
-		// assignAgent changes the room data to add the agent serving the conversation. afterTakeInquiry expects room object to be updated
-		const { inquiry: returnedInquiry, user } = await this.assignAgent(inquiry as InquiryWithAgentInfo, agent);
-		const roomAfterUpdate = await LivechatRooms.findOneById(rid);
+			// assignAgent changes the room data to add the agent serving the conversation. afterTakeInquiry expects room object to be updated
+			const { inquiry: returnedInquiry, user } = await this.assignAgent(inquiry, agent);
+			const roomAfterUpdate = await LivechatRooms.findOneById(rid);
 
-		if (!roomAfterUpdate) {
-			// This should never happen
-			throw new Error('error-room-not-found');
-		}
+			if (!roomAfterUpdate) {
+				// This should never happen
+				throw new Error('error-room-not-found');
+			}
 
-		void Apps.self?.triggerEvent(AppEvents.IPostLivechatAgentAssigned, { room: roomAfterUpdate, user });
-		void afterTakeInquiry({ inquiry: returnedInquiry, room: roomAfterUpdate, agent });
+			void Apps.self?.triggerEvent(AppEvents.IPostLivechatAgentAssigned, { room: roomAfterUpdate, user });
+			void afterTakeInquiry({ inquiry: returnedInquiry, room: roomAfterUpdate, agent });
 
-		void notifyOnLivechatInquiryChangedById(inquiry._id, 'updated', {
-			status: LivechatInquiryStatus.TAKEN,
-			takenAt: new Date(),
-			defaultAgent: undefined,
-			estimatedInactivityCloseTimeAt: undefined,
-			queuedAt: undefined,
-		});
+			void notifyOnLivechatInquiryChangedById(inquiry._id, 'updated', {
+				status: LivechatInquiryStatus.TAKEN,
+				takenAt: new Date(),
+				defaultAgent: undefined,
+				estimatedInactivityCloseTimeAt: undefined,
+				queuedAt: undefined,
+			});
 
-		return roomAfterUpdate;
+			return roomAfterUpdate;
+		} finally {
+			await lock.unlock();
+		}
 	},
 
 	async transferRoom(room, guest, transferData) {

apps/meteor/app/livechat/server/lib/conditionalLockAgent.ts

@@ -0,0 +1,35 @@
+import { Users } from '@rocket.chat/models';
+
+import { settings } from '../../../settings/server';
+
+type LockResult = {
+	acquired: boolean;
+	required: boolean;
+	unlock: () => Promise<void>;
+};
+
+export async function conditionalLockAgent(agentId: string): Promise<LockResult> {
+	// Lock and chats limits enforcement are only required when waiting_queue is enabled
+	const shouldLock = settings.get<boolean>('Livechat_waiting_queue');
+
+	if (!shouldLock) {
+		return {
+			acquired: false,
+			required: false,
+			unlock: async () => {
+				// no-op
+			},
+		};
+	}
+
+	const lockTime = new Date();
+	const lockAcquired = await Users.acquireAgentLock(agentId, lockTime);
+
+	return {
+		acquired: !!lockAcquired,
+		required: true,
+		unlock: async () => {
+			await Users.releaseAgentLock(agentId, lockTime);
+		},
+	};
+}

apps/meteor/tests/unit/app/livechat/server/lib/conditionalLockAgent.spec.ts

@@ -0,0 +1,84 @@
+import { expect } from 'chai';
+import proxyquire from 'proxyquire';
+import sinon from 'sinon';
+
+const mockUsers = {
+	acquireAgentLock: sinon.stub(),
+	releaseAgentLock: sinon.stub(),
+};
+
+const mockSettings = {
+	get: sinon.stub(),
+};
+
+const { conditionalLockAgent } = proxyquire.noCallThru().load('../../../../../../app/livechat/server/lib/conditionalLockAgent', {
+	'@rocket.chat/models': { Users: mockUsers },
+	'../../../settings/server': { settings: mockSettings },
+});
+
+describe('conditionalLockAgent', () => {
+	beforeEach(() => {
+		mockUsers.acquireAgentLock.reset();
+		mockUsers.releaseAgentLock.reset();
+		mockSettings.get.reset();
+	});
+
+	describe('when waiting_queue is enabled', () => {
+		beforeEach(() => {
+			mockSettings.get.withArgs('Livechat_waiting_queue').returns(true);
+		});
+
+		it('should return acquired: true when lock is successfully acquired', async () => {
+			mockUsers.acquireAgentLock.resolves(true);
+
+			const result = await conditionalLockAgent('agent1');
+
+			expect(result.acquired).to.equal(true);
+			expect(result.required).to.equal(true);
+			expect(mockUsers.acquireAgentLock.calledOnce).to.equal(true);
+		});
+
+		it('should return acquired: false when lock is already held by another process', async () => {
+			mockUsers.acquireAgentLock.resolves(false);
+
+			const result = await conditionalLockAgent('agent1');
+
+			expect(result.acquired).to.equal(false);
+			expect(result.required).to.equal(true);
+			expect(mockUsers.acquireAgentLock.calledOnce).to.equal(true);
+		});
+
+		it('should call releaseAgentLock when unlock is called', async () => {
+			mockUsers.acquireAgentLock.resolves(true);
+			mockUsers.releaseAgentLock.resolves(true);
+
+			const result = await conditionalLockAgent('agent1');
+			await result.unlock();
+
+			expect(mockUsers.releaseAgentLock.calledOnce).to.equal(true);
+			expect(mockUsers.releaseAgentLock.firstCall.args[0]).to.equal('agent1');
+			expect(mockUsers.releaseAgentLock.firstCall.args[1]).to.be.instanceOf(Date);
+		});
+	});
+
+	describe('when waiting_queue is disabled', () => {
+		beforeEach(() => {
+			mockSettings.get.withArgs('Livechat_waiting_queue').returns(false);
+		});
+
+		it('should return acquired: false and required: false without calling acquireAgentLock', async () => {
+			const result = await conditionalLockAgent('agent1');
+
+			expect(result.acquired).to.equal(false);
+			expect(result.required).to.equal(false);
+			expect(mockUsers.acquireAgentLock.called).to.equal(false);
+		});
+
+		it('should have a no-op unlock function', async () => {
+			const result = await conditionalLockAgent('agent1');
+			await result.unlock(); // should not throw an error
+
+			expect(mockUsers.releaseAgentLock.called).to.equal(false);
+		});
+	});
+});

packages/core-typings/src/ILivechatAgent.ts

@@ -14,6 +14,8 @@ export interface ILivechatAgent extends IUser {
 	lastRoutingTime: Date;
 	livechatStatusSystemModified?: boolean;
 	openBusinessHours?: string[];
+	agentLockedAt?: Date;
+	agentLocked?: boolean;
 }
 
 export type AvailableAgentsAggregation = {

packages/model-typings/src/models/IUsersModel.ts

@@ -138,6 +138,9 @@ export interface IUsersModel extends IBaseModel<IUser> {
 		queueInfo: { chats: number; chatsForDepartment?: number };
 	}>;
 
+	acquireAgentLock(agentId: IUser['_id'], lockTime: Date, lockTimeoutMs?: number): Promise<boolean>;
+	releaseAgentLock(agentId: IUser['_id'], lockTime: Date): Promise<boolean>;
+
 	findAllResumeTokensByUserId(userId: IUser['_id']): Promise<{ tokens: IMeteorLoginToken[] }[]>;
 
 	findActiveByUsernameOrNameRegexWithExceptionsAndConditions<T extends Document = IUser>(

packages/models/src/models/Users.ts

@@ -58,6 +58,15 @@ const queryStatusAgentOnline = (extraFilters = {}, isLivechatEnabledWhenAgentIdl
 	}),
 });
 
+const queryAvailableAgentsForSelection = (extraFilters = {}, isLivechatEnabledWhenAgentIdle?: boolean): Filter<IUser> => ({
+	...queryStatusAgentOnline(extraFilters, isLivechatEnabledWhenAgentIdle),
+	$and: [
+		{
+			$or: [{ agentLocked: { $exists: false } }, { agentLockedAt: { $lt: new Date(Date.now() - 5000) } }],
+		},
+	],
+});
+
 export class UsersRaw extends BaseRaw<IUser, DefaultFields<IUser>> implements IUsersModel {
 	constructor(db: Db, trash?: Collection<RocketChatRecordDeleted<IUser>>) {
 		super(db, 'users', trash, {
@@ -586,7 +595,7 @@ export class UsersRaw extends BaseRaw<IUser, DefaultFields<IUser>> implements IU
 		isEnabledWhenAgentIdle?: boolean,
 		ignoreUsernames?: string[],
 	): Promise<{ agentId: string; username?: string; lastRoutingTime?: Date; count: number; departments?: any[] }> {
-		const match = queryStatusAgentOnline(
+		const match = queryAvailableAgentsForSelection(
 			{ ...(ignoreAgentId && { _id: { $ne: ignoreAgentId } }), ...(ignoreUsernames?.length && { username: { $nin: ignoreUsernames } }) },
 			isEnabledWhenAgentIdle,
 		);
@@ -667,7 +676,7 @@ export class UsersRaw extends BaseRaw<IUser, DefaultFields<IUser>> implements IU
 		isEnabledWhenAgentIdle?: boolean,
 		ignoreUsernames?: string[],
 	): Promise<{ agentId: string; username?: string; lastRoutingTime?: Date; departments?: any[] }> {
-		const match = queryStatusAgentOnline(
+		const match = queryAvailableAgentsForSelection(
 			{ ...(ignoreAgentId && { _id: { $ne: ignoreAgentId } }), ...(ignoreUsernames?.length && { username: { $nin: ignoreUsernames } }) },
 			isEnabledWhenAgentIdle,
 		);
@@ -827,6 +836,41 @@ export class UsersRaw extends BaseRaw<IUser, DefaultFields<IUser>> implements IU
 		return agent;
 	}
 
+	async acquireAgentLock(agentId: IUser['_id'], lockTime: Date, lockTimeoutMs = 5000): Promise<boolean> {
+		const result = await this.updateOne(
+			{
+				_id: agentId,
+				$or: [{ agentLocked: { $exists: false } }, { agentLockedAt: { $lt: new Date(Date.now() - lockTimeoutMs) } }],
+			},
+			{
+				$set: {
+					agentLocked: true,
+					agentLockedAt: lockTime,
+				},
+			},
+		);
+
+		return result.modifiedCount > 0;
+	}
+
+	async releaseAgentLock(agentId: IUser['_id'], lockTime: Date): Promise<boolean> {
+		const result = await this.updateOne(
+			{
+				_id: agentId,
+				agentLocked: true,
+				agentLockedAt: lockTime,
+			},
+			{
+				$unset: {
+					agentLocked: 1,
+					agentLockedAt: 1,
+				},
+			},
+		);
+
+		return result.modifiedCount > 0;
+	}
+
 	findAllResumeTokensByUserId(userId: IUser['_id']): Promise<{ tokens: IMeteorLoginToken[] }[]> {
 		return this.col
 			.aggregate<{ tokens: IMeteorLoginToken[] }>([
@@ -1926,7 +1970,7 @@ export class UsersRaw extends BaseRaw<IUser, DefaultFields<IUser>> implements IU
 			username: { $nin: unavailableAgents },
 		};
 
-		const query = queryStatusAgentOnline(extraFilters, enabledWhenAgentIdle);
+		const query = queryAvailableAgentsForSelection(extraFilters, enabledWhenAgentIdle);
 
 		const sort: Record<string, SortDirection> = {
 			livechatCount: 1,