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 diff --git a/src/lib/bulk/bulkOperations.ts b/src/lib/bulk/bulkOperations.ts index 71c6beaa..66bc3baa 100644 --- a/src/lib/bulk/bulkOperations.ts +++ b/src/lib/bulk/bulkOperations.ts @@ -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,6 +92,7 @@ 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; @@ -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(); } });