Реализация SSE и устранение потери сообщений: взгляд со стороны фронтенда

Код и названия в этой статье — упрощённые и изменённые примеры, призванные облегчить понимание.

Введение

Мне поручили добавить функцию чат-бота на основе ИИ в проект, над которым я работал. Когда пользователь вводил вопрос, ИИ должен был сгенерировать ответ и отобразить его на экране, как в чат-боте. Сервер был устроен так, чтобы возвращать результат после завершения обработки ИИ.

На бэкенде уже была конечная точка, возвращавшая Flux<ServerSentEvent<...>>. Я впервые столкнулся с SSE и впервые видел этот код, поэтому прежде чем подключать его к фронтенду, начал с изучения того, что это такое. Но во время тестирования сообщение о завершении так и не пришло.

Причину удалось найти быстрее, чем мы ожидали. Я отвечал за фронтенд, поэтому уже подключил экран к SSE. Мы вместе с разработчиком бэкенда изучили симптомы и сузили круг возможных причин. Разработчик бэкенда внёс изменения на стороне сервера, а я слушал объяснения того, что и почему изменилось, и документировал их. В этой статье с точки зрения фронтенда рассказывается, как на самом деле работает SSE и как, исходя из наблюдаемых симптомов, мы нашли причину проблемы.

1. Что такое SSE?

SSE — это метод, при котором сервер поддерживает одно соединение открытым и передаёт данные клиенту по мере их появления. В браузерах есть встроенный API для их получения — EventSource.

Типичный запрос и ответ завершаются следующим образом.

클라이언트: GET /api/data
서버: [응답 완성] → 200 OK + 본문 전체 → 연결 종료

Ответ SSE не завершается.

클라이언트: GET /api/tasks/{taskId}/stream
서버: 200 OK, Content-Type: text/event-stream
서버: data: {"status":"GENERATING"}\n\n
서버: data: {"status":"GENERATING"}\n\n
서버: data: {"status":"SUCCESS"}\n\n
서버: (연결 종료)

Согласно спецификации, сообщение считается завершённым, когда после data: следуют два символа перевода строки. Браузер разбирает данные в этом формате и при получении каждого фрагмента генерирует событие message.

Поэтому фронтенду достаточно один раз зарегистрировать обработчик.

eventSource.addEventListener("message", handleEventMessage);

После такой регистрации, если сервер отправит 100 фрагментов, handleEventMessageтоже будет вызван 100 раз.

Это напрямую соответствует Fluxна бэкенде. Flux— это тип, который последовательно передаёт несколько значений с течением времени, а Spring WebFlux кодирует каждое значение из Fluxкак data: {...}и записывает его в поток ответа. Количество раз, которое Fluxвыдаёт значение, совпадает с количеством вызовов обработчика события message

.

2. Что, если бы мы использовали другой подход?

На самом деле у нас был только один вариант. На бэкенде уже была конечная точка SSE, и мне оставалось только подключиться к ней. И всё же во время настройки я задумался, обязательно ли использовать SSE и как бы выглядело решение с другим подходом, поэтому сравнил его с альтернативами.

При опросе API-запрос выполняется повторно через короткие интервалы. Это самый простой в реализации подход, который работает в любой серверной среде, но он порождает множество ненужных запросов, а скорость реакции ограничена интервалом опроса.

При длительном опросе отправляется запрос, а сервер задерживает ответ до изменения состояния. Этот подход обеспечивает более быструю реакцию, чем обычный опрос, но серверу приходится долго поддерживать соединения открытыми. Его преимущества перед SSE неочевидны, при этом реализация становится сложнее.

	Опрос	Длительный опрос	SSE	WebSocket
Направление	Одностороннее	Одностороннее	Одностороннее	Двустороннее
Сложность реализации	Очень низкая	Средняя	Низкая	Относительно высокая
Скорость реакции	Низкая	Средний	Высокий	Очень высокий
Автоматическое переподключение	Н/Д	Реализовано напрямую	Встроенная поддержка браузера	Реализовано напрямую

В нашем случае передача данных была однонаправленной: нам нужно было только отправить клиенту результаты, сформированные сервером. После того как пользователь отправлял вопрос, клиенту больше нечего было отправлять на сервер, а серверу нужно было лишь отправить результат после завершения обработки.

Если учитывать только это требование, было бы достаточно опроса. Однако при опросе приходится отправлять запросы через регулярные интервалы, не зная, когда результат будет готов, а время его получения зависит от этого интервала. С SSE клиент может ждать по одному соединению и получить результат, как только он станет доступен. Это подходило для интерфейса чат-бота, которому нужно было отображать состояние ожидания, а затем сразу переключаться на результат. Оглядываясь назад, думаю, что использовать SSE, который уже был доступен, тоже было вполне разумным решением.

3. Проблемы с аутентификацией JWT и выбор библиотеки

Сначала наш чат-бот вызывал API, чтобы отправить вопрос и получить идентификатор задачи, а затем открывал SSE-соединение, используя этот идентификатор.

Реализация на стороне фронтенда не казалась сложной. EventSourceэкземпляр и добавить обработчик — и готово.

const eventSource = new EventSource(url);

eventSource.onmessage = (event) => {
  const data = JSON.parse(event.data);
  console.log(data);
};

Однако, когда я попробовал, сразу же получил ошибку аутентификации.

Встроенный в браузер EventSource не поддерживает пользовательские заголовки. В конструктор можно передать только URL и withCredentialsОбе библиотеки. В нашем проекте для аутентификации использовались токены JWT в заголовках запросов, но добавить такой заголовок было просто невозможно.

Я нашел две библиотеки, которые позволяли обойти эту проблему.

event-source-polyfill использует внутри XHR, чтобы имитировать API EventSource с поддержкой пользовательских заголовков. Его API почти идентичен встроенному в браузер EventSource , поэтому существующий код с использованием addEventListener или close() и других методов можно использовать без изменений.

@microsoft/fetch-event-source — это библиотека, которая повторно реализует SSE на основе fetch() , а не полифилл. Вы можете использовать все возможности, которые предоставляет fetch: можно отправлять POST-запросы, напрямую обращаться к объекту ответа для проверки и полностью контролировать стратегию повторных попыток. Кроме того, библиотека закрывает соединение, когда вкладка переходит в фоновый режим, и подключается заново, когда вкладка снова становится видимой.

Если судить только по функциональности, второй вариант гибче. Однако наш бэкенд использовал GET, а на другом экране проекта мы уже использовали event-source-polyfill. Не было смысла подключать две библиотеки для одной и той же задачи, а возможность сохранить существующую структуру кода почти без изменений стала важным аргументом. Поэтому мы выбрали первый вариант.

import { EventSourcePolyfill } from "event-source-polyfill";

const eventSource = new EventSourcePolyfill(url, {
  headers: {
    Authorization: `Bearer ${token}`,
  },
  heartbeatTimeout: 5 * 60 * 1000,
});

eventSource.addEventListener("message", handleEventMessage);
eventSource.addEventListener("error", handleEventError);

4. Симптом: после первого сообщения ответа нет

Когда я проверил работу, первое сообщение пришло без проблем.

eventMessage { status: "GENERATING", message: "Generating response...", ... }

Но на этом всё. Даже спустя долгое время после того, как ИИ должен был завершить генерацию ответа, сообщение о завершении так и не пришло. Через пять минут ожидания я получил такую ошибку.

EventSource failed.
Error: No activity within 300000 milliseconds. 277 chars received. Reconnecting.

За пять минут, заданных параметром heartbeatTimeout, данные так и не поступили, поэтому библиотека отключилась и попыталась подключиться заново. 277 символов — значит, пришло только самое первое сообщение.

5. Анализ первопричины

5-1. Исключение фронтенда и повторной отправки сообщения

Сначала нужно было исключить любую вероятность того, что проблема связана с фронтендом. Запустив чат-бота в браузере, я скопировал идентификатор задачи и токен прямо из инструментов разработчика и отправил запрос на тот же эндпоинт с помощью инструмента для тестирования API.

Результат был тем же. Пришло только первое сообщение, а затем поток остановился. Поскольку тот же симптом наблюдался и в запросе, при котором не участвовали React, библиотека и регистрация обработчика, фронтенд исключили из числа подозреваемых. Заголовок ответа Content-Type также имел значение text/event-stream, так что дело было не в формате.

Когда я отправлял один и тот же запрос несколько раз, каждый раз, когда я отправлял егоGENERATINGСообщение приходило сразу, и его содержимое всегда было одинаковым. Изучив код бэкенда, я нашёл причину.

public Flux<ServerSentEvent<TaskMessage>> subscribe(AiTask task) {
    String taskId = task.getId();

    Sinks.Many<TaskMessage> channel = this.channelMap.computeIfAbsent(taskId,
            id -> Sinks.many().replay().all());

    channel.tryEmitNext(TaskMessage.of(task, GENERATING));

    return channel.asFlux().map(m -> ServerSentEvent.builder(m).build());
}

tryEmitNext безусловно выполняется в теле метода — ещё до того, как метод возвращает Flux. Сообщение создавалось не потому, что сервер обнаруживал прогресс и отправлял его: оно появлялось именно в этот момент, просто потому, что поступал запрос. Иными словами, мой запрос не получал событие — он его создавал. Статус тоже всегда был GENERATING, поэтому сколько бы раз я ни вызывал метод, содержимое оставалось одинаковым.

5-2. Почему сообщение о завершении было отброшено

Оставался вопрос: почему сообщение о завершении так и не приходило. Код обработки завершения выглядел так.

private void completeTask(TaskMessage message) {
    String taskId = message.getTaskId();

    Sinks.Many<TaskMessage> channel = this.channelMap.remove(taskId);

    if (channel != null) {
        channel.tryEmitNext(message);
        channel.tryEmitComplete();
    }
}

Канал добавлялся в channelMap только один раз — в уже упомянутом нами вызове computeIfAbsent. Но этот код выполняется только тогда, когда фронтенд пытается установить SSE-соединение.

Проблема возникала, когда вызов ИИ сразу завершался с ошибкой и заканчивался гораздо быстрее обычного. Обработка завершения запускалась до того, как устанавливалось SSE-соединение с фронтендом. Вот как выглядел общий порядок действий.

Поскольку код пытался получить канал, который никогда не был зарегистрирован, removeвозвращалnull. Поскольку войти в условный блок не удавалось, для сообщения о завершении не находилось получателя, и оно отбрасывалось, не оставив в логах ни единой строки. Более того, обработчик завершения был устроен так, что удалял канал из карты сразу после его получения. Поэтому даже если бы канал там был, подключившийся с опозданием клиент не смог бы его найти.

6. Исправление — регистрация каналов и изменение способа их получения

Причин было две. Канал каждой задачи следовало добавлять в карту при регистрации задачи, но этот шаг отсутствовал. Кроме того, при обработке завершения канал удалялся из карты сразу после получения. Бэкенд-инженер внёс исправления в трёх местах.

Сначала мы добавили канал в карту при регистрации задачи. Событие вызова ИИ отправляется отдельно, асинхронно, после завершения этого метода.

public String registerTask(String prompt) {
    AiTask task = taskRepository.save(AiTask.create(prompt));

    this.channelMap.put(task.getId(), Sinks.many().replay().all());

    return task.getId();
}

Затем в обработчике завершения мы заменили получение канала с помощью removeна get.

private void completeTask(TaskMessage message) {
    Sinks.Many<TaskMessage> channel = this.channelMap.get(message.getTaskId());

    if (channel != null) {
        channel.tryEmitNext(message);
        channel.tryEmitComplete();
    }
}

Наконец, мы перенесли удаление канала на сторону SSE-соединения. Как только подписчик получает сигнал о завершении, канал удаляется из карты. Мы также добавили условие, чтобы ожидающее сообщение не отправлялось повторно, если подписчик уже подключён.

public Flux<ServerSentEvent<TaskMessage>> subscribe(AiTask task) {
    String taskId = task.getId();

    Sinks.Many<TaskMessage> channel = this.channelMap.computeIfAbsent(taskId,
            id -> Sinks.many().replay().all());

    if (channel.currentSubscriberCount() == 0) {
        channel.tryEmitNext(TaskMessage.of(task, GENERATING));
    }

    return channel.asFlux()
            .doOnComplete(() -> this.channelMap.remove(taskId))
            .map(m -> ServerSentEvent.builder(m).build());
}

Теперь, даже если работа ИИ завершается до установления SSE-соединения, сообщение о завершении сохраняется в канале, который остаётся в карте. Если фронтенд подключается с опозданием, он подписывается на тот же канал, а replay().all()повторно отправляет с самого начала все сообщения, накопившиеся за это время, поэтому фронтенд тоже может получить сообщение о завершении. К этому моменту канал уже закрыт, поэтому ожидающее сообщение, которое было бы отправлено при установлении соединения, не доставляется и игнорируется. После получения результата канал также удаляется из карты.

Независимо от того, что происходило раньше — устанавливалось соединение или завершалась работа ИИ, — теперь фронтенд получал один и тот же результат.

Заключение

Больше, чем сама технология SSE, мне запомнился процесс последовательного сужения области поиска проблемы. Сначала я подозревал собственный код — например, заголовки или настройки библиотеки. Но когда я отправил тот же запрос с помощью инструмента для тестирования API, сразу стало ясно, что проблема не на стороне фронтенда. Это выяснилось в первую очередь, и стало понятно, где искать причину при обсуждении с бэкенд-инженером.

Совместный разбор кода сервера также показал мне, насколько важен порядок выполнения в асинхронной структуре. Код отправки и код получения по отдельности казались правильными, но именно порядок определял, когда создавался канал для получения и когда он удалялся. Я понял, что даже если ваша задача на фронтенде — только обрабатывать соединение, необходимо понимать, когда и где сервер размещает сообщения перед отправкой. Это поможет быстрее находить подобные проблемы.

Понг