diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index d2b5d832..556d58ac 100644 --- a/.github/workflows/ci.yml +++ b/.github/workflows/ci.yml @@ -125,6 +125,9 @@ jobs: working-directory: ./realtime run: go vet ./... + # -race 로 돌린다. RealTime 은 본업이 동시 fan-out(AMQP 컨슈머가 Dispatch 하는 동안 + # HTTP 핸들러들이 Subscribe/Unsubscribe)이라, 경합은 여기서 못 잡으면 운영에서 + # 간헐적 패닉·유실로만 드러난다. realtime/CLAUDE.md §11 도 -race 를 규정한다. - name: Test working-directory: ./realtime - run: go test ./... + run: go test -race ./... diff --git a/realtime/Makefile b/realtime/Makefile index fc26360a..8d96cdbc 100644 --- a/realtime/Makefile +++ b/realtime/Makefile @@ -1,4 +1,4 @@ -.PHONY: build run test fmt lint tidy clean +.PHONY: build run test test-race fmt lint tidy clean GO ?= go PKG ?= ./... @@ -12,6 +12,10 @@ run: test: $(GO) test $(PKG) +# CI 가 도는 것과 같은 형태. 동시성 변경을 했다면 push 전에 이걸로 확인한다. +test-race: + $(GO) test -race $(PKG) + fmt: $(GO) fmt $(PKG) diff --git a/realtime/internal/session/registry_test.go b/realtime/internal/session/registry_test.go index ed0f12b8..48c61f43 100644 --- a/realtime/internal/session/registry_test.go +++ b/realtime/internal/session/registry_test.go @@ -1,6 +1,7 @@ package session import ( + "sync" "testing" "time" ) @@ -44,3 +45,69 @@ func TestUnsubscribeRemovesChannelEntry(t *testing.T) { t.Errorf("delivered = %d, want 0 after unsubscribe", n) } } + +// Registry 는 RealTime 서버 전체의 공유 가변 상태다. 운영에서는 AMQP 컨슈머 고루틴이 +// Dispatch 하는 동안 HTTP 핸들러들이 Subscribe/Unsubscribe 한다 — 그런데 이 조합을 +// 검증하는 테스트가 없었다. +// +// Dispatch 는 의도적으로 락을 놓은 뒤 채널에 쓴다(느린 구독자가 락을 잡고 있으면 다른 +// 구독자까지 막히므로). 그 설계 때문에 "복사한 구독자 목록"과 "지금 살아있는 구독자"가 +// 어긋나는 창이 생기고, 여기서 어긋남이 자료 경합이 되지 않는지 확인한다. +// +// `-race` 와 함께 돌 때 의미가 있다 (CI 의 `go test -race ./...`). +func TestRegistryConcurrentSubscribeDispatchUnsubscribe(t *testing.T) { + r := NewRegistry() + target := ch(ChannelSession, 1) + + const ( + churnGoroutines = 8 + churnIterations = 200 + dispatchers = 4 + ) + + stop := make(chan struct{}) + var dispatchWG, churnWG sync.WaitGroup + + for i := 0; i < dispatchers; i++ { + dispatchWG.Add(1) + go func() { + defer dispatchWG.Done() + ev := Event{ID: "1", Type: "SESSION_MESSAGE", Data: []byte(`{"a":1}`)} + for { + select { + case <-stop: + return + default: + } + // 느린 구독자 타임아웃은 짧게 — 버퍼가 찬 구독자 때문에 테스트가 늘어지지 않게. + r.Dispatch(target, ev, time.Millisecond) + } + }() + } + + for i := 0; i < churnGoroutines; i++ { + churnWG.Add(1) + go func() { + defer churnWG.Done() + for j := 0; j < churnIterations; j++ { + sub := r.Subscribe(target, 2) + // 한 건 정도 읽어 Dispatch 가 항상 타임아웃으로만 끝나지 않게 한다. + select { + case <-sub.Ch: + default: + } + r.Unsubscribe(target, sub) + } + }() + } + + churnWG.Wait() + close(stop) + dispatchWG.Wait() + + // 모든 구독이 해제됐으면 전달 대상이 남아 있으면 안 된다. id 기반 제거가 churn 중 + // 어긋나면 해제된 구독자가 목록에 남아 여기서 0 이 아니게 된다. + if delivered := r.Dispatch(target, Event{ID: "x"}, time.Millisecond); delivered != 0 { + t.Fatalf("expected no subscribers after full unsubscribe, delivered=%d", delivered) + } +}