fix(realtime): 멈춘 SSE 구독자가 전체 fan-out 을 세우던 문제

docker-compose.yml

@@ -172,6 +172,7 @@ services:
       REALTIME_CORE_BASE_URL: ${CORE_INTERNAL_BASE_URL:-http://backend:38010}
       REALTIME_INTERNAL_API_KEY: ${CORE_INTERNAL_API_KEY:-local-development-internal-api-key}
       REALTIME_WS_WRITE_TIMEOUT: ${REALTIME_WS_WRITE_TIMEOUT:-10s}
+      REALTIME_SSE_WRITE_TIMEOUT: ${REALTIME_SSE_WRITE_TIMEOUT:-10s}
     depends_on:
       rabbitmq:
         condition: service_healthy

realtime/.env.example

@@ -13,5 +13,8 @@ REALTIME_JWT_SECRET=local-development-jwt-secret-must-be-replaced
 REALTIME_CORE_BASE_URL=http://localhost:38010
 REALTIME_INTERNAL_API_KEY=local-development-internal-api-key
 REALTIME_WS_WRITE_TIMEOUT=10s
+# SSE write 타임아웃. 없으면 멈춘 클라이언트가 쓰기 고루틴을 무기한 붙잡고,
+# 그 여파로 다른 세션의 이벤트까지 늦어진다.
+REALTIME_SSE_WRITE_TIMEOUT=10s
 # AI 음성 스트림 WS 프록시 대상 (compose 내부에서는 ws://ai:8000/internal/voice/stream)
 REALTIME_AI_WS_URL=ws://ai:8000/internal/voice/stream

realtime/CLAUDE.md

@@ -138,7 +138,19 @@ Heartbeat (proxy keepalive):
 
 - `session.Registry`는 sync.RWMutex로 보호.
 - 한 sessionId에 여러 SSE 연결(다중 디바이스/탭) 가능 → slice.
-- slow consumer 처리: `Dispatch`가 `slowTimeout`을 초과하면 그 구독자만 drop, 다른 구독자는 영향 없음.
+- slow consumer 처리: `Dispatch`가 `slowTimeout`을 초과하면 그 구독자만 drop.
+  - **`slowTimeout` 은 호출 전체의 예산이다 — 구독자마다 새로 세지 않는다.** 2026-09-29
+    이전에는 구독자마다 리셋해서 멈춘 탭 3개면 5초 × 3 = 15초가 걸렸다. AMQP 컨슈머는
+    단일 스레드 + `prefetch=1` 이라(`messaging.Consumer`) 여기서 막히는 시간이 곧 **다른
+    세션들의 이벤트 지연**이다. "다른 구독자는 영향 없음" 이라고 적혀 있었지만 사실이
+    아니었다 — 같은 채널의 뒷 구독자도, 다른 채널의 세션도 같이 밀렸다.
+  - 예산을 다 써도 남은 구독자에게 **논블로킹 전송**은 시도한다. 앞사람이 멈췄다는 이유로
+    건강한 구독자가 이벤트를 잃으면 안 된다.
+- **SSE 쓰기에도 데드라인이 있다**(`REALTIME_SSE_WRITE_TIMEOUT`, 기본 10s). WS 는 처음부터
+  `WSWriteTimeout` 이 있었지만 SSE 는 없어서, 멈춘 클라이언트(노트북 덮개를 닫은 경우 등)가
+  TCP 를 열어둔 채 수신 창을 막으면 `Fprintf` 가 무기한 블록됐다. 그 고루틴이 `sub.Ch` 를
+  비우지 못하면 버퍼가 차고, 그 채널로 가는 모든 `Dispatch` 가 slow-consumer 타임아웃을
+  물게 된다. 데드라인이 있어야 포기하고 구독을 해제할 수 있다.
 - 단일 인스턴스 가정. 다중 인스턴스 전환 시 fanout exchange + 인스턴스별 자기 큐 패턴 필요.
 
 ---
@@ -159,6 +171,7 @@ Heartbeat (proxy keepalive):
 | `REALTIME_INTERNAL_API_KEY` | `local-development-internal-api-key` | **Core `CORE_INTERNAL_API_KEY`와 동일값 필수** (AI 서버도 공유). `X-Internal-API-Key` |
 | `REALTIME_AI_WS_URL` | `ws://localhost:8000/internal/voice/stream` | AI 음성 스트림 WS base URL (RT3 오디오 프록시 업스트림). compose 내부는 `ws://ai:8000/internal/voice/stream` |
 | `REALTIME_WS_WRITE_TIMEOUT` | `10s` | WS write 타임아웃 |
+| `REALTIME_SSE_WRITE_TIMEOUT` | `10s` | SSE write 타임아웃. 없으면 멈춘 클라이언트가 쓰기 고루틴을 무기한 붙잡는다 |
 
 ---
 

realtime/cmd/realtime/main.go

@@ -37,7 +37,7 @@ func main() {
 		Handler:   handlerAdapter{d: dispatcher},
 	}
 
-	sseHandler := transport.NewSSEHandler(registry, cfg.SSEBufferSize, cfg.SSEPingInterval)
+	sseHandler := transport.NewSSEHandler(registry, cfg.SSEBufferSize, cfg.SSEPingInterval, cfg.SSEWriteTimeout)
 	verifier := auth.NewStreamTokenVerifier(cfg.JWTSecret)
 	coreClient := core.NewClient(cfg.CoreBaseURL, cfg.InternalApiKey, cfg.WSWriteTimeout)
 	wsHandler := transport.NewWSHandler(registry, coreClient, cfg.WSWriteTimeout)

realtime/internal/config/config.go

@@ -7,16 +7,20 @@ import (
 )
 
 type Config struct {
-	ListenAddr             string        `env:"REALTIME_LISTEN_ADDR" envDefault:":38020"`
-	RabbitMQURL            string        `env:"REALTIME_RABBITMQ_URL" envDefault:"amqp://stackup:stackup@localhost:38050/"`
-	LogLevel               string        `env:"REALTIME_LOG_LEVEL" envDefault:"info"`
-	QueueName              string        `env:"REALTIME_QUEUE_NAME" envDefault:"q.realtime.session.notify"`
-	JWTSecret              string        `env:"REALTIME_JWT_SECRET" envDefault:"local-development-jwt-secret-must-be-replaced"`
-	CoreBaseURL            string        `env:"REALTIME_CORE_BASE_URL" envDefault:"http://localhost:38010"`
-	InternalApiKey         string        `env:"REALTIME_INTERNAL_API_KEY" envDefault:"local-development-internal-api-key"`
-	AIWSURL                string        `env:"REALTIME_AI_WS_URL" envDefault:"ws://localhost:8000/internal/voice/stream"`
-	WSWriteTimeout         time.Duration `env:"REALTIME_WS_WRITE_TIMEOUT" envDefault:"10s"`
-	SSEPingInterval        time.Duration `env:"REALTIME_SSE_PING_INTERVAL" envDefault:"30s"`
+	ListenAddr      string        `env:"REALTIME_LISTEN_ADDR" envDefault:":38020"`
+	RabbitMQURL     string        `env:"REALTIME_RABBITMQ_URL" envDefault:"amqp://stackup:stackup@localhost:38050/"`
+	LogLevel        string        `env:"REALTIME_LOG_LEVEL" envDefault:"info"`
+	QueueName       string        `env:"REALTIME_QUEUE_NAME" envDefault:"q.realtime.session.notify"`
+	JWTSecret       string        `env:"REALTIME_JWT_SECRET" envDefault:"local-development-jwt-secret-must-be-replaced"`
+	CoreBaseURL     string        `env:"REALTIME_CORE_BASE_URL" envDefault:"http://localhost:38010"`
+	InternalApiKey  string        `env:"REALTIME_INTERNAL_API_KEY" envDefault:"local-development-internal-api-key"`
+	AIWSURL         string        `env:"REALTIME_AI_WS_URL" envDefault:"ws://localhost:8000/internal/voice/stream"`
+	WSWriteTimeout  time.Duration `env:"REALTIME_WS_WRITE_TIMEOUT" envDefault:"10s"`
+	SSEPingInterval time.Duration `env:"REALTIME_SSE_PING_INTERVAL" envDefault:"30s"`
+	// SSEWriteTimeout bounds one SSE write. WS already had WSWriteTimeout; SSE had
+	// nothing, so a stalled client could pin its writer goroutine indefinitely and —
+	// through the slow-consumer path — delay every other session (see sse.go).
+	SSEWriteTimeout        time.Duration `env:"REALTIME_SSE_WRITE_TIMEOUT" envDefault:"10s"`
 	SSESlowConsumerTimeout time.Duration `env:"REALTIME_SSE_SLOW_CONSUMER_TIMEOUT" envDefault:"5s"`
 	SSEBufferSize          int           `env:"REALTIME_SSE_BUFFER_SIZE" envDefault:"16"`
 }

realtime/internal/session/registry.go

@@ -58,33 +58,51 @@ func (r *Registry) Unsubscribe(channel Channel, sub *Subscriber) {
 	}
 }
 
-// Dispatch sends ev to all subscribers of channel. Each send waits up to
-// slowTimeout before dropping that subscriber's delivery. Returns the number
-// of subscribers that received the event.
+// Dispatch sends ev to all subscribers of channel, giving up on stalled ones.
+// Returns the number of subscribers that received the event.
+//
+// slowTimeout is a budget for the whole call, not per subscriber. It used to be
+// reset for each one, so K stalled subscribers blocked the caller for K ×
+// slowTimeout. That matters because the AMQP consumer is single-threaded with
+// prefetch=1 (messaging.Consumer): whatever blocks here delays every *other*
+// session's events too, not just the slow one's. Three stalled tabs on one
+// channel could stall the whole fan-out for 15s at the default 5s.
+//
+// With a shared deadline the worst case is one slowTimeout per message no
+// matter how many subscribers are stuck.
 func (r *Registry) Dispatch(channel Channel, ev Event, slowTimeout time.Duration) int {
 	r.mu.RLock()
 	subs := append([]*Subscriber(nil), r.subs[channel]...)
 	r.mu.RUnlock()
 
-	// Reuse a single timer across subscribers. time.After would leak one timer
-	// goroutine per (subscriber × event) until slowTimeout elapsed.
+	if len(subs) == 0 {
+		return 0
+	}
+
+	// One timer for the whole dispatch. time.After would leak a timer goroutine
+	// per (subscriber × event) until it fired.
 	timer := time.NewTimer(slowTimeout)
 	defer timer.Stop()
 
 	delivered := 0
+	budgetLeft := true
 	for _, s := range subs {
-		if !timer.Stop() {
+		if !budgetLeft {
+			// Budget already spent by an earlier stalled subscriber. Still try a
+			// non-blocking send — a healthy subscriber must not lose an event
+			// because someone before it in the list was stuck.
 			select {
-			case <-timer.C:
+			case s.Ch <- ev:
+				delivered++
 			default:
 			}
+			continue
 		}
-		timer.Reset(slowTimeout)
 		select {
 		case s.Ch <- ev:
 			delivered++
 		case <-timer.C:
-			// drop slow consumer's delivery
+			budgetLeft = false
 		}
 	}
 	return delivered

realtime/internal/session/registry_test.go

@@ -111,3 +111,67 @@ func TestRegistryConcurrentSubscribeDispatchUnsubscribe(t *testing.T) {
 		t.Fatalf("expected no subscribers after full unsubscribe, delivered=%d", delivered)
 	}
 }
+
+// AMQP 컨슈머는 단일 스레드 + prefetch=1 이라, 여기서 막히는 시간이 곧 **다른 세션들의
+// 이벤트 지연**이 된다. 예전에는 구독자마다 slowTimeout 을 새로 세어서 멈춘 탭 3개면
+// 5초 × 3 = 15초 동안 전체 fan-out 이 멈췄다.
+func TestDispatchBudgetIsSharedAcrossSubscribers(t *testing.T) {
+	r := NewRegistry()
+	ch := Channel{Kind: ChannelSession, ID: 1}
+
+	// 버퍼 1짜리 구독자 3명을 만들고 전부 가득 채워 둔다(= 멈춘 클라이언트).
+	const stalled = 3
+	for i := 0; i < stalled; i++ {
+		sub := r.Subscribe(ch, 1)
+		sub.Ch <- Event{ID: "fill"}
+	}
+
+	const budget = 150 * time.Millisecond
+	start := time.Now()
+	delivered := r.Dispatch(ch, Event{ID: "x"}, budget)
+	elapsed := time.Since(start)
+
+	if delivered != 0 {
+		t.Fatalf("stalled subscribers should receive nothing, got %d", delivered)
+	}
+	// 예산이 구독자 수만큼 곱해지면 여기서 걸린다.
+	if max := budget * 2; elapsed > max {
+		t.Fatalf("dispatch took %v, expected <= %v (budget must be shared, not per subscriber)",
+			elapsed, max)
+	}
+}
+
+// 예산을 앞선 구독자가 다 써도, 뒤의 건강한 구독자가 그 이유로 이벤트를 잃으면 안 된다.
+func TestHealthySubscriberStillGetsEventAfterStalledOne(t *testing.T) {
+	r := NewRegistry()
+	ch := Channel{Kind: ChannelSession, ID: 1}
+
+	stalledSub := r.Subscribe(ch, 1)
+	stalledSub.Ch <- Event{ID: "fill"}
+	healthy := r.Subscribe(ch, 4)
+
+	delivered := r.Dispatch(ch, Event{ID: "x"}, 50*time.Millisecond)
+
+	if delivered != 1 {
+		t.Fatalf("expected the healthy subscriber to receive it, delivered=%d", delivered)
+	}
+	select {
+	case ev := <-healthy.Ch:
+		if ev.ID != "x" {
+			t.Fatalf("healthy subscriber got %q", ev.ID)
+		}
+	default:
+		t.Fatal("healthy subscriber received nothing")
+	}
+}
+
+func TestDispatchToEmptyChannelDoesNotWait(t *testing.T) {
+	r := NewRegistry()
+
+	start := time.Now()
+	r.Dispatch(Channel{Kind: ChannelSession, ID: 999}, Event{ID: "x"}, time.Second)
+
+	if elapsed := time.Since(start); elapsed > 50*time.Millisecond {
+		t.Fatalf("dispatch to an empty channel took %v", elapsed)
+	}
+}

realtime/internal/transport/router_scope_test.go

@@ -26,7 +26,7 @@ func documentRequest(t *testing.T, id string, claims auth.Claims) *http.Request
 }
 
 func newDocumentHandler() http.HandlerFunc {
-	return scopedChannel(NewSSEHandler(session.NewRegistry(), 4, time.Hour),
+	return scopedChannel(NewSSEHandler(session.NewRegistry(), 4, time.Hour, time.Second),
 		session.ChannelDocument, "DOCUMENT")
 }
 

realtime/internal/transport/sse.go

@@ -15,14 +15,27 @@ type SSEHandler struct {
 	BufferSize      int
 	PingInterval    time.Duration
 	HeartbeatPrefix string
+	// WriteTimeout bounds a single response write. Without it a stalled client
+	// (laptop sleep, dead network) blocks this goroutine until the OS gives up,
+	// which can take minutes — see ServeChannel.
+	WriteTimeout time.Duration
 }
 
-func NewSSEHandler(r *session.Registry, bufferSize int, pingInterval time.Duration) *SSEHandler {
+func NewSSEHandler(
+	r *session.Registry,
+	bufferSize int,
+	pingInterval time.Duration,
+	writeTimeout time.Duration,
+) *SSEHandler {
+	if writeTimeout <= 0 {
+		writeTimeout = 10 * time.Second
+	}
 	return &SSEHandler{
 		Registry:        r,
 		BufferSize:      bufferSize,
 		PingInterval:    pingInterval,
 		HeartbeatPrefix: ": ping ",
+		WriteTimeout:    writeTimeout,
 	}
 }
 
@@ -48,6 +61,23 @@ func (h *SSEHandler) ServeChannel(w http.ResponseWriter, r *http.Request, channe
 	sub := h.Registry.Subscribe(channel, h.BufferSize)
 	defer h.Registry.Unsubscribe(channel, sub)
 
+	// Every write gets a deadline.
+	//
+	// The WS path already had one (WSWriteTimeout); SSE had none. A stalled
+	// client — laptop lid closed mid-interview, network drop — leaves the TCP
+	// connection open while its receive window stays full, so Fprintf blocks
+	// here indefinitely. This goroutine then stops draining sub.Ch, the buffer
+	// fills, and every Dispatch to this channel pays the slow-consumer timeout.
+	// The AMQP consumer is single-threaded, so that stall delays *other*
+	// sessions' events too. Bounding the write lets us give up and unsubscribe.
+	rc := http.NewResponseController(w)
+	deadline := func() bool {
+		return rc.SetWriteDeadline(time.Now().Add(h.WriteTimeout)) == nil
+	}
+	// A server without deadline support (test recorders, some middlewares)
+	// simply keeps the old behaviour rather than failing the stream.
+	supportsDeadline := deadline()
+
 	ctx := r.Context()
 	ticker := time.NewTicker(h.PingInterval)
 	defer ticker.Stop()
@@ -58,15 +88,25 @@ func (h *SSEHandler) ServeChannel(w http.ResponseWriter, r *http.Request, channe
 			slog.Info("sse.unsubscribe", "channel_kind", channel.Kind, "channel_id", channel.ID, "reason", "client_close")
 			return
 		case <-ticker.C:
+			if supportsDeadline {
+				deadline()
+			}
 			if _, err := fmt.Fprintf(w, "%s%d\n\n", h.HeartbeatPrefix, time.Now().Unix()); err != nil {
+				slog.Info("sse.unsubscribe", "channel_kind", channel.Kind,
+					"channel_id", channel.ID, "reason", "ping_write_failed", "err", err)
 				return
 			}
 			flusher.Flush()
 		case ev, ok := <-sub.Ch:
 			if !ok {
 				return
 			}
+			if supportsDeadline {
+				deadline()
+			}
 			if _, err := writeSSE(w, ev); err != nil {
+				slog.Info("sse.unsubscribe", "channel_kind", channel.Kind,
+					"channel_id", channel.ID, "reason", "write_failed", "err", err)
 				return
 			}
 			flusher.Flush()

realtime/internal/transport/sse_test.go

@@ -12,7 +12,7 @@ import (
 
 func TestSSEDocumentChannelReceivesEvent(t *testing.T) {
 	reg := session.NewRegistry()
-	h := NewSSEHandler(reg, 4, time.Hour)
+	h := NewSSEHandler(reg, 4, time.Hour, time.Second)
 
 	req := httptest.NewRequest("GET", "/realtime/stream/documents/101", nil)
 	ctx, cancel := context.WithCancel(req.Context())