워크플로우 에디터를 개발하면서 실행 중인 노드의 상태를 실시간으로 보여주는 기능이 필요했다.

하지만 일반적인 API처럼 실행이 모두 끝난 뒤 결과를 한 번에 받아 완료 상태를 표시하면, 실행되는 동안 사용자는 현재 어디까지 진행됐는지 알 수 없었다.
이를 구현하기 위해 SSE를 사용하게 되었는데 그 과정에서 서로 다른 실행의 상태가 섞이는 문제가 생겼고,
새 실행을 시작하기 전에 기존 스트림을 AbortController로 정리하는 방식으로 해결했다.

이번 글에서는 SSE와 AbortController가 각각 무엇인지 살펴보고, 실제 워크플로우 에디터에 어떻게 적용하여 해결했는지를 작성해보려고 한다!

1. SSE란?

일반적인 HTTP 요청은 요청 한 번에 응답 한 번을 받는다.

클라이언트 ── 요청 ──▶ 서버
클라이언트 ◀─ 응답 ─── 서버

하지만 작업 진행 상태처럼 서버에서 중간 데이터를 계속 받아야 하는 경우도 있다.

작업 시작 → 30% 완료 → 60% 완료 → 작업 완료

이럴 때 사용할 수 있는 방식 중 하나가 바로 SSE(Server-Sent Events) , HTTP 연결을 열어둔 상태에서 서버가 클라이언트에게 데이터를 여러 번 보내는 데 사용할 수 있는 WEB API이다.

서버는 아래와 같은 형태로 이벤트를 보낼 수 있다.

event: progress
data: {"percent":30}

event: progress
data: {"percent":60}

event: completed
data: {"result":"success"}

여기서 event는 이벤트 이름이고, data는 전달할 데이터다.
아래에서 나오겠지만 progress, completed 같은 이벤트 이름은 SSE에서 정해놓은 것이 아니라 서버와 클라이언트가 서로 정해서 사용하면 된다.

2. SSE는 어떻게 사용할까?

브라우저에서는 SSE를 사용할 수 있도록 EventSource라는 API를 기본으로 제공한다.

const eventSource = new EventSource("/api/stream");

이렇게 연결해두면 서버에서 이벤트가 올 때마다 프론트에서 받아 처리할 수 있다.
예를 들어 서버에서 다음과 같이 progress이벤트를 보낸다고 해보자.

event: progress
data: {"percent":50}

프론트에서는 같은 이벤트 이름으로 기다리면 된다.

eventSource.addEventListener("progress", (event) => {
  const data = JSON.parse(event.data);

  console.log(data.percent);
});

서버가 progress 이벤트를 여러 번 보내면 등록한 함수도 그때마다 실행된다.

progress 30%
progress 60%
progress 90%

더 이상 이벤트를 받을 필요가 없다면 직접 연결을 닫을 수 있다.

eventSource.close();

EventSource만으로 부족한 경우

EventSource는 SSE를 간단하게 사용할 수 있다는 장점이 있지만, 요청을 자유롭게 구성하기는 어렵다.
기본적으로 GET 방식으로 연결하기 때문에 일반적인 fetch처럼 POST 요청을 보내거나 body에 데이터를 담을 수 없다.
예를 들어 fetch에서는 아래처럼 요청할 수 있다.

fetch("/api/stream", {
  method: "POST",
  body: JSON.stringify({
    message: "hello",
  }),
});

하지만 EventSource에서는 이런 방식으로 사용할 수 없다.

이럴 때는 POST와 body를 사용할 수 있으면서 SSE 이벤트도 받을 수 있는 @microsoft/fetch-event-source를 사용할 수 있다.

await fetchEventSource("/api/stream", {
  method: "POST",
  body: JSON.stringify({
    nodes,
    edges,
  }),

  onmessage(event) {
    console.log(event.data);
  },
});

즉 서버에서 이벤트를 받기만 한다면 EventSource로 충분하지만, 데이터를 함께 보내면서 SSE 연결을 유지해야 한다면 fetch 기반 방식이 더 적합하다.

3. AbortController란?

SSE처럼 오래 유지되는 요청은 중간에 더 이상 필요하지 않을 수도 있다.
이때 요청을 중단할 수 있게 해주는 브라우저 API가 AbortController다.

먼저 AbortController를 만든다.

const controller = new AbortController();

그리고 요청에 signal을 전달해 둘을 연결한다.

fetch("/api/data", {
  signal: controller.signal,
});

이후 더 이상 요청이 필요하지 않다면 abort()를 호출한다.

controller.abort();

fetch 기반 SSE에서도 같은 방식으로 사용할 수 있다.

const controller = new AbortController();

await fetchEventSource("/api/stream", {
  signal: controller.signal,

  onmessage(event) {
    console.log(event.data);
  },
});

이 상태에서 abort()를 호출하면 열려 있던 SSE 연결을 끊고 더 이상 이벤트를 받지 않는다.

controller.abort();

여기서 중요한 점은 한 번 abort()한 controller는 다시 사용할 수 없기 때문에, 새로운 요청을 시작할 때는 새로운 AbortController를 만들어야 한다.

또한 abort()는 프론트에서 요청이나 SSE 연결을 끊는 것일 뿐이다.
서버에서 이미 실행 중인 작업까지 자동으로 멈추는 것은 아니기 때문에, 서버 작업 자체를 중단해야 한다면 별도의 처리가 필요하다.

4. 워크플로우 에디터에는 어떻게 적용했을까?

워크플로우를 실행하면 서버에서 다음과 같은 SSE 이벤트를 보내줬다.
WorkflowStarted, NodeStarted, NodeCompleted, NodeFailed, WorkflowCompleted

Canvas에서 실행 이벤트 처리하기

이렇게 SSE에서 받은 이벤트는 Canvas에서 공통 callback으로 처리했다.

const callbacks = {
  onNodeStarted: (nodeUuid: string) => {
    // 실행 중 상태로 변경
  },

  onNodeCompleted: (nodeUuid: string) => {
    // 완료 상태로 변경
  },

  onNodeFailed: (nodeUuid: string) => {
    // 실패 상태로 변경
  },

  onWorkflowCompleted: (status: string) => {
    // 전체 실행 완료 처리
  },
};

예를 들어 node-a의 실행이 시작되면

onNodeStarted("node-a");

가 호출되고, Canvas에서는 해당 Node를 실행 중 상태로 바꾼다.

onNodeStarted("node-a")
        ↓
runningNodeIds에 node-a 추가
        ↓
node-a를 실행 중으로 표시

완료됐을 때도 같은 방식이다.

onNodeCompleted("node-a")
        ↓
runningNodeIds에서 제거
completedNodeIds에 추가
        ↓
node-a를 완료 상태로 표시

Store와 Node UI

SSE에서 실행 이벤트를 받으면 Zustand Store에서 각 Node의 현재 상태를 관리했다.

{
  runningNodeIds: new Set<string>(), // 노드 id 포함
  completedNodeIds: new Set<string>(),
  failedNodeIds: new Set<string>(),
}

각 Set에는 해당 상태에 있는 Node의 ID를 넣어주었다.
실행이 끝나면 실행 중 목록에서는 빼고 완료 목록에 넣는다.

이렇게 해서 Node 컴포넌트에서는 자신의 ID가 어느 Set에 들어 있는지만 확인하는 방식을 사용했다.

const isRunning = runningNodeIds.has(nodeId); // 실행중인 노드인가 -> pending ui
const isCompleted = completedNodeIds.has(nodeId); // 완료된 노드인가 -> check ui
const isFailed = failedNodeIds.has(nodeId); // 실패한 노드인가 -> failed ui

이렇게 상태를 따로 관리해 Node 컴포넌트는 SSE 이벤트를 직접 처리할 필요 없이, 현재 상태에 맞춰 UI만 그리도록 했다.

5. 이전 스트림이 새 실행에 섞이는 문제 발생

SSE를 적용한 뒤 실행을 반복해서 테스트하면서 문제가 발생했다.
원인은 첫 번째 실행이 끝나기 전에 다시 실행하면, 새 실행이 시작된 뒤에도 이전 SSE 연결이 남아있던 것이었다.
그러면 이전 실행의 이벤트가 뒤늦게 들어오면서 새 실행의 상태와 섞이게 되었다.
그래서 아직 실행하지 않은 Node가 완료 표시가 되기도 했다..^^

이를 막기 위해 새 실행을 시작하기 전에 기존 SSE 연결을 먼저 끊는 방식으로 해결했다.

get().abortController?.abort(); // 기존 SSE 스트림 종료
const abortController = new AbortController(); // 새 AbortController 생성

set({
  runningNodeIds: new Set(),
  completedNodeIds: new Set(),
  failedNodeIds: new Set(),
  abortController,
});

사용자가 직접 중지 버튼을 누른 경우에는 서버에서 실행 중인 워크플로우 자체도 멈춰야 했다.
앞서 설명했듯 abortController.abort()는 프론트의 SSE 연결만 끊기 때문이다.

await StopWorkflow(executionId);
abortController.abort();

다시 실행할 때는 이전 SSE 스트림을 정리하고, 중지할 때는 서버의 실행과 SSE 스트림을 함께 종료하는 방식으로 SSE 중복 오류를 해결했다!