SSE 이벤트 통합 replay - Redis Stream 기반 Last-Event-ID 복구#756
Closed
m-a-king wants to merge 3 commits into
Closed
Conversation
- 유저별 Redis Stream(sse:events:{userId})을 SSE 이벤트 로그로 두고 notification·silent-sync 를 emit 시점에 적재한다. stream entry id 를 SSE id 필드에 실어, 재연결 시 Last-Event-ID 초과분을 종류 구분 없이 원본 그대로 replay 한다
- 1차 설계(알림 PK 앵커, PR #751 닫음)는 비영속 silent-sync 가 복구 대상에서 빠져 이 작업의 동인(아이템 등록 화면 파싱 동기화 유실)을 못 닫았다. SSE 의 lastEventId 는 스트림당 스칼라 하나라 두 id 공간을 섞을 수 없어 통합 이벤트 로그로 전환
- 저장소를 인메모리가 아닌 Redis 에 둔 이유: blue-green 배포 전환이 곧 전 연결 동시 재연결 시점인데 인메모리 버퍼는 그 순간 비어 있고, 스케일아웃 시 유저별 단조 id 채번도 중앙 저장소가 필요하다
- payload 를 한 번 직렬화(JSON 문자열)해 live 전송·로그·replay 가 같은 바이트를 공유한다. Redis 객체 저장 규약대로 직렬화 호환성 테스트(payload 스냅샷 + 스트림 필드 키 고정)를 함께 둔다
- 상한 초과 공백은 replay 통째 생략(부분 replay 는 뒤에 조용한 구멍을 남긴다). MAX_LEN(200) > REPLAY_LIMIT(100) 관계로 trim 이 만든 구멍을 연속인 척 replay 하는 경우를 배제하고, 관계 자체를 테스트로 고정
- 적재 실패는 id 없이 live 전송만 하는 degrade — id 없는 이벤트는 클라 lastEventId 를 갱신하지 않아 복구 기준점을 오염시키지 않는다
- 스냅샷 실측 중 발견: createdAt 은 JacksonConfig 의 KST 변환(+09:00 오프셋)으로 나가는데 스펙 문서는 "오프셋 없음"으로 낡아 있었다. 실측대로 정정
- 파일별로 중복이던 recording emitter 4개를 support/RecordingSseEmitter 로 통합(#578 의 공유 test double 항목). 와이어 JSON 트리 단언으로 바뀌어 직렬화 회귀도 함께 잡는다
|
Discord 스레드 연동용 메타데이터입니다. discord-pr-bot 워크플로가 자동 생성하며, 수정·삭제하면 PR 과 Discord 알림 연동이 끊깁니다. |
|
No actionable comments were generated in the recent review. 🎉 ℹ️ Recent review info⚙️ Run configurationConfiguration used: Path: .coderabbit.yml Review profile: CHILL Plan: Pro Plus Run ID: 📒 Files selected for processing (2)
🚧 Files skipped from review as they are similar to previous changes (2)
WalkthroughSSE notification과 silent-sync 이벤트에 Redis Stream 로그와 ChangesSSE 재연결 replay
Estimated code review effort: 4 (Complex) | ~45 minutes Sequence Diagram(s)sequenceDiagram
participant Client
participant NotificationSseController
participant SseReconnectReplayer
participant RedisSseEventLog
participant LocalSseDelivery
Client->>NotificationSseController: Last-Event-ID 헤더로 재연결
NotificationSseController->>SseReconnectReplayer: 누락 이벤트 replay 요청
SseReconnectReplayer->>RedisSseEventLog: 기준 id 이후 이벤트 조회
RedisSseEventLog-->>SseReconnectReplayer: 이벤트 레코드 반환
SseReconnectReplayer->>LocalSseDelivery: 이벤트 재전송
LocalSseDelivery-->>Client: id/name/payload 순서대로 전달
Assessment against linked issues
🚥 Pre-merge checks | ✅ 1 | ❌ 1❌ Failed checks (1 warning)
✅ Passed checks (1 passed)
✨ Finishing Touches📝 Generate docstrings
🧪 Generate unit tests (beta)
Comment |
![coderabbitai[bot]](https://avatars.githubusercontent.com/in/347564?s=60&v=4){kind=small}
There was a problem hiding this comment.
Actionable comments posted: 2
Caution
Some comments are outside the diff and can’t be posted inline due to platform limitations.
⚠️ Outside diff range comments (1)
src/main/kotlin/com/depromeet/piki/notification/controller/NotificationSseController.kt (1)
36-52: 🗄️ Data Integrity & Integration | 🟠 Major | 🏗️ Heavy lift재연결 전환 구간에서 live/replay 순서가 원자적으로 보장되지 않습니다. 최신 라이브 이벤트 뒤에 과거 replay가 적용되면 화면 상태와 Last-Event-ID 커서가 함께 역행할 수 있습니다.
src/main/kotlin/com/depromeet/piki/notification/controller/NotificationSseController.kt#L36-L52: replay 완료 전 라이브 이벤트를 버퍼링하고 replay 이후 ID 순서로 flush하도록 전환 과정을 직렬화하세요.src/test/kotlin/com/depromeet/piki/notification/sse/NotificationSseIntegrationTest.kt#L329-L370: replay 도중 라이브 발행을 끼워 넣어 ID 순서와 최종 최신 상태를 검증하세요.notification-sse-spec.md#L320-L323: Last-Event-ID는 현재 값보다 큰 Stream ID일 때만 저장하도록 예시를 수정하세요.경로 지침의 동시성·데이터 정합성 우선 원칙과 PR의 발생 순서 보장 계약에 따른 검토입니다.
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@src/main/kotlin/com/depromeet/piki/notification/controller/NotificationSseController.kt` around lines 36 - 52, Update NotificationSseController’s reconnect flow so live events are buffered while replayer.replayMissed completes, then flushed in Stream ID order before normal live delivery resumes; preserve registration before replay and unregister safely on failure. In src/test/kotlin/com/depromeet/piki/notification/sse/NotificationSseIntegrationTest.kt lines 329-370, add coverage that publishes during replay and verifies ordered IDs and the final latest state. In notification-sse-spec.md lines 320-323, revise the Last-Event-ID example to persist only IDs greater than the current stored Stream ID.Source: Path instructions
🤖 Prompt for all review comments with AI agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
Inline comments:
In `@src/main/kotlin/com/depromeet/piki/notification/sse/LocalSseDelivery.kt`:
- Around line 63-76: Update the SSE delivery flow around subscribe(),
replayTo(), deliver(), and deliverSilentSync() so each user’s register → replay
→ live fan-out sequence is serialized through a shared per-user lock or queue.
Ensure live sends cannot interleave with replay on the same emitter, while
preserving existing eviction and delivery behavior.
In `@src/main/kotlin/com/depromeet/piki/notification/sse/RedisSseEventLog.kt`:
- Around line 38-48: RedisSseEventLog의 이벤트 적재 흐름을 수정해 XADD, XTRIM, EXPIRE를 하나의
Lua 스크립트에서 원자적으로 수행하고 생성된 record ID를 반환하도록 하세요. XADD 이후 housekeeping 단계의 실패를 전체
적재 실패로 처리해 null을 반환하지 않도록 하며, live 이벤트 전송에 실제 Stream ID가 항상 사용되게 하세요.
---
Outside diff comments:
In
`@src/main/kotlin/com/depromeet/piki/notification/controller/NotificationSseController.kt`:
- Around line 36-52: Update NotificationSseController’s reconnect flow so live
events are buffered while replayer.replayMissed completes, then flushed in
Stream ID order before normal live delivery resumes; preserve registration
before replay and unregister safely on failure. In
src/test/kotlin/com/depromeet/piki/notification/sse/NotificationSseIntegrationTest.kt
lines 329-370, add coverage that publishes during replay and verifies ordered
IDs and the final latest state. In notification-sse-spec.md lines 320-323,
revise the Last-Event-ID example to persist only IDs greater than the current
stored Stream ID.
🪄 Autofix (Beta)
Fix all unresolved CodeRabbit comments on this PR:
- Push a commit to this branch (recommended)
- Create a new PR with the fixes
ℹ️ Review info
⚙️ Run configuration
Configuration used: Path: .coderabbit.yml
Review profile: CHILL
Plan: Pro Plus
Run ID: 6838978d-846c-44b2-afa6-e99f8ef4a27c
📒 Files selected for processing (21)
notification-sse-spec.mdsrc/main/kotlin/com/depromeet/piki/notification/controller/NotificationSseApi.ktsrc/main/kotlin/com/depromeet/piki/notification/controller/NotificationSseController.ktsrc/main/kotlin/com/depromeet/piki/notification/sse/LocalSseDelivery.ktsrc/main/kotlin/com/depromeet/piki/notification/sse/RedisSseEventLog.ktsrc/main/kotlin/com/depromeet/piki/notification/sse/SseEventLog.ktsrc/main/kotlin/com/depromeet/piki/notification/sse/SseReconnectReplayer.ktsrc/test/kotlin/com/depromeet/piki/notification/service/NotificationBadgeSyncAsyncIntegrationTest.ktsrc/test/kotlin/com/depromeet/piki/notification/service/NotificationCleanupSchedulerIntegrationTest.ktsrc/test/kotlin/com/depromeet/piki/notification/sse/NotificationSseIntegrationTest.ktsrc/test/kotlin/com/depromeet/piki/notification/sse/RedisSseEventLogIntegrationTest.ktsrc/test/kotlin/com/depromeet/piki/notification/sse/SsePayloadSerializationCompatibilityTest.ktsrc/test/kotlin/com/depromeet/piki/notification/sse/TournamentItemParsedSseIntegrationTest.ktsrc/test/kotlin/com/depromeet/piki/support/IntegrationStubs.ktsrc/test/kotlin/com/depromeet/piki/support/RecordingSseEmitter.ktsrc/test/kotlin/com/depromeet/piki/support/StubSseEventLog.ktsrc/test/resources/sse-payload/notification-reference.jsonsrc/test/resources/sse-payload/notification-tournament-routed.jsonsrc/test/resources/sse-payload/notification-wish-parsing.jsonsrc/test/resources/sse-payload/silent-sync-tournament-item-parsed.jsonsrc/test/resources/sse-payload/silent-sync-unread-count-changed.json
- XADD 성공 뒤 trim·EXPIRE 만 실패해도 null 을 돌려줘, 로그엔 남았는데 live 는 id 없이 나가는 불일치가 있었다 (이후 더 큰 id 를 커서로 잡은 클라이언트는 그 레코드를 replay 로도 못 받는다) — CodeRabbit 지적 수용 - 적재(XADD) 실패만 degrade(null) 로 처리하고, 정리(trim·TTL)는 별도 격리해 실패해도 id 를 반환한다. trim·TTL 은 멱등 정리라 다음 적재가 자연 재시도한다 - 제안된 Lua 원자화는 반영하지 않았다 — id 만 살리면 정리의 원자성은 정확성에 불필요하고 스크립트 유지 비용만 남는다 - 스펙 APP 예시의 커서 저장을 "지금보다 큰 id 일 때만" 으로 보강 — 순서 어긋난 도착에도 커서가 역행하지 않아 다음 재연결의 중복 replay 가 준다
Collaborator
Author
|
CodeRabbit review body 의 Outside diff range 코멘트(
|
|
Tip For best results, initiate chat on the files or code changes.
✏️ Learnings added
|
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
1 participant
Add this suggestion to a batch that can be applied as a single commit.This suggestion is invalid because no changes were made to the code.Suggestions cannot be applied while the pull request is closed.Suggestions cannot be applied while viewing a subset of changes.Only one suggestion per line can be applied in a batch.Add this suggestion to a batch that can be applied as a single commit.Applying suggestions on deleted lines is not supported.You must change the existing code in this line in order to create a valid suggestion.Outdated suggestions cannot be applied.This suggestion has been applied or marked resolved.Suggestions cannot be applied from pending reviews.Suggestions cannot be applied on multi-line comments.Suggestions cannot be applied while the pull request is queued to merge.Suggestion cannot be applied right now. Please check back later.
Situation
silent-sync의TOURNAMENT_ITEM_PARSED)가 SSE 단독 채널·비영속이라, 놓치면 카드가 낡은 채 남는다.notification만 replay 했다. 비영속 silent-sync 가 복구 대상에서 빠져 정작 동인을 못 닫았고, "스트림 절반만 신뢰 가능"한 비대칭과 FE 재조회 계약 병행 부담이 남았다. SSE 의 lastEventId 는 스트림당 스칼라 하나라 알림 PK 와 별도 id 공간을 섞을 수도 없다.Task
notification·silent-sync)를 균일하게 복구하는 통합 이벤트 로그를 설계·구현한다. FE 는 "놓친 이벤트는 다시 온다, 이벤트 id 로 dedup" 규칙 하나만 가지면 된다.Action
저장소 결정
동작 계약
sse:events:{userId})에 XADD 하고, stream entry id 를 SSEid필드에 싣는다. 연결이 없어도 적재된다(이후 재연결에서 복구).subscribe가Last-Event-ID헤더(형식 검증, 아니면 첫 연결 취급)를 받아 초과분을 발생 순서대로 그 연결에만 재전송한다. 등록 후 조회 순서로 "유실 대신 중복"을 택하고 클라가 이벤트 id 로 dedup 한다.구현
SseEventLog(인터페이스) ·RedisSseEventLog(XADD + 정확 trim + TTL 24h) ·SseReconnectReplayer.LocalSseDelivery가 직렬화·적재·id 부여를 한 곳(loggedEvent)에 모으고, replay 도 같은 write 경로(sendOrEvict)를 쓴다.notification-sse-spec.md(id 필드·replay 계약·dedup·상한·TTL 창·WEB 자동/APP 저장 예시)와NotificationSseApiOpenAPI 반영.검증
RedisSseEventLogIntegrationTest): XADD id 단조·exclusive 조회·limit·정확 trim·TTL, MAX_LEN > REPLAY_LIMIT 관계 고정.support/RecordingSseEmitter로 통합(SSE 파싱 동기화: 이벤트 배선(AFTER_COMMIT+@Async) 통합 테스트 + RecordingSseEmitter 공유 #578 의 공유 test double 항목). payload 단언이 와이어 JSON 트리 기준으로 바뀌어 직렬화 회귀도 함께 잡는다.Result
TOURNAMENT_ITEM_ADDED)와 끝내는 신호(TOURNAMENT_ITEM_PARSED)가 모두 재연결 replay 로 복구된다. 웹(EventSource)은 클라 변경 없이 동작하고, 앱은 마지막 이벤트 id 저장·전송만 추가하면 된다.createdAt은JacksonConfig의 KST 변환으로+09:00오프셋 포함 형식으로 나가는데, 스펙 문서가 "오프셋 없음"으로 낡아 있었다.연관 이슈
Summary by CodeRabbit
Last-Event-ID를 기준으로 누락된notification과silent-sync를 발생 순서대로 재전송합니다(이벤트 ID 포함).createdAt형식 및 재생 대상 판정 기준을 명확히 했습니다.