throttle down max number of processing requests in batch

avoid loading node by checking how many requests
are already currently in progress

apps/server/src/modules/deletion/api/uc/deletion-request.uc.spec.ts

@@ -160,15 +160,23 @@ describe(DeletionRequestUc.name, () => {
 				const deletionRequest = deletionRequestFactory.buildWithId({ deleteAfter: new Date('2023-01-01') });
 
 				configService.get.mockImplementation((key) => deletionTestConfig()[key]);
+				deletionRequestService.findInProgressCount.mockResolvedValueOnce(1);
 
 				return { deletionRequest };
 			};
+			it('should call deletionRequestService.findInProgressCount', async () => {
+				setup();
+
+				await uc.executeDeletionRequests();
+
+				expect(deletionRequestService.findInProgressCount).toHaveBeenCalled();
+			});
 			it('should call deletionRequestService.findAllItemsToExecute', async () => {
 				setup();
 
 				await uc.executeDeletionRequests();
 
-				expect(deletionRequestService.findAllItemsToExecute).toHaveBeenCalled();
+				expect(deletionRequestService.findAllItemsToExecute).toHaveBeenCalledWith(1, undefined);
 			});
 
 			it('should call deletionRequestService.markDeletionRequestAsPending to update status of deletionRequests', async () => {
@@ -203,6 +211,7 @@ describe(DeletionRequestUc.name, () => {
 				});
 
 				deletionRequestService.findAllItemsToExecute.mockResolvedValueOnce([deletionRequest]);
+				deletionRequestService.findInProgressCount.mockResolvedValueOnce(0);
 
 				await uc.executeDeletionRequests();
 
@@ -215,6 +224,7 @@ describe(DeletionRequestUc.name, () => {
 				const deletionRequestToExecute = deletionRequestFactory.build({ deleteAfter: new Date('2023-01-01') });
 
 				deletionRequestService.findAllItemsToExecute.mockResolvedValueOnce([deletionRequestToExecute]);
+				deletionRequestService.findInProgressCount.mockResolvedValueOnce(1);
 				eventBus.publish.mockRejectedValueOnce(new Error());
 
 				return {

apps/server/src/modules/deletion/api/uc/deletion-request.uc.ts

@@ -81,14 +81,18 @@ export class DeletionRequestUc implements IEventHandler<DataDeletedEvent> {
 	async executeDeletionRequests(limit?: number, getFailed?: boolean): Promise<void> {
 		this.logger.debug({ action: 'executeDeletionRequests', limit });
 
-		limit = limit ?? this.configService.get<number>('ADMIN_API__DELETION_MAX_CONCURRENT_DELETION_REQUESTS');
+		const inProgress = await this.deletionRequestService.findInProgressCount();
+
+		const max = limit
+			? limit - inProgress
+			: this.configService.get<number>('ADMIN_API__DELETION_MAX_CONCURRENT_DELETION_REQUESTS') - inProgress;
 
 		let deletionRequests: DeletionRequest[] = [];
 
-		if (limit > 0) {
+		if (max > 0) {
 			do {
 				// eslint-disable-next-line no-await-in-loop
-				deletionRequests = await this.deletionRequestService.findAllItemsToExecute(limit, getFailed);
+				deletionRequests = await this.deletionRequestService.findAllItemsToExecute(max, getFailed);
 
 				this.logger.debug({ action: 'processing deletion request', deletionRequests });
 

apps/server/src/modules/deletion/domain/service/deletion-request.service.spec.ts

@@ -152,6 +152,36 @@ describe(DeletionRequestService.name, () => {
 		});
 	});
 
+	describe('findInProgressCount', () => {
+		describe('when finding in progress deletionRequests', () => {
+			const setup = () => {
+				const thresholdNewerMs = configService.get<number>('ADMIN_API__DELETION_CONSIDER_FAILED_AFTER_MS') ?? 1000;
+				const newerThan = new Date(Date.now() - thresholdNewerMs);
+
+				const count = 2;
+
+				deletionRequestRepo.findInProgressCount.mockResolvedValue(count);
+
+				return { count, newerThan };
+			};
+
+			it('should call deletionRequestRepo.findInProgressCount', async () => {
+				const { newerThan } = setup();
+
+				await service.findInProgressCount();
+
+				expect(deletionRequestRepo.findInProgressCount).toBeCalledWith(newerThan);
+			});
+
+			it('should return count of in progress deletionRequests', async () => {
+				const { count } = setup();
+				const result = await service.findInProgressCount();
+
+				expect(result).toEqual(count);
+			});
+		});
+	});
+
 	describe('update', () => {
 		describe('when updating deletionRequest', () => {
 			const setup = () => {

apps/server/src/modules/deletion/domain/service/deletion-request.service.ts

@@ -48,14 +48,19 @@ export class DeletionRequestService {
 		return deletionRequest;
 	}
 
-	async findAllItemsToExecute(limit: number, getFailed = false): Promise<DeletionRequest[]> {
+	public async findAllItemsToExecute(limit: number, getFailed = false): Promise<DeletionRequest[]> {
 		const deletionRequests = getFailed
 			? await this.deletionRequestRepo.findAllFailedItems(limit, this.olderThan, this.newerThan)
 			: await this.deletionRequestRepo.findAllItems(limit);
 
 		return deletionRequests;
 	}
 
+	public async findInProgressCount(): Promise<number> {
+		const count = await this.deletionRequestRepo.findInProgressCount(this.newerThan);
+		return count;
+	}
+
 	async update(deletionRequestToUpdate: DeletionRequest): Promise<void> {
 		await this.deletionRequestRepo.update(deletionRequestToUpdate);
 	}

apps/server/src/modules/deletion/repo/deletion-request.repo.spec.ts

@@ -286,6 +286,55 @@ describe(DeletionRequestRepo.name, () => {
 		});
 	});
 
+	describe('findInProgressCount', () => {
+		const setup = async () => {
+			const newerThan = new Date(2023, 8, 1);
+
+			const deletionRequestEntity1: DeletionRequestEntity = deletionRequestEntityFactory.build({
+				createdAt: new Date(2023, 7, 1),
+				updatedAt: new Date(2023, 8, 2),
+				deleteAfter: new Date(2023, 8, 1),
+				status: StatusModel.PENDING,
+			});
+			const deletionRequestEntity2: DeletionRequestEntity = deletionRequestEntityFactory.build({
+				createdAt: new Date(2023, 7, 1),
+				updatedAt: new Date(2023, 8, 2),
+				deleteAfter: new Date(2023, 8, 1),
+				status: StatusModel.FAILED,
+			});
+			const deletionRequestEntity3: DeletionRequestEntity = deletionRequestEntityFactory.build({
+				createdAt: new Date(2023, 8, 1),
+				updatedAt: new Date(2023, 7, 1),
+				deleteAfter: new Date(2023, 9, 1),
+				status: StatusModel.FAILED,
+			});
+			const deletionRequestEntity4: DeletionRequestEntity = deletionRequestEntityFactory.build({
+				createdAt: new Date(2023, 8, 1),
+				updatedAt: new Date(2023, 8, 1),
+				deleteAfter: new Date(2023, 9, 1),
+				status: StatusModel.REGISTERED,
+			});
+
+			await em.persistAndFlush([
+				deletionRequestEntity1,
+				deletionRequestEntity2,
+				deletionRequestEntity3,
+				deletionRequestEntity4,
+			]);
+			em.clear();
+
+			return { newerThan };
+		};
+
+		it('should return the count of deletionRequests with status pending or failed', async () => {
+			const { newerThan } = await setup();
+
+			const result = await repo.findInProgressCount(newerThan);
+
+			expect(result).toBe(2);
+		});
+	});
+
 	describe('update', () => {
 		describe('when updating deletionRequest', () => {
 			const setup = async () => {

apps/server/src/modules/deletion/repo/deletion-request.repo.ts

@@ -69,6 +69,15 @@ export class DeletionRequestRepo {
 		return mapped;
 	}
 
+	public async findInProgressCount(newerThan: Date): Promise<number> {
+		const scope = new DeletionRequestScope();
+		scope.byStatusAndDate([StatusModel.PENDING, StatusModel.FAILED], undefined, newerThan);
+
+		const count = await this.em.count(DeletionRequestEntity, scope.query);
+
+		return count;
+	}
+
 	async update(deletionRequest: DeletionRequest): Promise<void> {
 		const deletionRequestEntity = DeletionRequestMapper.mapToEntity(deletionRequest);
 		const referencedEntity = this.em.getReference(DeletionRequestEntity, deletionRequestEntity.id);