Skip to content

Commit f10909e

Browse files
i3monthsi3months
authored
Merge pull request #126 from Team-StackUp/feature/sweeper-concurrency-hardening
fix(session): 세션 종료 동시성 하드닝 (#124 후속)
2 parents 301822f + a7f4fc6 commit f10909e

10 files changed

Lines changed: 117 additions & 38 deletions

File tree

‎backend/CLAUDE.md‎

Lines changed: 8 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -362,8 +362,15 @@ docker compose up -d
362362
- **세션 시간초과 자동 종료 본 구현**: `@EnableScheduling`(`common/config/SchedulingConfig`) + `SessionTimeoutSweeper`
363363
(`@Scheduled` 기본 5분 주기)가 `maxDurationMinutes` 초과한 IN_PROGRESS 세션을 찾아 `SessionTimeoutService.endTimedOut`
364364
호출 — 답변 있으면 COMPLETED(→`SessionEndedEvent(DURATION_EXCEEDED)`→피드백), 없으면 INTERRUPTED(피드백 없음).
365-
좀비 세션(자기소개 미답변·STT 실패·탭 종료) 방지. 멀티 인스턴스 중복 실행도 IN_PROGRESS 재확인+피드백 멱등으로 안전.
365+
좀비 세션(자기소개 미답변·STT 실패·탭 종료) 방지.
366366
주기: `interview.session.sweep-interval-ms`(기본 300000)·`sweep-initial-delay-ms`(기본 60000).
367+
- **동시 종료 안전(원자적 전이)**: 모든 종료 경로(스위퍼·수동 `SessionService.end`·콜백 `endSession`)는
368+
`InterviewSessionRepository.finishIfInProgress`(조건부 UPDATE `WHERE status=IN_PROGRESS`)로 전이를
369+
차지하고, **영향 행 1인 트랜잭션만** `SessionEndedEvent` 를 발행한다 → DB 행 락으로 직렬화돼 동시
370+
종료 시에도 `generate.feedback` 가 중복 발행되지 않는다.
371+
- **종료 후 콜백 드롭**: `QuestionsCallbackService.apply` 와 `SessionFollowupRequester.onAnswerSubmitted` 는
372+
세션이 `isTerminal()` 이면 처리를 건너뛴다 → 자동종료 뒤 늦게 도착한 POOL/FOLLOWUP 콜백이나
373+
막판 답변 발화가 종료 세션에 질문·placeholder 를 추가하는 사후 변조를 차단.
367374
- **질문별 복기 본 구현**: V19 로 `interview_messages` 에 `model_answer`/`answer_rewrite`/`coaching_comment`
368375
추가. AI 가 `callback.feedback.answerCoaching[{messageId,…}]` 로 답변별 모범 답안·리라이트·코칭을 보내면
369376
`FeedbackCallbackService` 가 각 메시지에 `recordCoaching` 기록. `MessageResult`/`MessageResponse` 가 답변

‎backend/src/main/java/com/stackup/stackup/session/application/QuestionsCallbackService.java‎

Lines changed: 23 additions & 10 deletions
Original file line numberDiff line numberDiff line change
@@ -18,6 +18,7 @@
1818
import com.stackup.stackup.session.domain.SessionQuestionPool;
1919
import com.stackup.stackup.session.domain.SessionQuestionPoolRepository;
2020
import com.stackup.stackup.session.domain.SessionStatus;
21+
import java.time.Instant;
2122
import java.util.List;
2223
import lombok.RequiredArgsConstructor;
2324
import org.slf4j.Logger;
@@ -68,6 +69,14 @@ public void apply(QuestionsCallbackEnvelope envelope) {
6869
markProcessed(envelope.messageId());
6970
return;
7071
}
72+
// 종료된 세션에 늦게 도착한 콜백은 드롭(멱등 마킹). 스위퍼 자동종료·수동종료 후
73+
// 처리 대기 중이던 POOL/FOLLOWUP 콜백이 종료 세션을 사후 변조하는 것을 방지.
74+
if (session.getStatus().isTerminal()) {
75+
log.info("callback.questions for terminal session, drop. id={}, status={}, messageId={}",
76+
sessionId, session.getStatus(), envelope.messageId());
77+
markProcessed(envelope.messageId());
78+
return;
79+
}
7180

7281
if (payload.isPool()) {
7382
applyInitialQuestion(session, payload);
@@ -163,18 +172,22 @@ private void publishQuestionEvents(InterviewSession session, InterviewMessage me
163172
}
164173

165174
private void endSession(InterviewSession session, String reason) {
166-
try {
167-
session.end();
168-
events.publishEvent(RealtimeNotifyEvent.session(session.getId(), SseEventType.SESSION_STATE,
169-
new SessionStateNotice(session.getId(), session.getStatus().name(), reason)));
170-
events.publishEvent(RealtimeNotifyEvent.user(session.getUser().getId(), SseEventType.SESSION_STATE,
171-
new SessionStateNotice(session.getId(), session.getStatus().name(), reason)));
172-
events.publishEvent(new SessionEndedEvent(session.getUser().getId(), session.getId(), reason));
173-
log.info("session auto-completed. sessionId={}, reason={}", session.getId(), reason);
174-
} catch (IllegalStateException e) {
175-
log.warn("auto-end skipped — session not IN_PROGRESS. sessionId={}, status={}",
175+
// 원자적 종료 전이: IN_PROGRESS 일 때만 1행 갱신. 0이면 다른 트랜잭션(스위퍼·수동 종료)이
176+
// 먼저 종료한 것 → 종료 부수효과 미발행(중복 피드백 방지).
177+
int claimed = sessionRepository.finishIfInProgress(
178+
session.getId(), SessionStatus.COMPLETED, Instant.now());
179+
if (claimed == 0) {
180+
log.warn("auto-end skipped — session already ended. sessionId={}, status={}",
176181
session.getId(), session.getStatus());
182+
return;
177183
}
184+
session.end(); // 인메모리 동기화(SSE status 표기용). DB는 위 조건부 UPDATE 로 이미 COMPLETED.
185+
events.publishEvent(RealtimeNotifyEvent.session(session.getId(), SseEventType.SESSION_STATE,
186+
new SessionStateNotice(session.getId(), session.getStatus().name(), reason)));
187+
events.publishEvent(RealtimeNotifyEvent.user(session.getUser().getId(), SseEventType.SESSION_STATE,
188+
new SessionStateNotice(session.getId(), session.getStatus().name(), reason)));
189+
events.publishEvent(new SessionEndedEvent(session.getUser().getId(), session.getId(), reason));
190+
log.info("session auto-completed. sessionId={}, reason={}", session.getId(), reason);
178191
}
179192

180193
private void applyFollowup(InterviewSession session, QuestionsCallbackPayload payload) {

‎backend/src/main/java/com/stackup/stackup/session/application/SessionFollowupRequester.java‎

Lines changed: 10 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -14,6 +14,7 @@
1414
import com.stackup.stackup.session.domain.InterviewSession;
1515
import com.stackup.stackup.session.domain.MessageRole;
1616
import com.stackup.stackup.session.domain.SessionContextRepository;
17+
import com.stackup.stackup.session.domain.SessionStatus;
1718
import java.util.ArrayList;
1819
import java.util.List;
1920
import lombok.RequiredArgsConstructor;
@@ -53,6 +54,15 @@ public void onAnswerSubmitted(AnswerSubmittedEvent event) {
5354
}
5455
InterviewSession session = parent.getSession();
5556

57+
// 종료된 세션이면 꼬리질문/풀 생성을 요청하지 않는다. maxDuration 직전 제출한 답변의
58+
// AFTER_COMMIT 발화가 스위퍼 자동종료와 겹칠 때, 종료 세션에 placeholder·질문이
59+
// 추가되거나 불필요한 AI 호출이 발생하는 것을 방지(콜백단 terminal 가드와 같은 선상).
60+
if (session.getStatus().isTerminal()) {
61+
log.info("generate.followup skipped — session already terminal. sessionId={}, status={}",
62+
session.getId(), session.getStatus());
63+
return;
64+
}
65+
5666
// RAG(자료 근거/correctness) 용 세션 컨텍스트 문서 — 피드백 발행부와 동일 패턴.
5767
List<Long> contextDocumentIds = contextRepository.findBySession_Id(session.getId()).stream()
5868
.map(c -> c.getDocument().getId())

‎backend/src/main/java/com/stackup/stackup/session/application/SessionService.java‎

Lines changed: 8 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -16,8 +16,10 @@
1616
import com.stackup.stackup.session.domain.SessionMode;
1717
import com.stackup.stackup.session.domain.SessionContext;
1818
import com.stackup.stackup.session.domain.SessionContextRepository;
19+
import com.stackup.stackup.session.domain.SessionStatus;
1920
import com.stackup.stackup.user.domain.User;
2021
import com.stackup.stackup.user.domain.UserRepository;
22+
import java.time.Instant;
2123
import java.util.ArrayList;
2224
import java.util.LinkedHashSet;
2325
import java.util.List;
@@ -108,11 +110,14 @@ public SessionResult start(Long userId, Long sessionId) {
108110
@Transactional
109111
public SessionResult end(Long userId, Long sessionId) {
110112
InterviewSession session = loadOwned(userId, sessionId);
111-
try {
112-
session.end();
113-
} catch (IllegalStateException e) {
113+
// 원자적 종료 전이: IN_PROGRESS 일 때만 1행 갱신. 0이면 다른 트랜잭션(스위퍼 등)이
114+
// 먼저 종료한 것 → 기존 동작과 동일하게 INVALID_STATE. 1을 받은 호출자만 종료 이벤트 발행.
115+
int claimed = sessionRepository.finishIfInProgress(
116+
sessionId, SessionStatus.COMPLETED, Instant.now());
117+
if (claimed == 0) {
114118
throw new DomainException(ApiErrorCode.SESSION_INVALID_STATE);
115119
}
120+
session.end(); // 응답·인메모리 동기화(DB는 위 조건부 UPDATE 로 이미 COMPLETED).
116121
events.publishEvent(new SessionEndedEvent(userId, sessionId, "USER_REQUEST"));
117122
return SessionResult.of(session, contextDocumentIds(sessionId));
118123
}

‎backend/src/main/java/com/stackup/stackup/session/application/SessionTimeoutService.java‎

Lines changed: 21 additions & 18 deletions
Original file line numberDiff line numberDiff line change
@@ -8,6 +8,7 @@
88
import com.stackup.stackup.session.domain.InterviewSessionRepository;
99
import com.stackup.stackup.session.domain.MessageRole;
1010
import com.stackup.stackup.session.domain.SessionStatus;
11+
import java.time.Instant;
1112
import lombok.RequiredArgsConstructor;
1213
import org.slf4j.Logger;
1314
import org.slf4j.LoggerFactory;
@@ -38,28 +39,30 @@ public void endTimedOut(Long sessionId) {
3839
boolean hasAnswer =
3940
messageRepository.existsBySession_IdAndRole(sessionId, MessageRole.INTERVIEWEE);
4041
Long userId = session.getUser().getId();
41-
try {
42-
if (hasAnswer) {
43-
session.end(); // COMPLETED → SessionEndedEvent 로 피드백 생성
44-
publishState(session, userId);
45-
events.publishEvent(new SessionEndedEvent(userId, sessionId, REASON));
46-
log.info("session auto-completed (timeout). sessionId={}", sessionId);
47-
} else {
48-
session.interrupt(); // 답변 없음 → 피드백 없이 정리
49-
publishState(session, userId);
50-
log.info("session auto-interrupted (timeout, no answers). sessionId={}", sessionId);
51-
}
52-
} catch (IllegalStateException e) {
53-
log.warn("auto-end skipped — session not IN_PROGRESS. sessionId={}, status={}",
54-
sessionId, session.getStatus());
42+
SessionStatus target = hasAnswer ? SessionStatus.COMPLETED : SessionStatus.INTERRUPTED;
43+
44+
// 원자적 종료 전이: IN_PROGRESS 일 때만 1행이 갱신된다. 0이면 다른 트랜잭션
45+
// (다른 스위퍼 인스턴스·수동 종료·콜백 종료)이 먼저 종료한 것 → 부수효과 미발행.
46+
int claimed = sessionRepository.finishIfInProgress(sessionId, target, Instant.now());
47+
if (claimed == 0) {
48+
log.info("session auto-end skipped — already ended by another tx. sessionId={}", sessionId);
49+
return;
50+
}
51+
52+
publishState(sessionId, userId, target);
53+
if (target == SessionStatus.COMPLETED) {
54+
// COMPLETED → SessionEndedEvent 로 피드백 생성. 전이를 차지한 트랜잭션에서 단 한 번만.
55+
events.publishEvent(new SessionEndedEvent(userId, sessionId, REASON));
56+
log.info("session auto-completed (timeout). sessionId={}", sessionId);
57+
} else {
58+
log.info("session auto-interrupted (timeout, no answers). sessionId={}", sessionId);
5559
}
5660
}
5761

58-
private void publishState(InterviewSession session, Long userId) {
59-
SessionStateNotice notice =
60-
new SessionStateNotice(session.getId(), session.getStatus().name(), REASON);
62+
private void publishState(Long sessionId, Long userId, SessionStatus status) {
63+
SessionStateNotice notice = new SessionStateNotice(sessionId, status.name(), REASON);
6164
events.publishEvent(RealtimeNotifyEvent.session(
62-
session.getId(), SseEventType.SESSION_STATE, notice));
65+
sessionId, SseEventType.SESSION_STATE, notice));
6366
events.publishEvent(RealtimeNotifyEvent.user(
6467
userId, SseEventType.SESSION_STATE, notice));
6568
}

‎backend/src/main/java/com/stackup/stackup/session/domain/InterviewSessionRepository.java‎

Lines changed: 13 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -1,10 +1,12 @@
11
package com.stackup.stackup.session.domain;
22

3+
import java.time.Instant;
34
import java.util.List;
45
import java.util.Optional;
56
import org.springframework.data.domain.Page;
67
import org.springframework.data.domain.Pageable;
78
import org.springframework.data.jpa.repository.JpaRepository;
9+
import org.springframework.data.jpa.repository.Modifying;
810
import org.springframework.data.jpa.repository.Query;
911
import org.springframework.data.repository.query.Param;
1012

@@ -36,4 +38,15 @@ List<Long> findRecentSessionIds(@Param("userId") Long userId,
3638
List<InterviewSession> findByStatusAndDeletedFalse(SessionStatus status);
3739

3840
List<InterviewSession> findByUser_IdAndStatusOrderByEndedAtDesc(Long userId, SessionStatus status, Pageable pageable);
41+
42+
// 원자적 종료 전이: IN_PROGRESS 일 때만 단일 조건부 UPDATE 로 종료 상태로 바꾼다.
43+
// 영향 행 수(0 또는 1)를 반환 — 1을 받은 호출자만 '종료를 차지(claim)'했으므로
44+
// SessionEndedEvent 등 종료 부수효과를 단 한 번만 발행한다. 동시 종료(스위퍼·수동·콜백)
45+
// 시 DB 행 락으로 직렬화돼 정확히 하나만 1을 받는다(중복 피드백 발행 방지).
46+
@Modifying
47+
@Query("update InterviewSession s set s.status = :to, s.endedAt = :now "
48+
+ "where s.id = :id and s.status = com.stackup.stackup.session.domain.SessionStatus.IN_PROGRESS")
49+
int finishIfInProgress(@Param("id") Long id,
50+
@Param("to") SessionStatus to,
51+
@Param("now") Instant now);
3952
}

‎backend/src/main/java/com/stackup/stackup/session/domain/SessionStatus.java‎

Lines changed: 6 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -5,5 +5,10 @@ public enum SessionStatus {
55
IN_PROGRESS,
66
INTERRUPTED,
77
COMPLETED,
8-
CANCELLED
8+
CANCELLED;
9+
10+
// 더 이상 진행/전이가 불가능한 종료 상태. 종료 후 도착하는 비동기 콜백을 드롭하는 가드에 사용.
11+
public boolean isTerminal() {
12+
return this == INTERRUPTED || this == COMPLETED || this == CANCELLED;
13+
}
914
}

‎backend/src/test/java/com/stackup/stackup/session/application/QuestionsCallbackServiceTest.java‎

Lines changed: 4 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -306,6 +306,8 @@ void apply_followupDontKnow_deletesPlaceholderAndAdvances() {
306306
// advanceToNextGeneral 내부에서 sessionRepository.findById 재호출
307307
when(poolRepository.findFirstBySessionIdAndUsedFalseOrderByIdxAsc(22L))
308308
.thenReturn(java.util.Optional.empty());
309+
// endSession 의 원자적 종료 전이 — 전이를 차지(1).
310+
when(sessionRepository.finishIfInProgress(any(), any(), any())).thenReturn(1);
309311

310312
service.apply(env);
311313

@@ -322,6 +324,8 @@ void advanceToNextGeneral_endsWithMaxReached_whenMainQuestionsHitLimit() {
322324
// maxQuestions=5; 메인질문 5개를 이미 던진 상태 → 다음 advance 에서 풀을 보지 않고 종료.
323325
ReflectionTestUtils.setField(session, "totalQuestionCount", 5);
324326
when(sessionRepository.findById(33L)).thenReturn(Optional.of(session));
327+
// endSession 의 원자적 종료 전이 — 전이를 차지(1).
328+
when(sessionRepository.finishIfInProgress(any(), any(), any())).thenReturn(1);
325329

326330
service.advanceToNextGeneral(33L);
327331

‎backend/src/test/java/com/stackup/stackup/session/application/SessionServiceTest.java‎

Lines changed: 1 addition & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -187,6 +187,7 @@ void end_transitionsInProgressToCompleted() {
187187
session.start();
188188
when(sessionRepository.findByIdAndUser_IdAndDeletedFalse(50L, 1L))
189189
.thenReturn(Optional.of(session));
190+
when(sessionRepository.finishIfInProgress(any(), any(), any())).thenReturn(1);
190191

191192
SessionResult result = service.end(1L, 50L);
192193

‎backend/src/test/java/com/stackup/stackup/session/application/SessionTimeoutServiceTest.java‎

Lines changed: 23 additions & 5 deletions
Original file line numberDiff line numberDiff line change
@@ -1,7 +1,7 @@
11
package com.stackup.stackup.session.application;
22

3-
import static org.assertj.core.api.Assertions.assertThat;
43
import static org.mockito.ArgumentMatchers.any;
4+
import static org.mockito.ArgumentMatchers.eq;
55
import static org.mockito.Mockito.never;
66
import static org.mockito.Mockito.verify;
77
import static org.mockito.Mockito.when;
@@ -15,6 +15,7 @@
1515
import com.stackup.stackup.session.domain.SessionMode;
1616
import com.stackup.stackup.session.domain.SessionStatus;
1717
import com.stackup.stackup.user.domain.User;
18+
import java.time.Instant;
1819
import java.util.Optional;
1920
import org.junit.jupiter.api.Test;
2021
import org.junit.jupiter.api.extension.ExtendWith;
@@ -33,30 +34,46 @@ class SessionTimeoutServiceTest {
3334
@InjectMocks SessionTimeoutService service;
3435

3536
@Test
36-
void endTimedOut_withAnswer_completesAndEmitsEndedEvent() {
37+
void endTimedOut_withAnswer_claimsCompletedAndEmitsEndedEvent() {
3738
InterviewSession session = inProgressFixture(50L);
3839
when(sessionRepository.findById(50L)).thenReturn(Optional.of(session));
3940
when(messageRepository.existsBySession_IdAndRole(50L, MessageRole.INTERVIEWEE)).thenReturn(true);
41+
when(sessionRepository.finishIfInProgress(eq(50L), eq(SessionStatus.COMPLETED), any(Instant.class)))
42+
.thenReturn(1);
4043

4144
service.endTimedOut(50L);
4245

43-
assertThat(session.getStatus()).isEqualTo(SessionStatus.COMPLETED);
4446
verify(events).publishEvent(any(SessionEndedEvent.class));
4547
}
4648

4749
@Test
48-
void endTimedOut_withoutAnswer_interruptsAndDoesNotTriggerFeedback() {
50+
void endTimedOut_withoutAnswer_claimsInterruptedAndDoesNotTriggerFeedback() {
4951
InterviewSession session = inProgressFixture(51L);
5052
when(sessionRepository.findById(51L)).thenReturn(Optional.of(session));
5153
when(messageRepository.existsBySession_IdAndRole(51L, MessageRole.INTERVIEWEE)).thenReturn(false);
54+
when(sessionRepository.finishIfInProgress(eq(51L), eq(SessionStatus.INTERRUPTED), any(Instant.class)))
55+
.thenReturn(1);
5256

5357
service.endTimedOut(51L);
5458

55-
assertThat(session.getStatus()).isEqualTo(SessionStatus.INTERRUPTED);
5659
// 답변이 없으면 피드백을 만들지 않는다 — SessionEndedEvent 미발행.
5760
verify(events, never()).publishEvent(any(SessionEndedEvent.class));
5861
}
5962

63+
@Test
64+
void endTimedOut_lostRace_emitsNothing() {
65+
// 조건부 UPDATE 가 0행 → 다른 트랜잭션이 먼저 종료. 어떤 이벤트도 발행하지 않는다(중복 방지).
66+
InterviewSession session = inProgressFixture(53L);
67+
when(sessionRepository.findById(53L)).thenReturn(Optional.of(session));
68+
when(messageRepository.existsBySession_IdAndRole(53L, MessageRole.INTERVIEWEE)).thenReturn(true);
69+
when(sessionRepository.finishIfInProgress(eq(53L), eq(SessionStatus.COMPLETED), any(Instant.class)))
70+
.thenReturn(0);
71+
72+
service.endTimedOut(53L);
73+
74+
verify(events, never()).publishEvent(any());
75+
}
76+
6077
@Test
6178
void endTimedOut_notInProgress_isNoop() {
6279
InterviewSession session = inProgressFixture(52L);
@@ -67,6 +84,7 @@ void endTimedOut_notInProgress_isNoop() {
6784

6885
verify(events, never()).publishEvent(any());
6986
verify(messageRepository, never()).existsBySession_IdAndRole(any(), any());
87+
verify(sessionRepository, never()).finishIfInProgress(any(), any(), any());
7088
}
7189

7290
private InterviewSession inProgressFixture(Long id) {

0 commit comments

Comments
 (0)