Отмена чтения API-потока в JavaScript без потери состояния интерфейса

Граница ответственности AbortController

AbortController управляет сигналом отмены в клиентском коде. Вызов controller.abort(reason) может сделать последующие действия операции отменёнными, однако это не является подтверждением того, что сервер прекратил вычисление, провайдер перестал передавать данные или биллинг был изменён. Приложение должно отдельно определить, как отображать отмену, ошибку транспорта и ответ провайдера. Скрытый повтор запроса при отмене здесь не используется.

function createOperation() {
  return { id: Symbol("operation"), controller: new AbortController(),
    reader: null, text: "" };
}

Владение операцией и защита от устаревших чанков

При запуске новой операции предыдущая становится устаревшей независимо от того, успела ли она завершиться. Идентификатор операции защищает состояние интерфейса от позднего callback, а signal.aborted не позволяет дописывать данные после отмены. Решение о сохранении уже полученного текста должно быть явным: отмена не обязана автоматически очищать частичный результат. Здесь одна позиция интерфейса: для независимых виджетов нужен отдельный владелец. Возвращаемый handle содержит cancel() и done. Пример принимает уже декодированные строки, не байты fetch и не SSE-фреймы. Состояние старой операции проверяется во всех ветках, включая ошибку и нормальное завершение после abort.

async function consumeStream(operation, stream, onUpdate) {
  const reader = stream.getReader();
  const signal = operation.controller.signal;
  operation.reader = reader;
  let complete = false;
  // AbortSignal alone does not cancel an arbitrary reader.read().
  const cancelRead = () => { void reader.cancel().catch(() => {}); };
  signal.addEventListener("abort", cancelRead, { once: true });
  try {
    if (signal.aborted) cancelRead();
    while (!signal.aborted) {
      const { value, done } = await reader.read();
      if (signal.aborted) break;
      if (done) { complete = true; break; }
      // This example accepts decoded text, not fetch() byte chunks or SSE frames.
      if (typeof value !== "string") throw new TypeError("text chunks required");
      operation.text += value;
      onUpdate(operation.text);
    }
  } finally {
    signal.removeEventListener("abort", cancelRead);
    if (!complete) cancelRead();
    reader.releaseLock();
    if (operation.reader === reader) operation.reader = null;
  }
}

let currentOperation = null;
function startOperation(stream, setState) {
  const operation = createOperation();
  const previous = currentOperation;
  currentOperation = operation;
  previous?.controller.abort();
  const done = consumeStream(operation, stream, (text) => {
    if (currentOperation === operation) setState({ status: "streaming", text });
  }).then(() => {
    if (currentOperation === operation) {
      setState({ status: operation.controller.signal.aborted ? "cancelled" : "complete",
        text: operation.text });
    }
  }).catch(() => {
    if (currentOperation === operation) {
      setState({ status: operation.controller.signal.aborted ? "cancelled" : "error",
        text: operation.text, reason: "stream_stopped" });
    }
  });
  return {
    done,
    cancel() { if (currentOperation === operation) operation.controller.abort(); }
  };
}

Reader, cancel и освобождение lock

Полученный через getReader() reader удерживает lock у ReadableStream. Если чтение закончено, отменено или завершилось ошибкой, освобождение lock следует выполнять в finally. Если приложение владеет самим потоком и хочет сообщить ему о прекращении потребления, можно рассмотреть reader.cancel(reason); это отдельное действие от releaseLock(). Конкретное поведение underlying source и серверного соединения зависит от реализации потока и транспортного слоя, поэтому его нельзя автоматически трактовать как отзыв уже выполненной серверной работы. Сам AbortSignal не прерывает reader.read() произвольного ReadableStream: обработчик abort явно вызывает reader.cancel() и обрабатывает отклонение его promise. Already-aborted сигнал обрабатывается до первого read. В finally удаляется listener и освобождается lock; сырой error и signal.reason не попадают в журнал. Underlying source должен иметь ограниченный, кооперативный cancel; пример не гарантирует завершение произвольного стороннего источника.

Synthetic-модель гонок без fetch

Для локальной проверки не требуется настоящий fetch, ключ API или подключение к провайдеру. Synthetic harness должен позволять вручную выпускать чанки, завершать поток и завершать чтение ошибкой. Отдельные сценарии должны покрывать отмену до первого чанка, отмену после частичного текста, нормальное завершение, ошибку чтения и позднюю доставку события от предыдущей операции. Нижний пример является заготовкой для проверки логики и не заявляет о результатах выполнения в конкретном окружении. Источник ниже не создаёт отдельного незавершимого pull-promise. После cancel или close нельзя вызывать push: это недопустимая операция источника, а не способ проверки гонки. Для позднего события сначала поставьте чанк в очередь, затем замените операцию до выполнения promise continuation.

function makeControlledStream() {
  let source, cancelled = false;
  const stream = new ReadableStream({
    start(controller) { source = controller; },
    cancel() { cancelled = true; }
  });
  return {
    stream,
    push(text) { source.enqueue(text); },
    close() { source.close(); },
    fail(error) { source.error(error); },
    get cancelled() { return cancelled; }
  };
}

Матрица сценариев и пределы вывода

Проверки следует описывать как ожидаемые переходы состояния, а не как обещание поведения внешнего API. Для отмены до первого чанка ожидается отсутствие добавленного текста и состояние cancelled. Для отмены после частичного текста ожидается сохранение или очистка частичного текста в соответствии с решением интерфейса, но не добавление будущих чанков. Для завершения ожидается complete, для ошибки — error, а поздний callback старой операции не должен менять активный результат. Такая матрица не измеряет задержки, серверную отмену, расходы, биллинг, сетевые прокси или совместимость конкретного SDK.

{
  const fixture = makeControlledStream();
  const states = [];
  const handle = startOperation(fixture.stream, state => states.push(state));
  handle.cancel();
  await handle.done;
  if (fixture.stream.locked) throw new Error("reader lock not released");
  if (states.at(-1)?.status !== "cancelled") throw new Error("wrong final state");
}
// Run as an ES module in a runtime with AbortController and ReadableStream.