Типизация асинхронной конкурентности: AbortSignal, каналы и очереди

+-------------------------------------------------------------------------+
|                  АРХИТЕКТУРА КОНКУРЕНТНОГО КОНВЕЙЕРА                    |
+-------------------------------------------------------------------------+
| [Producer (SSE / WebSocket / LLM Stream)]                               |
|   |--> Порождает поток сообщений: AsyncIterable<ServerPacket>           |
+-------------------------------------------------------------------------+
                                    |
                                    v (Backpressure Queue / Channel<T>)
+-------------------------------------------------------------------------+
| [Типобезопасный буферизированный канал: Channel<ServerPacket>]          |
|   |-- push(packet: T): Promise<void>  (Блокируется при переполнении)    |
|   |-- pop(signal?: AbortSignal): Promise<T> (Блокируется при пустоте)   |
|   `-- close(): void                                                     |
+-------------------------------------------------------------------------+
                                    |
                                    v (Worker Pool: Concurrency Limit = 3)
+-------------------------------------------------------------------------+
| [Параллельный пул потребителей (Consumers / Workers)]                   |
|   |-- Worker 1: Обработка пакета A (AbortSignal.timeout(5000))          |
|   |-- Worker 2: Обработка пакета B (Прерывание по abort())              |
|   `-- Worker 3: Обработка пакета C (Успешная финализация)               |
+-------------------------------------------------------------------------+

Композиция сигналов отмены:
[Сигнал тайм-аута: AbortSignal.timeout(5000)] -----\
                                                    +--> [AbortSignal.any([...])] --> fetch(url, { signal })
[Сигнал кнопки UI: userCancelController.signal] ---/

Запрос прерывается, если прошло 5 секунд ИЛИ если пользователь нажал «Отмена»!

export interface FetchWithTimeoutOptions extends RequestInit {
  readonly timeoutMs?: number;
  readonly externalSignal?: AbortSignal;
}

/**
 * Выполняет сетевой запрос с композитной отменой: по таймауту ИЛИ по внешнему сигналу.
 */
export async function fetchWithResilience<TResponse>(
  url: string,
  options: FetchWithTimeoutOptions = {}
): Promise<TResponse> {
  const { timeoutMs = 8000, externalSignal, ...fetchOptions } = options;

  // 1. Создаем сигнал таймаута
  const timeoutSignal = AbortSignal.timeout(timeoutMs);

  // 2. Объединяем таймаут и внешний сигнал через AbortSignal.any
  const combinedSignal = externalSignal
    ? AbortSignal.any([timeoutSignal, externalSignal])
    : timeoutSignal;

  try {
    const response = await fetch(url, {
      ...fetchOptions,
      signal: combinedSignal,
    });

    if (!response.ok) {
      throw new Error(`[HttpError] HTTP ${response.status}: ${response.statusText}`);
    }

    return (await response.json()) as TResponse;
  } catch (error: unknown) {
    if (combinedSignal.aborted) {
      // Различаем причину отмены: таймаут или действие пользователя
      if (timeoutSignal.aborted) {
        throw new Error(`[TimeoutError] Запрос к ${url} превысил лимит в ${timeoutMs}ms`);
      }
      throw new Error(`[AbortError] Запрос к ${url} был отменен вызывающей стороной`);
    }

    throw error instanceof Error ? error : new Error(String(error));
  }
}

interface Resolver<T> {
  resolve: (value: T) => void;
  reject: (reason?: unknown) => void;
}

/**
 * Типобезопасный асинхронный канал передачи данных (Communicating Sequential Processes).
 */
export class AsyncChannel<T> implements AsyncIterable<T> {
  private readonly buffer: T[] = [];
  private readonly receivers: Resolver<T>[] = [];
  private isClosed = false;

  constructor(private readonly maxCapacity: number = Infinity) {}

  /**
   * Запись элемента в канал.
   */
  public push(item: T): boolean {
    if (this.isClosed) {
      throw new Error('[ChannelError] Попытка записи в закрытый канал');
    }

    // Если есть ждущие потребители — отдаем первому в очереди
    if (this.receivers.length > 0) {
      const receiver = this.receivers.shift()!;
      receiver.resolve(item);
      return true;
    }

    // Проверка емкости буфера
    if (this.buffer.length >= this.maxCapacity) {
      // Буфер переполнен: вытесняем старейший элемент (Drop-oldest backpressure)
      this.buffer.shift();
    }

    this.buffer.push(item);
    return true;
  }

  /**
   * Асинхронное чтение элемента из канала с поддержкой отмены.
   */
  public async pop(signal?: AbortSignal): Promise<T | null> {
    if (this.buffer.length > 0) {
      return this.buffer.shift()!;
    }

    if (this.isClosed) {
      return null;
    }

    return new Promise<T | null>((resolve, reject) => {
      const resolver: Resolver<T> = {
        resolve: (val) => {
          cleanUp();
          resolve(val);
        },
        reject: (err) => {
          cleanUp();
          reject(err);
        },
      };

      const onAbort = () => {
        const index = this.receivers.indexOf(resolver);
        if (index > -1) {
          this.receivers.splice(index, 1);
        }
        cleanUp();
        reject(new Error('[AbortError] Ожидание канала прервано по сигналу'));
      };

      const cleanUp = () => {
        signal?.removeEventListener('abort', onAbort);
      };

      if (signal?.aborted) {
        onAbort();
        return;
      }

      signal?.addEventListener('abort', onAbort);
      this.receivers.push(resolver);
    });
  }

  public close(): void {
    this.isClosed = true;
    // Оповещаем всех ждущих читателей о закрытии
    while (this.receivers.length > 0) {
      const receiver = this.receivers.shift()!;
      receiver.resolve(null as unknown as T);
    }
  }

  /**
   * Реализация протокола AsyncIterable для использования в цикле for await.
   */
  public async *[Symbol.asyncIterator](): AsyncIterator<T> {
    while (true) {
      const value = await this.pop();
      if (value === null && this.isClosed) {
        break;
      }
      if (value !== null) {
        yield value;
      }
    }
  }
}

export type AsyncTask<T> = (signal: AbortSignal) => Promise<T>;

/**
 * Ограничивает количество параллельно выполняемых асинхронных задач.
 */
export class ConcurrencyPool {
  private activeCount = 0;
  private readonly queue: Array<() => void> = [];

  constructor(public readonly concurrencyLimit: number) {
    if (concurrencyLimit < 1) {
      throw new RangeError('Лимит параллелизма обязан быть >= 1');
    }
  }

  /**
   * Запускает задачу с соблюдением пула параллелизма и контролем отмены.
   */
  public run<T>(task: AsyncTask<T>, signal?: AbortSignal): Promise<T> {
    return new Promise<T>((resolve, reject) => {
      const execute = async () => {
        if (signal?.aborted) {
          this.next();
          reject(new Error('[AbortError] Задача отменена до начала исполнения'));
          return;
        }

        this.activeCount++;
        try {
          // Создаем локальный контроллер с поддержкой внешнего сигнала
          const ctrl = new AbortController();
          const onAbort = () => ctrl.abort();
          signal?.addEventListener('abort', onAbort);

          const result = await task(ctrl.signal);
          signal?.removeEventListener('abort', onAbort);
          resolve(result);
        } catch (err: unknown) {
          reject(err);
        } finally {
          this.activeCount--;
          this.next();
        }
      };

      if (this.activeCount < this.concurrencyLimit) {
        void execute();
      } else {
        this.queue.push(execute);
      }
    });
  }

  private next(): void {
    if (this.queue.length > 0 && this.activeCount < this.concurrencyLimit) {
      const nextTask = this.queue.shift();
      nextTask?.();
    }
  }
}