Method parallelLimit

packages/db/src/buffers/base-buffer.ts:215–247

* Run `fn` over `items` with a bounded concurrency. Preserves order of * `results`. If any worker throws, the helper propagates the first error * (consistent with `Promise.all`); other in-flight promises run to * completion but their resolved values/errors are discarded.

(
    items: T[],
    fn: (item: T, index: number) => Promise<R>,
    concurrency: number = this.chInsertConcurrency
  )

   * completion but their resolved values/errors are discarded.
   */
  protected async parallelLimit<T, R>(
    items: T[],
    fn: (item: T, index: number) => Promise<R>,
    concurrency: number = this.chInsertConcurrency
  ): Promise<R[]> {
    if (items.length === 0) {
      return [];
    }
    if (concurrency <= 1 || items.length === 1) {
      const out: R[] = [];
      for (let i = 0; i < items.length; i++) {
        out.push(await fn(items[i] as T, i));
      }
      return out;
    }
    const results = new Array<R>(items.length);
    let nextIndex = 0;
    const worker = async (): Promise<void> => {
      while (true) {
        const idx = nextIndex++;
        if (idx >= items.length) {
          return;
        }
        results[idx] = await fn(items[idx] as T, idx);
      }
    };
    const workers = Array.from(
      { length: Math.min(concurrency, items.length) },
      () => worker()
    );
    await Promise.all(workers);
    return results;
  }

  /**
   * Subclasses call this from within onFlush to record what they actually