[FIX] 청중 재동기화 요청(SYNC_REQUEST) 룸별 레이트리밋

src/main/java/com/debatetimer/service/sharing/SharingRoomRegistry.java

@@ -14,10 +14,13 @@ public class SharingRoomRegistry {
 
     static final Duration FINISHED_TTL = Duration.ofDays(1);
     static final Duration VERSION_TTL = Duration.ofDays(1);
+    static final Duration SYNC_REQUEST_TTL = Duration.ofDays(1);
+    static final Duration SYNC_REQUEST_INTERVAL = Duration.ofMillis(500);
     private static final int ROOM_LOCK_STRIPES = 64;
 
     private final Map<Long, Instant> finishedRoomExpirations = new ConcurrentHashMap<>();
     private final Map<Long, AcceptedVersion> lastAcceptedVersions = new ConcurrentHashMap<>();
+    private final Map<Long, Instant> lastSyncRequestedAt = new ConcurrentHashMap<>();
     private final Object[] roomLocks = createRoomLocks();
     private final Clock clock;
 
@@ -77,6 +80,33 @@ public void resetVersion(long roomId) {
         runExclusively(roomId, () -> lastAcceptedVersions.remove(roomId));
     }
 
+    /**
+     * 룸별로 사회자에게 보내는 상태 공유 요청의 최소 간격을 강제한다.
+     * 여러 청중이 동시에 입장하거나 재구독하면 요청이 그만큼 늘어나고, 사회자의 응답은 룸 전체로 중계되므로
+     * 메시지가 크게 증폭된다. 짧은 간격의 뒤따르는 요청은 직전 요청의 응답이 대신하므로 버린다.
+     *
+     * @return 요청을 보내도 되면 true
+     */
+    public boolean tryAcquireSyncRequest(long roomId) {
+        Instant now = clock.instant();
+        AtomicBoolean acquired = new AtomicBoolean(false);
+        lastSyncRequestedAt.compute(roomId, (id, lastRequested) -> {
+            if (lastRequested != null && now.isBefore(lastRequested.plus(SYNC_REQUEST_INTERVAL))) {
+                return lastRequested;
+            }
+            acquired.set(true);
+            return now;
+        });
+        return acquired.get();
+    }
+
+    /**
+     * 룸의 상태 공유 요청 간격 기준을 지운다. 다음 요청은 간격과 무관하게 허용된다.
+     */
+    public void resetSyncRequest(long roomId) {
+        lastSyncRequestedAt.remove(roomId);
+    }
+
     /**
      * 같은 룸의 작업(버전 수락·룸 상태 변경·중계, 버전 초기화)을 하나씩 실행한다.
      * 인바운드 메시지는 여러 스레드에서 동시에 처리될 수 있어, 낮은 버전의 상태 변경이나 중계가
@@ -92,6 +122,8 @@ public void removeExpired() {
         finishedRoomExpirations.values().removeIf(this::isExpired);
         lastAcceptedVersions.values()
                 .removeIf(accepted -> isExpired(accepted.acceptedAt().plus(VERSION_TTL)));
+        lastSyncRequestedAt.values()
+                .removeIf(requestedAt -> isExpired(requestedAt.plus(SYNC_REQUEST_TTL)));
     }
 
     private boolean isExpired(Instant expiration) {

src/main/java/com/debatetimer/service/sharing/SharingService.java

@@ -70,11 +70,13 @@ public void share(long roomId, String chairmanSessionId, SharingRequest request)
 
     /**
      * 사회자의 구독은 공유 시작을 뜻하므로, 최신 사회자로 등록하고 이전에 종료된 룸이라도 다시 진행 상태로 되돌린다.
+     * 새 사회자는 직전 공유 요청의 응답을 보낸 적이 없으므로, 상태 공유 요청 간격 기준도 함께 지운다.
      */
     public void startChairman(long roomId, String chairmanSessionId, String simpSessionId) {
         sharingRoomRegistry.runExclusively(roomId, () -> {
             if (claimChairman(roomId, chairmanSessionId, simpSessionId).isAccepted()) {
                 sharingRoomRegistry.reopen(roomId);
+                sharingRoomRegistry.resetSyncRequest(roomId);
             }
         });
     }
@@ -84,6 +86,8 @@ public void startChairman(long roomId, String chairmanSessionId, String simpSess
      * - 종료된 룸이면 청중에게 종료를 알린다.
      * - 활성 사회자가 없으면 청중에게 사회자 부재를 알린다.
      * - 그 외에는 사회자에게 현재 상태 공유를 요청한다.
+     * 사회자의 응답은 룸 전체로 중계되므로, 짧은 간격의 뒤따르는 요청은 직전 요청의 응답이 대신한다.
+     * 그래서 룸별 최소 간격 안의 요청은 보내지 않는다.
      */
     public void joinAudience(long roomId) {
         if (sharingRoomRegistry.isFinished(roomId)) {
@@ -95,6 +99,9 @@ public void joinAudience(long roomId) {
                     new SharingResponse(TimerEventType.CHAIRMAN_ABSENT));
             return;
         }
+        if (!sharingRoomRegistry.tryAcquireSyncRequest(roomId)) {
+            return;
+        }
         messagingTemplate.convertAndSend(CHAIRMAN_CHANNEL_PREFIX + roomId, ChairmanSharingRequest.syncRequest(roomId));
     }
 

src/test/java/com/debatetimer/controller/sharing/SharingWebSocketControllerTest.java

@@ -65,6 +65,7 @@ void startChairman() throws InterruptedException {
     void reopenRoom() {
         sharingRoomRegistry.reopen(ROOM_ID);
         sharingRoomRegistry.resetVersion(ROOM_ID);
+        sharingRoomRegistry.resetSyncRequest(ROOM_ID);
         chairmanSessionRegistry.remove(ROOM_ID);
     }
 

src/test/java/com/debatetimer/event/sharing/RoomSubscribeInterceptorTest.java

@@ -59,6 +59,7 @@ void setUpRoomOwner() {
     void reopenRoom() {
         sharingRoomRegistry.reopen(ROOM_ID);
         sharingRoomRegistry.resetVersion(ROOM_ID);
+        sharingRoomRegistry.resetSyncRequest(ROOM_ID);
         chairmanSessionRegistry.remove(ROOM_ID);
     }
 
@@ -220,6 +221,41 @@ class AfterMessageHandled {
         }
     }
 
+    @Nested
+    class SyncRequestInterval {
+
+        @Test
+        void 청중이_연달아_구독해도_사회자에게_정보공유_트리거를_한_번만_발송한다() throws Exception {
+            QueueFrameHandler<ChairmanSharingRequest> chairmanHandler = new QueueFrameHandler<>(
+                    ChairmanSharingRequest.class);
+            subscribeChairman(chairmanHandler);
+
+            stompSession.subscribe("/room/" + ROOM_ID, new MessageFrameHandler<>(SharingResponse.class));
+            assertThat(chairmanHandler.poll(3L)).isNotNull(); // 첫 구독 처리 완료 대기
+            stompSession.subscribe("/room/" + ROOM_ID, new MessageFrameHandler<>(SharingResponse.class));
+
+            assertThat(chairmanHandler.poll(2L)).isNull();
+        }
+
+        @Test
+        void 요청_간격_기준을_지우면_사회자에게_정보공유_트리거를_다시_발송한다() throws Exception {
+            QueueFrameHandler<ChairmanSharingRequest> chairmanHandler = new QueueFrameHandler<>(
+                    ChairmanSharingRequest.class);
+            subscribeChairman(chairmanHandler);
+            stompSession.subscribe("/room/" + ROOM_ID, new MessageFrameHandler<>(SharingResponse.class));
+            assertThat(chairmanHandler.poll(3L)).isNotNull(); // 첫 구독 처리 완료 대기
+
+            sharingRoomRegistry.resetSyncRequest(ROOM_ID);
+            stompSession.subscribe("/room/" + ROOM_ID, new MessageFrameHandler<>(SharingResponse.class));
+
+            ChairmanSharingRequest sharingRequest = chairmanHandler.poll(3L);
+            assertAll(
+                    () -> assertThat(sharingRequest).isNotNull(),
+                    () -> assertThat(sharingRequest.type()).isEqualTo(ChairmanNoticeType.SYNC_REQUEST)
+            );
+        }
+    }
+
     @Nested
     class ChairmanPresence {
 

src/test/java/com/debatetimer/service/sharing/SharingRoomRegistryTest.java

@@ -266,6 +266,69 @@ class RemoveExpiredVersion {
         }
     }
 
+    @Nested
+    class TryAcquireSyncRequest {
+
+        @Test
+        void 처음_요청은_허용한다() {
+            assertThat(sharingRoomRegistry.tryAcquireSyncRequest(1L)).isTrue();
+        }
+
+        @Test
+        void 최소_간격_안의_뒤따르는_요청은_거절한다() {
+            sharingRoomRegistry.tryAcquireSyncRequest(1L);
+
+            clock.advance(SharingRoomRegistry.SYNC_REQUEST_INTERVAL.minusMillis(1));
+
+            assertThat(sharingRoomRegistry.tryAcquireSyncRequest(1L)).isFalse();
+        }
+
+        @Test
+        void 같은_시각의_연속_요청은_거절한다() {
+            sharingRoomRegistry.tryAcquireSyncRequest(1L);
+
+            assertThat(sharingRoomRegistry.tryAcquireSyncRequest(1L)).isFalse();
+        }
+
+        @Test
+        void 최소_간격이_지난_요청은_허용한다() {
+            sharingRoomRegistry.tryAcquireSyncRequest(1L);
+
+            clock.advance(SharingRoomRegistry.SYNC_REQUEST_INTERVAL);
+
+            assertThat(sharingRoomRegistry.tryAcquireSyncRequest(1L)).isTrue();
+        }
+
+        @Test
+        void 거절된_요청은_마지막_요청_시각을_바꾸지_않는다() {
+            sharingRoomRegistry.tryAcquireSyncRequest(1L);
+            clock.advance(SharingRoomRegistry.SYNC_REQUEST_INTERVAL.minusMillis(1));
+            sharingRoomRegistry.tryAcquireSyncRequest(1L);
+
+            clock.advance(Duration.ofMillis(1));
+
+            assertThat(sharingRoomRegistry.tryAcquireSyncRequest(1L)).isTrue();
+        }
+
+        @Test
+        void 룸마다_요청_간격을_따로_관리한다() {
+            sharingRoomRegistry.tryAcquireSyncRequest(1L);
+
+            assertThat(sharingRoomRegistry.tryAcquireSyncRequest(2L)).isTrue();
+        }
+
+        @Test
+        void 유지_시간이_지난_요청_기록을_제거한다() {
+            sharingRoomRegistry.tryAcquireSyncRequest(1L);
+            clock.advance(SharingRoomRegistry.SYNC_REQUEST_TTL);
+
+            sharingRoomRegistry.removeExpired();
+            clock.advance(SharingRoomRegistry.SYNC_REQUEST_TTL.negated().plusMillis(1)); // 제거되지 않았다면 거절되는 시점
+
+            assertThat(sharingRoomRegistry.tryAcquireSyncRequest(1L)).isTrue();
+        }
+    }
+
     private static class MutableClock extends Clock {
 
         private Instant instant;

src/test/java/com/debatetimer/service/sharing/SharingServiceTest.java

@@ -20,7 +20,9 @@
 import com.debatetimer.dto.sharing.response.SharingResponse;
 import com.debatetimer.exception.custom.DTClientErrorException;
 import java.time.Clock;
+import java.time.Duration;
 import java.time.Instant;
+import java.time.ZoneId;
 import java.time.ZoneOffset;
 import java.util.List;
 import org.junit.jupiter.api.BeforeEach;
@@ -37,14 +39,15 @@ class SharingServiceTest {
     private static final String CHAIRMAN = "tab-a";
     private static final String SIMP_SESSION = "simp-1";
 
+    private MutableClock clock;
     private SharingRoomRegistry sharingRoomRegistry;
     private ChairmanSessionRegistry chairmanSessionRegistry;
     private SimpMessageSendingOperations messagingTemplate;
     private SharingService sharingService;
 
     @BeforeEach
     void setUp() {
-        Clock clock = Clock.fixed(NOW, ZoneOffset.UTC);
+        clock = new MutableClock(NOW);
         sharingRoomRegistry = new SharingRoomRegistry(clock);
         chairmanSessionRegistry = new ChairmanSessionRegistry(clock);
         messagingTemplate = mock(SimpMessageSendingOperations.class);
@@ -282,5 +285,56 @@ class JoinAudience {
                     () -> verify(messagingTemplate, never()).convertAndSend(eq("/room/" + ROOM_ID), any(Object.class))
             );
         }
+
+        @Test
+        void 최소_간격_안에_다시_입장하면_상태_공유를_요청하지_않는다() {
+            sharingService.startChairman(ROOM_ID, CHAIRMAN, SIMP_SESSION);
+
+            sharingService.joinAudience(ROOM_ID);
+            clock.advance(SharingRoomRegistry.SYNC_REQUEST_INTERVAL.minusMillis(1));
+            sharingService.joinAudience(ROOM_ID);
+
+            assertThat(sentChairmanNotices(1)).hasSize(1);
+        }
+
+        @Test
+        void 최소_간격이_지난_뒤_입장하면_상태_공유를_다시_요청한다() {
+            sharingService.startChairman(ROOM_ID, CHAIRMAN, SIMP_SESSION);
+
+            sharingService.joinAudience(ROOM_ID);
+            clock.advance(SharingRoomRegistry.SYNC_REQUEST_INTERVAL);
+            sharingService.joinAudience(ROOM_ID);
+
+            assertThat(sentChairmanNotices(2))
+                    .allMatch(notice -> notice.type() == ChairmanNoticeType.SYNC_REQUEST);
+        }
+    }
+
+    private static class MutableClock extends Clock {
+
+        private Instant instant;
+
+        private MutableClock(Instant instant) {
+            this.instant = instant;
+        }
+
+        private void advance(Duration duration) {
+            instant = instant.plus(duration);
+        }
+
+        @Override
+        public ZoneId getZone() {
+            return ZoneOffset.UTC;
+        }
+
+        @Override
+        public Clock withZone(ZoneId zone) {
+            return this;
+        }
+
+        @Override
+        public Instant instant() {
+            return instant;
+        }
     }
 }