From d97e1df4da4fb82eb916c4ba903179090191e2b1 Mon Sep 17 00:00:00 2001 From: Moshood Mohammed Date: Mon, 31 Aug 2026 10:37:16 +0100 Subject: [PATCH 1/4] fix: Bound bulk operation concurrency (#1183) --- src/lib/bulk/bulkOperations.ts | 48 +++++++++++++++++++++++++++++++--- 1 file changed, 44 insertions(+), 4 deletions(-) diff --git a/src/lib/bulk/bulkOperations.ts b/src/lib/bulk/bulkOperations.ts index 71c6beaa..233c39f2 100644 --- a/src/lib/bulk/bulkOperations.ts +++ b/src/lib/bulk/bulkOperations.ts @@ -1,4 +1,4 @@ -import { apiClient } from '@/lib/api'; +import { apiClient } from '@lib/api'; export type BulkOperationType = 'create' | 'update' | 'delete'; @@ -30,6 +30,8 @@ export interface BulkResult { export interface BulkOptions { /** Batch size for processing (default: 50) */ batchSize?: number; + /** Maximum number of concurrent operations (default: 10) */ + concurrency?: number; /** Progress callback */ onProgress?: (progress: BulkProgress) => void; /** Cancellation token */ @@ -39,16 +41,50 @@ export interface BulkOptions { } const DEFAULT_BATCH_SIZE = 50; +const DEFAULT_CONCURRENCY = 10; /** - * Generic bulk operation processor with batching, progress tracking, and cancellation. + * A simple semaphore to limit concurrency of async operations. + */ +class Semaphore { + private capacity: number; + private active = 0; + private waiters: Array<() => void> = []; + + constructor(capacity: number) { + this.capacity = Math.max(1, Math.floor(capacity)); + } + + async acquire(): Promise { + if (this.active < this.capacity) { + this.active++; + return; + } + await new Promise((resolve) => { + this.waiters.push(resolve); + }); + } + + release(): void { + this.active--; + const next = this.waiters.shift(); + if (next) { + this.active++; + next(); + } + } +} + +/** + * Generic bulk operation processor with batching, progress tracking, cancellation, + * and bounded concurrency via a semaphore. */ async function processBulkOperation( items: T[], operation: BulkOperationType, options: BulkOptions = {}, ): Promise> { - const { batchSize = DEFAULT_BATCH_SIZE, onProgress, signal, endpoint } = options; + const { batchSize = DEFAULT_BATCH_SIZE, concurrency = DEFAULT_CONCURRENCY, onProgress, signal, endpoint } = options; const successful: BulkSuccessItem[] = []; const failed: BulkFailedItem[] = []; let completed = 0; @@ -56,10 +92,11 @@ async function processBulkOperation( const total = items.length; const endpointBase = endpoint || '/api/bulk'; + const semaphore = new Semaphore(concurrency); const reportProgress = (currentItem?: unknown) => { const percentage = total > 0 ? Math.round((completed / total) * 100) : 0; - onProgress?.({ completed, total, percentage, currentItem }); + onProgress?.{ completed, total, percentage, currentItem }); }; // Process in batches @@ -75,6 +112,7 @@ async function processBulkOperation( return { item, success: false, error: new Error('Operation cancelled') } as const; } + await semaphore.acquire(); try { let result: unknown; const url = `${endpointBase}/${operation}`; @@ -100,6 +138,8 @@ async function processBulkOperation( success: false, error: error instanceof Error ? error : new Error(String(error)), } as const; + } finally { + semaphore.release(); } }); From c72eeb5388b0fb9ef225262be671ace032564e93 Mon Sep 17 00:00:00 2001 From: Moshood Mohammed Date: Sat, 5 Sep 2026 09:57:42 +0100 Subject: [PATCH 2/4] fix(ci): resolve failing checks for #1316 --- src/lib/bulk/bulkOperations.ts | 6 ++++-- 1 file changed, 4 insertions(+), 2 deletions(-) diff --git a/src/lib/bulk/bulkOperations.ts b/src/lib/bulk/bulkOperations.ts index 233c39f2..d70e8737 100644 --- a/src/lib/bulk/bulkOperations.ts +++ b/src/lib/bulk/bulkOperations.ts @@ -49,7 +49,9 @@ const DEFAULT_CONCURRENCY = 10; class Semaphore { private capacity: number; private active = 0; - private waiters: Array<() => void> = []; + private waiters: Array<( + () => void + OK> = []; constructor(capacity: number) { this.capacity = Math.max(1, Math.floor(capacity)); @@ -96,7 +98,7 @@ async function processBulkOperation( const reportProgress = (currentItem?: unknown) => { const percentage = total > 0 ? Math.round((completed / total) * 100) : 0; - onProgress?.{ completed, total, percentage, currentItem }); + onProgress?.({ completed, total, percentage, currentItem }); }; // Process in batches From 25d954b0b6459e1cdd8d6043a88a168347940505 Mon Sep 17 00:00:00 2001 From: Moshood Mohammed Date: Sat, 5 Sep 2026 09:57:43 +0100 Subject: [PATCH 3/4] fix(ci): resolve failing checks for #1316 --- .github/workflows/ci.yml | 4 ++++ 1 file changed, 4 insertions(+) diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index 34d1c10e..82aed64f 100644 --- a/.github/workflows/ci.yml +++ b/.github/workflows/ci.yml @@ -6,6 +6,10 @@ on: - main - develop +concurrency: + group: ${{ github.workflow }}-${{ github.ref }} + cancel-in-progress: true + jobs: type-check: runs-on: ubuntu-latest From b540ed085ddefe8affbcc5d3056427b261fc3a3c Mon Sep 17 00:00:00 2001 From: Netty-kun Date: Sun, 6 Sep 2026 16:28:48 +0400 Subject: [PATCH 4/4] fix: repair semaphore type and api import in bounded bulk concurrency --- src/lib/bulk/bulkOperations.ts | 6 ++---- 1 file changed, 2 insertions(+), 4 deletions(-) diff --git a/src/lib/bulk/bulkOperations.ts b/src/lib/bulk/bulkOperations.ts index d70e8737..66bc3baa 100644 --- a/src/lib/bulk/bulkOperations.ts +++ b/src/lib/bulk/bulkOperations.ts @@ -1,4 +1,4 @@ -import { apiClient } from '@lib/api'; +import { apiClient } from '@/lib/api'; export type BulkOperationType = 'create' | 'update' | 'delete'; @@ -49,9 +49,7 @@ const DEFAULT_CONCURRENCY = 10; class Semaphore { private capacity: number; private active = 0; - private waiters: Array<( - () => void - OK> = []; + private waiters: Array<() => void> = []; constructor(capacity: number) { this.capacity = Math.max(1, Math.floor(capacity));