修复麦位状态

nacos_config/rc-service-external/application.yml

@@ -49,7 +49,7 @@ red-circle:
             noticeAccount: ${LIKEI_IM_NOTICE_ACCOUNT}
             newsletterAccount: ${LIKEI_IM_NEWSLETTER_ACCOUNT}
         agora:
-            appId: ${LIKEI_AGORA_APP_ID}
+            appId: ${LIKEI_RTC_APP_ID}
             appCertificate: ${LIKEI_AGORA_APP_CERTIFICATE}
             tokenExpireSecond: ${LIKEI_AGORA_TOKEN_EXPIRE_SECOND}
             privilegeExpireSecond: ${LIKEI_AGORA_PRIVILEGE_EXPIRE_SECOND}

rc-service/rc-service-live/live-application/src/main/java/com/red/circle/live/app/command/mic/LiveMicGoUpCmdExe.java

@@ -4,13 +4,15 @@ import com.red.circle.component.redis.service.RedisService;
 import com.red.circle.external.inner.endpoint.message.TrtcClient;
 import com.red.circle.framework.core.asserts.ResponseAssert;
 import com.red.circle.framework.core.response.CommonErrorCode;
+import com.red.circle.framework.dto.ResultResponse;
 import com.red.circle.live.app.convertor.LiveMicAppConvertor;
 import com.red.circle.live.app.dto.clientobject.LiveMicrophoneCO;
 import com.red.circle.live.app.dto.cmd.MicUpDownCmd;
-import com.red.circle.live.domain.gateway.LiveMicrophoneGateway;
-import com.red.circle.live.domain.live.dto.LiveMicrophoneNoticeDTO;
-import com.red.circle.live.infra.database.cache.service.UserAgoraTokenCacheService;
-import com.red.circle.live.infra.database.mongo.service.LiveMicSettingService;
+import com.red.circle.live.domain.gateway.LiveMicrophoneGateway;
+import com.red.circle.live.domain.live.dto.LiveMicrophoneNoticeDTO;
+import com.red.circle.live.infra.database.cache.service.LiveMicCacheService;
+import com.red.circle.live.infra.database.cache.service.UserAgoraTokenCacheService;
+import com.red.circle.live.infra.database.mongo.service.LiveMicSettingService;
 import com.red.circle.live.infra.util.DateTimeAsiaRiyadhUtils;
 import com.red.circle.live.inner.error.LiveMicErrorCode;
 import com.red.circle.other.inner.endpoint.live.RoomManagerClient;
@@ -43,41 +45,95 @@ public class LiveMicGoUpCmdExe {
   private static final Logger log = LoggerFactory.getLogger(LiveMicGoUpCmdExe.class);
   private final LiveMicAppConvertor liveMicAppConvertor;
   private final LiveMicSettingService liveMicSettingService;
-  private final LiveMicrophoneGateway liveMicrophoneGateway;
-  private final TrtcClient trtcClient;
-  private final UserAgoraTokenCacheService userAgoraTokenCacheService;
+  private final LiveMicrophoneGateway liveMicrophoneGateway;
+  private final LiveMicCacheService liveMicCacheService;
+  private final TrtcClient trtcClient;
+  private final UserAgoraTokenCacheService userAgoraTokenCacheService;
   private final RedisService redisService;
   private final RoomDailyTaskClient roomDailyTaskClient;
   private final RoomManagerClient roomManagerClient;
   private final RedisTemplate redisTemplate;
   private final RoomMemberClient roomMemberClient;
 
-  public LiveMicrophoneCO execute(MicUpDownCmd cmd) {
-    checkPermissions(cmd);
-    LiveMicrophoneNoticeDTO noticeDTO = liveMicrophoneGateway.goUp(
+  public LiveMicrophoneCO execute(MicUpDownCmd cmd) {
+    boolean targetMicAlreadySelf = false;
+    boolean goUpCompleted = false;
+    try {
+      checkPermissions(cmd);
+      targetMicAlreadySelf = checkTargetMicAvailableOrSelf(cmd);
+      String roomToken = createPublisherToken(cmd);
+
+      LiveMicrophoneNoticeDTO noticeDTO = liveMicrophoneGateway.goUp(
+          cmd.getRoomId(),
+          cmd.getMickIndex(),
+          cmd.requiredReqUserId()
+      );
+      goUpCompleted = true;
+
+      cachePublisherToken(cmd, roomToken);
+      LiveMicrophoneCO notice = returnMicrophoneCO(cmd, noticeDTO);
+      notice.setRoomToken(roomToken);
+
+      // 房主邀请成员上麦
+      handleRoomDailyTask(cmd);
+      return notice;
+    } catch (RuntimeException e) {
+      if (goUpCompleted) {
+        rollbackPublisherToken(cmd);
+        if (!targetMicAlreadySelf) {
+          rollbackMicSeat(cmd, e);
+        }
+      }
+      throw e;
+    }
+  }
+
+  private String createPublisherToken(MicUpDownCmd cmd) {
+    ResultResponse<String> response = trtcClient.createAgoraTokenGoUp(
+        cmd.requiredReqUserId(),
+        cmd.getRoomId().toString()
+    );
+    String roomToken = Objects.nonNull(response) ? response.getBody() : null;
+    ResponseAssert.notNull(CommonErrorCode.THIRD_PARTY_SERVER_ERROR, roomToken);
+    return roomToken;
+  }
+
+  private void cachePublisherToken(MicUpDownCmd cmd, String roomToken) {
+    userAgoraTokenCacheService.remove(cmd.requiredReqUserId(), cmd.getRoomId());
+    userAgoraTokenCacheService.save(roomToken, cmd.requiredReqUserId(), cmd.getRoomId());
+  }
+
+  private boolean checkTargetMicAvailableOrSelf(MicUpDownCmd cmd) {
+    boolean targetMicAlreadySelf = liveMicCacheService.checkEqSelf(
         cmd.getRoomId(),
         cmd.getMickIndex(),
         cmd.requiredReqUserId()
     );
-    LiveMicrophoneCO notice = returnMicrophoneCO(cmd, noticeDTO);
-    notice.setRoomToken(refreshPublisherToken(cmd));
-    //notice.setRoomToken(createAgoraTokenGoUp(cmd));
-
-    // 房主邀请成员上麦
-    handleRoomDailyTask(cmd);
-
-    return notice;
+    if (targetMicAlreadySelf) {
+      return true;
+    }
+    ResponseAssert.isTrue(LiveMicErrorCode.MICROPHONE_OCCUPIED,
+        liveMicCacheService.checkAvailable(cmd.getRoomId(), cmd.getMickIndex()));
+    return false;
   }
 
-  private String refreshPublisherToken(MicUpDownCmd cmd) {
-    String roomToken = trtcClient.createAgoraTokenGoUp(
-        cmd.getReqUserId(),
-        cmd.getRoomId().toString()
-    ).getBody();
-    ResponseAssert.notNull(CommonErrorCode.THIRD_PARTY_SERVER_ERROR, roomToken);
-    userAgoraTokenCacheService.remove(cmd.getReqUserId(), cmd.getRoomId());
-    userAgoraTokenCacheService.save(roomToken, cmd.getReqUserId(), cmd.getRoomId());
-    return roomToken;
+  private void rollbackPublisherToken(MicUpDownCmd cmd) {
+    try {
+      userAgoraTokenCacheService.remove(cmd.requiredReqUserId(), cmd.getRoomId());
+    } catch (RuntimeException rollbackException) {
+      log.warn("上麦失败回滚 publisher token 失败 roomId={}, userId={}",
+          cmd.getRoomId(), cmd.requiredReqUserId(), rollbackException);
+    }
+  }
+
+  private void rollbackMicSeat(MicUpDownCmd cmd, RuntimeException cause) {
+    try {
+      liveMicrophoneGateway.goDown(cmd.getRoomId(), cmd.getMickIndex(), cmd.requiredReqUserId());
+    } catch (RuntimeException rollbackException) {
+      log.warn("上麦失败回滚麦位失败 roomId={}, micIndex={}, userId={}, cause={}",
+          cmd.getRoomId(), cmd.getMickIndex(), cmd.requiredReqUserId(), cause.getMessage(),
+          rollbackException);
+    }
   }
 
     private void handleRoomDailyTask(MicUpDownCmd cmd) {

rc-service/rc-service-live/live-application/src/main/java/com/red/circle/live/app/listener/UserHeartbeatListener.java

@@ -10,6 +10,7 @@ import com.red.circle.component.redis.service.RedisService;
 import com.red.circle.external.inner.model.dto.AgoraChannelStateDTO;
 import com.red.circle.framework.dto.ResultResponse;
 import com.red.circle.live.app.dto.cmd.HeartbeatStatusEnum;
+import com.red.circle.live.app.service.AgoraMicMissingConfirmService;
 import com.red.circle.live.app.service.AgoraMicStateCleanupService;
 import com.red.circle.live.app.service.AgoraRoomStateService;
 import com.red.circle.live.domain.gateway.LiveMicrophoneGateway;
@@ -51,8 +52,7 @@ import java.util.stream.Collectors;
 import lombok.RequiredArgsConstructor;
 import lombok.extern.slf4j.Slf4j;
 import org.springframework.context.annotation.Profile;
-import org.springframework.data.redis.core.RedisTemplate;
-import org.springframework.stereotype.Component;
+import org.springframework.stereotype.Component;
 
 /**
  * 用户心跳.
@@ -78,6 +78,7 @@ public class UserHeartbeatListener implements MessageListener {
   private final RedisService redisService;
   private final RoomMemberClient roomMemberClient;
   private final AgoraRoomStateService agoraRoomStateService;
+  private final AgoraMicMissingConfirmService agoraMicMissingConfirmService;
   private final AgoraMicStateCleanupService agoraMicStateCleanupService;
 
   private static final int HEARTBEAT_INTERVAL_SECONDS = 60;
@@ -214,10 +215,11 @@ public class UserHeartbeatListener implements MessageListener {
     UserProfileDTO actualUserProfile = ensureUserProfile(userProfile);
     AgoraChannelStateDTO channelState = agoraRoomStateService.getChannelState(roomId);
     if (channelState.querySuccess() && !isUserInAgoraChannel(channelState, actualUserProfile)) {
-      agoraMicStateCleanupService.cleanupUser(roomId, actualUserProfile.getId(),
+      handleAgoraUserMissing(roomId, actualUserProfile.getId(), channelState,
           "heartbeat agora user missing");
       return;
     }
+    clearAgoraMissingAfterPresent(channelState, roomId, actualUserProfile.getId());
 
     // 添加房间在线用户
     addRoomOnlineUser(actualUserProfile, roomProfile);
@@ -314,10 +316,11 @@ public class UserHeartbeatListener implements MessageListener {
     }
 
     if (!isUserInAgoraChannel(channelState, userProfile)) {
-      agoraMicStateCleanupService.cleanupUser(roomProfile.getId(), userProfile.getId(),
+      handleAgoraUserMissing(roomProfile.getId(), userProfile.getId(), channelState,
           "heartbeat agora user missing");
       return;
     }
+    clearAgoraMissingAfterPresent(channelState, roomProfile.getId(), userProfile.getId());
 
     refreshMicActiveTimeWithoutAgoraCheck(event, roomProfile, liveMicrophone);
   }
@@ -370,4 +373,28 @@ public class UserHeartbeatListener implements MessageListener {
         && channelState.getAudience().contains(agoraUid));
   }
 
+  private void handleAgoraUserMissing(Long roomId,
+      Long userId,
+      AgoraChannelStateDTO channelState,
+      String reason) {
+    if (!agoraMicMissingConfirmService.markMissingAndShouldCleanup(roomId, userId,
+        appendChannelStateReason(channelState, reason))) {
+      return;
+    }
+    agoraMicStateCleanupService.cleanupUser(roomId, userId, reason);
+  }
+
+  private void clearAgoraMissingAfterPresent(AgoraChannelStateDTO channelState, Long roomId, Long userId) {
+    if (Objects.nonNull(channelState) && channelState.querySuccess()) {
+      agoraMicMissingConfirmService.clearMissing(roomId, userId);
+    }
+  }
+
+  private String appendChannelStateReason(AgoraChannelStateDTO channelState, String reason) {
+    if (Objects.isNull(channelState)) {
+      return reason;
+    }
+    return reason + ", channelExist=" + channelState.channelExist();
+  }
+
 }

rc-service/rc-service-live/live-application/src/main/java/com/red/circle/live/app/scheduler/AgoraMicStateReconcileTask.java

@@ -2,6 +2,7 @@ package com.red.circle.live.app.scheduler;
 
 import com.red.circle.component.redis.service.RedisService;
 import com.red.circle.external.inner.model.dto.AgoraChannelStateDTO;
+import com.red.circle.live.app.service.AgoraMicMissingConfirmService;
 import com.red.circle.live.app.service.AgoraMicStateCleanupService;
 import com.red.circle.live.app.service.AgoraRoomStateService;
 import com.red.circle.live.domain.live.LiveMicrophone;
@@ -35,6 +36,7 @@ public class AgoraMicStateReconcileTask {
   private final RedisTemplate<String, Object> redisTemplate;
   private final LiveMicCacheService liveMicCacheService;
   private final AgoraRoomStateService agoraRoomStateService;
+  private final AgoraMicMissingConfirmService agoraMicMissingConfirmService;
   private final AgoraMicStateCleanupService agoraMicStateCleanupService;
 
   @Scheduled(fixedDelayString = "${red-circle.live.agora-mic-reconcile-delay-ms:30000}")
@@ -62,13 +64,15 @@ public class AgoraMicStateReconcileTask {
       return;
     }
 
-    if (!state.channelExist()) {
-      agoraMicStateCleanupService.cleanupRoom(roomId, "reconcile agora channel not exist");
+    List<LiveMicrophone> microphones = liveMicCacheService.listLiveMicrophone(roomId);
+    if (CollectionUtils.isEmpty(microphones)) {
       return;
     }
 
-    List<LiveMicrophone> microphones = liveMicCacheService.listLiveMicrophone(roomId);
-    if (CollectionUtils.isEmpty(microphones)) {
+    if (!state.channelExist()) {
+      for (LiveMicrophone microphone : microphones) {
+        cleanupMissingMicrophoneUser(roomId, microphone, state, "reconcile agora channel not exist");
+      }
       return;
     }
 
@@ -85,12 +89,30 @@ public class AgoraMicStateReconcileTask {
       }
       Long agoraUid = DataTypeUtils.toLong(microphone.getUser().getAccount());
       if (Objects.isNull(agoraUid) || !channelUids.contains(agoraUid)) {
-        agoraMicStateCleanupService.cleanupUser(roomId, microphone.getUser().getId(),
-            "reconcile agora user missing");
+        cleanupMissingMicrophoneUser(roomId, microphone, state, "reconcile agora user missing");
+      } else {
+        agoraMicMissingConfirmService.clearMissing(roomId, microphone.getUser().getId());
       }
     }
   }
 
+  private void cleanupMissingMicrophoneUser(Long roomId,
+      LiveMicrophone microphone,
+      AgoraChannelStateDTO state,
+      String reason) {
+    if (Objects.isNull(microphone)
+        || Objects.isNull(microphone.getUser())
+        || Objects.isNull(microphone.getUser().getId())) {
+      return;
+    }
+    Long userId = microphone.getUser().getId();
+    if (!agoraMicMissingConfirmService.markMissingAndShouldCleanup(roomId, userId,
+        reason + ", channelExist=" + state.channelExist())) {
+      return;
+    }
+    agoraMicStateCleanupService.cleanupUser(roomId, userId, reason);
+  }
+
   private Set<Long> scanSeatRoomIds() {
     Set<Long> roomIds = new HashSet<>();
     ScanOptions options = ScanOptions.scanOptions().match(SEAT_KEY_PREFIX + "*").count(200).build();

rc-service/rc-service-live/live-application/src/main/java/com/red/circle/live/app/service/AgoraMicMissingConfirmService.java

@@ -0,0 +1,102 @@
+package com.red.circle.live.app.service;
+
+import com.red.circle.component.redis.service.RedisService;
+import java.util.Objects;
+import java.util.concurrent.TimeUnit;
+import lombok.RequiredArgsConstructor;
+import lombok.extern.slf4j.Slf4j;
+import org.springframework.stereotype.Service;
+
+/**
+ * 声网 missing 状态二次确认.
+ */
+@Slf4j
+@Service
+@RequiredArgsConstructor
+public class AgoraMicMissingConfirmService {
+
+  private static final String KEY_PREFIX = "agora:mic:missing:";
+  private static final int MISSING_TTL_SECONDS = 10;
+  private static final int CONFIRM_MISSING_COUNT = 2;
+  private static final long CONFIRM_MISSING_AFTER_MILLIS = 5_000L;
+
+  private final RedisService redisService;
+
+  public boolean markMissingAndShouldCleanup(Long roomId, Long userId, String reason) {
+    if (Objects.isNull(roomId) || Objects.isNull(userId)) {
+      return false;
+    }
+
+    String key = missingKey(roomId, userId);
+    long nowMillis = System.currentTimeMillis();
+    MissingState nextState = MissingState.next(redisService.getString(key), nowMillis);
+    if (nextState.shouldCleanup(nowMillis)) {
+      redisService.delete(key);
+      log.warn("声网 missing 已二次确认，允许清理 roomId={}, userId={}, count={}, firstMissingAt={}, reason={}",
+          roomId, userId, nextState.count, nextState.firstMissingAtMillis, reason);
+      return true;
+    }
+
+    redisService.setString(key, nextState.serialize(), MISSING_TTL_SECONDS, TimeUnit.SECONDS);
+    log.warn("声网 missing 首次缓冲，跳过本次清理和刷新 roomId={}, userId={}, count={}, firstMissingAt={}, reason={}",
+        roomId, userId, nextState.count, nextState.firstMissingAtMillis, reason);
+    return false;
+  }
+
+  public void clearMissing(Long roomId, Long userId) {
+    if (Objects.isNull(roomId) || Objects.isNull(userId)) {
+      return;
+    }
+    redisService.delete(missingKey(roomId, userId));
+  }
+
+  private String missingKey(Long roomId, Long userId) {
+    return KEY_PREFIX + roomId + ":" + userId;
+  }
+
+  static class MissingState {
+
+    private final int count;
+    private final long firstMissingAtMillis;
+
+    private MissingState(int count, long firstMissingAtMillis) {
+      this.count = count;
+      this.firstMissingAtMillis = firstMissingAtMillis;
+    }
+
+    static MissingState next(String rawValue, long nowMillis) {
+      MissingState current = parse(rawValue, nowMillis);
+      long firstMissingAtMillis = current.count <= 0 ? nowMillis : current.firstMissingAtMillis;
+      return new MissingState(current.count + 1, firstMissingAtMillis);
+    }
+
+    boolean shouldCleanup(long nowMillis) {
+      return count >= CONFIRM_MISSING_COUNT
+          || nowMillis - firstMissingAtMillis >= CONFIRM_MISSING_AFTER_MILLIS;
+    }
+
+    String serialize() {
+      return count + ":" + firstMissingAtMillis;
+    }
+
+    private static MissingState parse(String rawValue, long nowMillis) {
+      if (Objects.isNull(rawValue) || rawValue.isBlank()) {
+        return new MissingState(0, nowMillis);
+      }
+      String[] parts = rawValue.split(":", 2);
+      if (parts.length != 2) {
+        return new MissingState(0, nowMillis);
+      }
+      try {
+        int count = Integer.parseInt(parts[0]);
+        long firstMissingAtMillis = Long.parseLong(parts[1]);
+        if (count <= 0 || firstMissingAtMillis <= 0 || firstMissingAtMillis > nowMillis) {
+          return new MissingState(0, nowMillis);
+        }
+        return new MissingState(count, firstMissingAtMillis);
+      } catch (NumberFormatException e) {
+        return new MissingState(0, nowMillis);
+      }
+    }
+  }
+}

rc-service/rc-service-live/live-application/src/test/java/com/red/circle/live/app/command/mic/LiveMicGoUpCmdExeTest.java

@@ -0,0 +1,222 @@
+package com.red.circle.live.app.command.mic;
+
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertSame;
+import static org.junit.Assert.fail;
+
+import com.red.circle.external.inner.endpoint.message.TrtcClient;
+import com.red.circle.framework.dto.ResultResponse;
+import com.red.circle.live.app.convertor.LiveMicAppConvertor;
+import com.red.circle.live.app.dto.clientobject.LiveMicUserCO;
+import com.red.circle.live.app.dto.clientobject.LiveMicrophoneCO;
+import com.red.circle.live.app.dto.cmd.MicUpDownCmd;
+import com.red.circle.live.domain.gateway.LiveMicrophoneGateway;
+import com.red.circle.live.domain.live.LiveMicUser;
+import com.red.circle.live.domain.live.LiveMicrophone;
+import com.red.circle.live.domain.live.dto.LiveMicrophoneDTO;
+import com.red.circle.live.domain.live.dto.LiveMicrophoneNoticeDTO;
+import com.red.circle.live.infra.database.cache.service.LiveMicCacheService;
+import com.red.circle.live.infra.database.cache.service.UserAgoraTokenCacheService;
+import com.red.circle.live.infra.database.mongo.entity.LiveMicSetting;
+import com.red.circle.live.infra.database.mongo.entity.LiveMicSettingSeat;
+import com.red.circle.live.infra.database.mongo.service.LiveMicSettingService;
+import com.red.circle.other.inner.endpoint.live.RoomManagerClient;
+import java.lang.reflect.InvocationHandler;
+import java.lang.reflect.Method;
+import java.lang.reflect.Proxy;
+import java.util.ArrayList;
+import java.util.List;
+import org.junit.Test;
+
+public class LiveMicGoUpCmdExeTest {
+
+  private static final Long ROOM_ID = 1001L;
+  private static final Long USER_ID = 2002L;
+  private static final Integer MIC_INDEX = 1;
+
+  @Test
+  public void executeShouldCreateTokenBeforeGoUpAndCacheAfterGoUp() {
+    Fixture fixture = new Fixture();
+
+    LiveMicrophoneCO result = fixture.exe.execute(command());
+
+    assertEquals("publisher-token", result.getRoomToken());
+    assertEquals(List.of("checkEqSelf", "checkAvailable", "createToken", "goUp", "removeToken",
+        "saveToken", "convert", "getRoom"), fixture.calls);
+  }
+
+  @Test
+  public void executeShouldRollbackMicAndTokenWhenCacheFailsAfterGoUp() {
+    Fixture fixture = new Fixture();
+    RuntimeException cacheException = new RuntimeException("cache failed");
+    fixture.cacheException = cacheException;
+
+    try {
+      fixture.exe.execute(command());
+      fail("Expected cache exception");
+    } catch (RuntimeException e) {
+      assertSame(cacheException, e);
+    }
+
+    assertEquals(List.of("checkEqSelf", "checkAvailable", "createToken", "goUp", "removeToken",
+        "saveToken", "removeToken", "goDown"), fixture.calls);
+  }
+
+  private static MicUpDownCmd command() {
+    MicUpDownCmd cmd = new MicUpDownCmd()
+        .setRoomId(ROOM_ID)
+        .setMickIndex(MIC_INDEX);
+    cmd.setReqUserId(USER_ID);
+    return cmd;
+  }
+
+  private static LiveMicrophoneNoticeDTO notice() {
+    return new LiveMicrophoneNoticeDTO()
+        .setMics(List.of(new LiveMicrophoneDTO().setRoomId(ROOM_ID).setMicIndex(MIC_INDEX)));
+  }
+
+  @SuppressWarnings("unchecked")
+  private static <T> T proxy(Class<T> type, InvocationHandler handler) {
+    return (T) Proxy.newProxyInstance(type.getClassLoader(), new Class<?>[] {type},
+        (Object proxy, Method method, Object[] args) -> {
+          if (method.getDeclaringClass().equals(Object.class)) {
+            return method.invoke(handler, args);
+          }
+          return handler.invoke(proxy, method, args);
+        });
+  }
+
+  private static class Fixture {
+
+    private final List<String> calls = new ArrayList<>();
+    private RuntimeException cacheException;
+
+    private final LiveMicAppConvertor liveMicAppConvertor = new LiveMicAppConvertor() {
+      @Override
+      public List<LiveMicrophoneCO> toListLiveMicrophoneCO(List<LiveMicrophoneDTO> liveMicrophones) {
+        return null;
+      }
+
+      @Override
+      public LiveMicrophoneCO toLiveMicrophoneCO(LiveMicrophoneDTO dto) {
+        calls.add("convert");
+        return new LiveMicrophoneCO();
+      }
+
+      @Override
+      public LiveMicUserCO toLiveMicUserCO(LiveMicUser liveMicUser) {
+        return null;
+      }
+    };
+
+    private final LiveMicSettingService liveMicSettingService = new LiveMicSettingService() {
+      @Override
+      public LiveMicSetting updateMicSize(Long roomId, int micSize) {
+        return null;
+      }
+
+      @Override
+      public void updateMicSeat(Long roomId, LiveMicSettingSeat settingSeat) {
+      }
+
+      @Override
+      public LiveMicSettingSeat getMicSeat(Long roomId, Integer micIndex) {
+        return null;
+      }
+
+      @Override
+      public List<LiveMicSettingSeat> listMicSeat(Long roomId) {
+        return null;
+      }
+
+      @Override
+      public boolean checkLock(Long roomId, Integer micIndex) {
+        return false;
+      }
+
+      @Override
+      public boolean exists(Long roomId) {
+        return false;
+      }
+    };
+
+    private final LiveMicrophoneGateway liveMicrophoneGateway = proxy(LiveMicrophoneGateway.class,
+        (Object proxy, Method method, Object[] args) -> {
+          if ("goUp".equals(method.getName())) {
+            calls.add("goUp");
+            return notice();
+          }
+          if ("goDown".equals(method.getName())) {
+            calls.add("goDown");
+            return null;
+          }
+          throw new UnsupportedOperationException(method.getName());
+        });
+
+    private final LiveMicCacheService liveMicCacheService = proxy(LiveMicCacheService.class,
+        (Object proxy, Method method, Object[] args) -> {
+          if ("checkEqSelf".equals(method.getName())) {
+            calls.add("checkEqSelf");
+            return false;
+          }
+          if ("checkAvailable".equals(method.getName())) {
+            calls.add("checkAvailable");
+            return true;
+          }
+          throw new UnsupportedOperationException(method.getName());
+        });
+
+    private final TrtcClient trtcClient = proxy(TrtcClient.class,
+        (Object proxy, Method method, Object[] args) -> {
+          if ("createAgoraTokenGoUp".equals(method.getName())) {
+            calls.add("createToken");
+            return ResultResponse.success("publisher-token");
+          }
+          throw new UnsupportedOperationException(method.getName());
+        });
+
+    private final UserAgoraTokenCacheService userAgoraTokenCacheService =
+        new UserAgoraTokenCacheService() {
+          @Override
+          public void remove(Long userId, Long roomId) {
+            calls.add("removeToken");
+          }
+
+          @Override
+          public void save(String agoraToken, Long userId, Long roomId) {
+            calls.add("saveToken");
+            if (cacheException != null) {
+              throw cacheException;
+            }
+          }
+
+          @Override
+          public String getUserAgoraToken(Long userId, Long roomId) {
+            return null;
+          }
+        };
+
+    private final RoomManagerClient roomManagerClient = proxy(RoomManagerClient.class,
+        (Object proxy, Method method, Object[] args) -> {
+          if ("getById".equals(method.getName())) {
+            calls.add("getRoom");
+            return ResultResponse.success(null);
+          }
+          throw new UnsupportedOperationException(method.getName());
+        });
+
+    private final LiveMicGoUpCmdExe exe = new LiveMicGoUpCmdExe(
+        liveMicAppConvertor,
+        liveMicSettingService,
+        liveMicrophoneGateway,
+        liveMicCacheService,
+        trtcClient,
+        userAgoraTokenCacheService,
+        null,
+        null,
+        roomManagerClient,
+        null,
+        null
+    );
+  }
+}

rc-service/rc-service-live/live-application/src/test/java/com/red/circle/live/app/service/AgoraMicMissingConfirmServiceTest.java

@@ -0,0 +1,42 @@
+package com.red.circle.live.app.service;
+
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertFalse;
+import static org.junit.Assert.assertTrue;
+
+import com.red.circle.live.app.service.AgoraMicMissingConfirmService.MissingState;
+import org.junit.Test;
+
+public class AgoraMicMissingConfirmServiceTest {
+
+  @Test
+  public void firstMissingShouldOnlyRecordState() {
+    MissingState state = MissingState.next(null, 1_000L);
+
+    assertFalse(state.shouldCleanup(1_000L));
+    assertEquals("1:1000", state.serialize());
+  }
+
+  @Test
+  public void secondConsecutiveMissingShouldCleanup() {
+    MissingState state = MissingState.next("1:1000", 2_000L);
+
+    assertTrue(state.shouldCleanup(2_000L));
+    assertEquals("2:1000", state.serialize());
+  }
+
+  @Test
+  public void delayedMissingShouldCleanupAfterConfirmWindow() {
+    MissingState state = MissingState.next("1:1000", 6_000L);
+
+    assertTrue(state.shouldCleanup(6_000L));
+  }
+
+  @Test
+  public void malformedStateShouldRestartConfirmWindow() {
+    MissingState state = MissingState.next("bad-value", 3_000L);
+
+    assertFalse(state.shouldCleanup(3_000L));
+    assertEquals("1:3000", state.serialize());
+  }
+}

rc-service/rc-service-live/live-infrastructure/src/main/java/com/red/circle/live/infra/repository/LiveMicrophoneGatewayImpl.java

@@ -164,6 +164,7 @@ public class LiveMicrophoneGatewayImpl implements LiveMicrophoneGateway {
       micOpsLock(roomId, micIndex);
 //      RLock lock   = redissonClient.getLock(roomId.toString()+micIndex.toString()+"room");
 //      lock.lock();
+    boolean micUpdated = false;
     try {
       // 验证用户是否真实存在
       ResponseAssert.notNull(UserErrorCode.USER_INFO_NOT_FOUND, userProfileClient.getByUserId(userId));
@@ -204,12 +205,24 @@ public class LiveMicrophoneGatewayImpl implements LiveMicrophoneGateway {
 
       // 上麦克风
       liveMicCacheService.goUp(roomId, micIndex, liveMicrophone);
+      micUpdated = true;
 
       // 清理活跃用户
       liveMicCacheService.refreshNotActiveUser(roomId);
 
       // 发送通知
       return micUserChangeNotice(roomProfile);
+    } catch (RuntimeException e) {
+      if (micUpdated) {
+        try {
+          liveMicCacheService.goDown(roomId, micIndex);
+          liveMicCacheService.refreshNotActiveUser(roomId);
+        } catch (RuntimeException rollbackException) {
+          log.warn("上麦写入后异常，回滚麦位失败 roomId={}, micIndex={}, userId={}",
+              roomId, micIndex, userId, rollbackException);
+        }
+      }
+      throw e;
     } finally {
 //      lock.unlock();
       micOpsUnlock(roomId, micIndex);