diff --git a/docker-compose.yml b/docker-compose.yml index 56450ac6..9911fed5 100644 --- a/docker-compose.yml +++ b/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 diff --git a/realtime/.env.example b/realtime/.env.example index 401c953a..a9102676 100644 --- a/realtime/.env.example +++ b/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 diff --git a/realtime/CLAUDE.md b/realtime/CLAUDE.md index 70981c91..008d9bf9 100644 --- a/realtime/CLAUDE.md +++ b/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 타임아웃. 없으면 멈춘 클라이언트가 쓰기 고루틴을 무기한 붙잡는다 | --- diff --git a/realtime/cmd/realtime/main.go b/realtime/cmd/realtime/main.go index 1eecbd0a..2beabe3d 100644 --- a/realtime/cmd/realtime/main.go +++ b/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) diff --git a/realtime/internal/config/config.go b/realtime/internal/config/config.go index fa15ac8c..5174262d 100644 --- a/realtime/internal/config/config.go +++ b/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"` } diff --git a/realtime/internal/session/registry.go b/realtime/internal/session/registry.go index 749805a1..6b591d58 100644 --- a/realtime/internal/session/registry.go +++ b/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 diff --git a/realtime/internal/session/registry_test.go b/realtime/internal/session/registry_test.go index 48c61f43..6be393b4 100644 --- a/realtime/internal/session/registry_test.go +++ b/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) + } +} diff --git a/realtime/internal/transport/router_scope_test.go b/realtime/internal/transport/router_scope_test.go index cf314c17..8940ba5a 100644 --- a/realtime/internal/transport/router_scope_test.go +++ b/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") } diff --git a/realtime/internal/transport/sse.go b/realtime/internal/transport/sse.go index 7a0c2bcd..3d568755 100644 --- a/realtime/internal/transport/sse.go +++ b/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,7 +88,12 @@ 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() @@ -66,7 +101,12 @@ func (h *SSEHandler) ServeChannel(w http.ResponseWriter, r *http.Request, channe 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() diff --git a/realtime/internal/transport/sse_test.go b/realtime/internal/transport/sse_test.go index db3fa31b..0bc5786c 100644 --- a/realtime/internal/transport/sse_test.go +++ b/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())