From e9562504837d123954b76fe370539d0d3bde98f6 Mon Sep 17 00:00:00 2001 From: hy001 Date: Sat, 11 Jul 2026 03:20:40 +0800 Subject: [PATCH] fix(live): index mic rooms for reconciliation --- .../scheduler/AgoraMicStateReconcileTask.java | 74 ++-- .../AgoraMicStateReconcileTaskIndexTest.java | 182 ++++++++++ ...aMicStateReconcileTaskRtcProviderTest.java | 1 - .../infra/database/cache/key/LiveMicKey.java | 16 +- .../cache/service/LiveMicCacheService.java | 24 +- .../service/impl/LiveMicCacheServiceImpl.java | 141 ++++++-- .../impl/LiveMicCacheServiceImplTest.java | 323 ++++++++++++++++++ 7 files changed, 696 insertions(+), 65 deletions(-) create mode 100644 rc-service/rc-service-live/live-application/src/test/java/com/red/circle/live/app/scheduler/AgoraMicStateReconcileTaskIndexTest.java create mode 100644 rc-service/rc-service-live/live-infrastructure/src/test/java/com/red/circle/live/infra/database/cache/service/impl/LiveMicCacheServiceImplTest.java diff --git a/rc-service/rc-service-live/live-application/src/main/java/com/red/circle/live/app/scheduler/AgoraMicStateReconcileTask.java b/rc-service/rc-service-live/live-application/src/main/java/com/red/circle/live/app/scheduler/AgoraMicStateReconcileTask.java index 67fe26fa..fe570392 100644 --- a/rc-service/rc-service-live/live-application/src/main/java/com/red/circle/live/app/scheduler/AgoraMicStateReconcileTask.java +++ b/rc-service/rc-service-live/live-application/src/main/java/com/red/circle/live/app/scheduler/AgoraMicStateReconcileTask.java @@ -16,10 +16,9 @@ import java.util.List; import java.util.Objects; import java.util.Set; import java.util.concurrent.TimeUnit; +import java.util.UUID; import lombok.RequiredArgsConstructor; import lombok.extern.slf4j.Slf4j; -import org.springframework.data.redis.core.RedisTemplate; -import org.springframework.data.redis.core.ScanOptions; import org.springframework.scheduling.annotation.Scheduled; import org.springframework.stereotype.Component; @@ -31,11 +30,20 @@ import org.springframework.stereotype.Component; @RequiredArgsConstructor public class AgoraMicStateReconcileTask { - private static final String LOCK_KEY = "agora:mic:reconcile:lock"; - private static final String SEAT_KEY_PREFIX = "LiveMick:SEAT:"; + /** + * v2 锁与旧版整数值/无 token 锁隔离。滚动发布期间旧实例仍会无条件 DEL 旧 key， + * 若复用同名锁就可能误删新实例的 UUID 锁，破坏 compare-delete 的安全保证。 + */ + private static final String LOCK_KEY = "agora:mic:reconcile:lock:v2"; + private static final long LOCK_TTL_MINUTES = 5L; + private static final String RELEASE_LOCK_SCRIPT = """ + if redis.call('get', KEYS[1]) == ARGV[1] then + return redis.call('del', KEYS[1]) + end + return 0 + """; private final RedisService redisService; - private final RedisTemplate redisTemplate; private final LiveMicCacheService liveMicCacheService; private final AgoraRoomStateService agoraRoomStateService; private final AgoraMicMissingConfirmService agoraMicMissingConfirmService; @@ -43,15 +51,19 @@ public class AgoraMicStateReconcileTask { @Scheduled(fixedDelayString = "${red-circle.live.agora-mic-reconcile-delay-ms:30000}") public void reconcileAgoraMicState() { - if (!redisService.setIfAbsent(LOCK_KEY, 1, 25, TimeUnit.SECONDS)) { + String lockToken = UUID.randomUUID().toString(); + // 两台 live 共用同一轮对账。TTL 覆盖正常有界房间集合的完整处理窗口；异常退出仍可自动恢复。 + if (!redisService.setIfAbsent(LOCK_KEY, lockToken, LOCK_TTL_MINUTES, TimeUnit.MINUTES)) { return; } try { - for (Long roomId : scanSeatRoomIds()) { + // 房间候选只来自麦位写链路维护的 Set。禁止为滚动升级或历史补数重新引入全库 SCAN； + // 老版本活跃麦位会在下一次心跳 goUp 时自动 SADD，未再心跳的旧 Hash 已有 120 秒 TTL 自然消失。 + for (Long roomId : liveMicCacheService.listReconcileRoomIds()) { reconcileRoom(roomId); } } finally { - redisService.delete(LOCK_KEY); + releaseLock(lockToken); } } @@ -59,6 +71,13 @@ public class AgoraMicStateReconcileTask { if (Objects.isNull(roomId)) { return; } + + List microphones = liveMicCacheService.listLiveMicrophone(roomId); + if (CollectionUtils.isEmpty(microphones)) { + // Hash TTL 到期和异常中断都可能留下 Set 成员；Lua 会在删除前再次 HLEN，避免误删并发 goUp。 + liveMicCacheService.pruneReconcileRoomIndexIfEmpty(roomId); + return; + } if (!shouldReconcileAgoraRoom(roomId)) { log.debug("声网麦位对账跳过非 Agora 房间 roomId={}", roomId); return; @@ -70,11 +89,6 @@ public class AgoraMicStateReconcileTask { return; } - List microphones = liveMicCacheService.listLiveMicrophone(roomId); - if (CollectionUtils.isEmpty(microphones)) { - return; - } - if (!state.channelExist()) { for (LiveMicrophone microphone : microphones) { cleanupMissingMicrophoneUser(roomId, microphone, state, "reconcile agora channel not exist"); @@ -130,27 +144,21 @@ public class AgoraMicStateReconcileTask { agoraMicStateCleanupService.cleanupUser(roomId, userId, reason); } - private Set scanSeatRoomIds() { - Set roomIds = new HashSet<>(); - ScanOptions options = ScanOptions.scanOptions().match(SEAT_KEY_PREFIX + "*").count(200).build(); - try (var cursor = redisTemplate.scan(options)) { - while (cursor.hasNext()) { - Object key = cursor.next(); - Long roomId = parseRoomId(Objects.toString(key, null)); - if (Objects.nonNull(roomId)) { - roomIds.add(roomId); - } + private void releaseLock(String lockToken) { + try { + Long removed = redisService.execute( + RELEASE_LOCK_SCRIPT, + Long.class, + List.of(LOCK_KEY), + lockToken + ); + if (!Objects.equals(removed, 1L)) { + // TTL 到期后锁可能已由下一实例接管；compare-delete 返回 0 时绝不能删除新持有者的锁。 + log.warn("声网麦位对账锁未由当前实例释放，等待持有者或 TTL 收敛"); } - } catch (Exception e) { - log.error("扫描声网麦位对账 Redis key 失败", e); + } catch (RuntimeException e) { + // 对账是补偿任务；释放异常不应触发无 token DEL，保留锁直到 TTL 是更安全的失败方式。 + log.error("释放声网麦位对账锁失败，等待 TTL 自动释放", e); } - return roomIds; - } - - private Long parseRoomId(String key) { - if (Objects.isNull(key) || !key.startsWith(SEAT_KEY_PREFIX)) { - return null; - } - return DataTypeUtils.toLong(key.substring(SEAT_KEY_PREFIX.length())); } } diff --git a/rc-service/rc-service-live/live-application/src/test/java/com/red/circle/live/app/scheduler/AgoraMicStateReconcileTaskIndexTest.java b/rc-service/rc-service-live/live-application/src/test/java/com/red/circle/live/app/scheduler/AgoraMicStateReconcileTaskIndexTest.java new file mode 100644 index 00000000..ef043919 --- /dev/null +++ b/rc-service/rc-service-live/live-application/src/test/java/com/red/circle/live/app/scheduler/AgoraMicStateReconcileTaskIndexTest.java @@ -0,0 +1,182 @@ +package com.red.circle.live.app.scheduler; + +import static org.junit.Assert.assertEquals; +import static org.junit.Assert.assertFalse; +import static org.junit.Assert.assertNotNull; +import static org.junit.Assert.assertTrue; + +import com.red.circle.component.redis.service.RedisService; +import com.red.circle.live.infra.database.cache.service.LiveMicCacheService; +import java.lang.reflect.Proxy; +import java.util.ArrayList; +import java.util.Collections; +import java.util.LinkedHashSet; +import java.util.List; +import java.util.Objects; +import java.util.Set; +import java.util.concurrent.TimeUnit; +import org.junit.Test; +import org.springframework.data.redis.core.RedisTemplate; +import org.springframework.data.redis.core.ValueOperations; +import org.springframework.data.redis.core.script.RedisScript; + +public class AgoraMicStateReconcileTaskIndexTest { + + private static final String LOCK_KEY = "agora:mic:reconcile:lock:v2"; + + @Test + public void schedulerShouldUseBoundedRoomIndexPruneEmptyHashAndReleaseOwnedLock() { + RecordingRedisTemplate redisTemplate = new RecordingRedisTemplate(true, 1L); + RecordingMicCache micCache = new RecordingMicCache(Set.of(1001L)); + AgoraMicStateReconcileTask task = task(redisTemplate, micCache.proxy()); + + task.reconcileAgoraMicState(); + + assertEquals(List.of(1001L), micCache.listedRoomIds); + assertEquals(List.of(1001L), micCache.prunedRoomIds); + assertEquals(LOCK_KEY, redisTemplate.lockKey); + assertNotNull(redisTemplate.lockToken); + assertEquals(5L, redisTemplate.lockTtl); + assertEquals(TimeUnit.MINUTES, redisTemplate.lockTtlUnit); + assertEquals(redisTemplate.lockToken, redisTemplate.releaseToken); + assertEquals(List.of(LOCK_KEY), redisTemplate.releaseKeys); + assertTrue(redisTemplate.releaseScript.contains("get")); + assertTrue(redisTemplate.releaseScript.contains("del")); + assertFalse(redisTemplate.directDeleteCalled); + } + + @Test + public void schedulerShouldNotReadIndexOrReleaseLockWhenAnotherInstanceOwnsIt() { + RecordingRedisTemplate redisTemplate = new RecordingRedisTemplate(false, 0L); + RecordingMicCache micCache = new RecordingMicCache(Set.of(1001L)); + AgoraMicStateReconcileTask task = task(redisTemplate, micCache.proxy()); + + task.reconcileAgoraMicState(); + + assertTrue(micCache.listedRoomIds.isEmpty()); + assertTrue(micCache.prunedRoomIds.isEmpty()); + assertEquals("", redisTemplate.releaseScript); + assertFalse(redisTemplate.directDeleteCalled); + } + + @Test + public void tokenMismatchShouldNeverFallBackToUnsafeDirectDelete() { + RecordingRedisTemplate redisTemplate = new RecordingRedisTemplate(true, 0L); + RecordingMicCache micCache = new RecordingMicCache(Collections.emptySet()); + + task(redisTemplate, micCache.proxy()).reconcileAgoraMicState(); + + assertNotNull(redisTemplate.releaseToken); + assertFalse(redisTemplate.directDeleteCalled); + } + + private static AgoraMicStateReconcileTask task( + RecordingRedisTemplate redisTemplate, + LiveMicCacheService micCache) { + return new AgoraMicStateReconcileTask( + new RedisService(redisTemplate), + micCache, + null, + null, + null + ); + } + + /** + * 仅实现调度器需要的两个索引动作。返回空麦位可验证残留 Set 成员会走原子 prune， + * 且测试类完全不提供 Redis SCAN 能力。 + */ + private static final class RecordingMicCache { + + private final Set roomIds; + private final List listedRoomIds = new ArrayList<>(); + private final List prunedRoomIds = new ArrayList<>(); + + private RecordingMicCache(Set roomIds) { + this.roomIds = new LinkedHashSet<>(roomIds); + } + + private LiveMicCacheService proxy() { + return (LiveMicCacheService) Proxy.newProxyInstance( + LiveMicCacheService.class.getClassLoader(), + new Class[] {LiveMicCacheService.class}, + (proxy, method, args) -> switch (method.getName()) { + case "listReconcileRoomIds" -> new LinkedHashSet<>(roomIds); + case "listLiveMicrophone" -> { + listedRoomIds.add((Long) args[0]); + yield Collections.emptyList(); + } + case "pruneReconcileRoomIndexIfEmpty" -> { + prunedRoomIds.add((Long) args[0]); + yield true; + } + case "toString" -> "RecordingMicCache"; + case "hashCode" -> System.identityHashCode(proxy); + case "equals" -> proxy == args[0]; + default -> throw new UnsupportedOperationException(method.getName()); + } + ); + } + } + + private static final class RecordingRedisTemplate extends RedisTemplate { + + private final boolean acquireLock; + private final Long releaseResult; + private String lockKey; + private Object lockToken; + private long lockTtl; + private TimeUnit lockTtlUnit; + private String releaseScript = ""; + private List releaseKeys = List.of(); + private Object releaseToken; + private boolean directDeleteCalled; + + private RecordingRedisTemplate(boolean acquireLock, Long releaseResult) { + this.acquireLock = acquireLock; + this.releaseResult = releaseResult; + } + + @Override + @SuppressWarnings("unchecked") + public ValueOperations opsForValue() { + return (ValueOperations) Proxy.newProxyInstance( + ValueOperations.class.getClassLoader(), + new Class[] {ValueOperations.class}, + (proxy, method, args) -> { + if (Objects.equals("setIfAbsent", method.getName()) && args.length == 4) { + lockKey = Objects.toString(args[0]); + lockToken = args[1]; + lockTtl = ((Number) args[2]).longValue(); + lockTtlUnit = (TimeUnit) args[3]; + return acquireLock; + } + if (method.getDeclaringClass().equals(Object.class)) { + return switch (method.getName()) { + case "toString" -> "RecordingValueOperations"; + case "hashCode" -> System.identityHashCode(proxy); + case "equals" -> proxy == args[0]; + default -> throw new UnsupportedOperationException(method.getName()); + }; + } + throw new UnsupportedOperationException(method.getName()); + } + ); + } + + @Override + @SuppressWarnings("unchecked") + public T execute(RedisScript script, List keys, Object... args) { + releaseScript = script.getScriptAsString(); + releaseKeys = List.copyOf(keys); + releaseToken = args[0]; + return (T) releaseResult; + } + + @Override + public Boolean delete(String key) { + directDeleteCalled = true; + return Boolean.TRUE; + } + } +} diff --git a/rc-service/rc-service-live/live-application/src/test/java/com/red/circle/live/app/scheduler/AgoraMicStateReconcileTaskRtcProviderTest.java b/rc-service/rc-service-live/live-application/src/test/java/com/red/circle/live/app/scheduler/AgoraMicStateReconcileTaskRtcProviderTest.java index a69c0000..c2c4e3e0 100644 --- a/rc-service/rc-service-live/live-application/src/test/java/com/red/circle/live/app/scheduler/AgoraMicStateReconcileTaskRtcProviderTest.java +++ b/rc-service/rc-service-live/live-application/src/test/java/com/red/circle/live/app/scheduler/AgoraMicStateReconcileTaskRtcProviderTest.java @@ -32,7 +32,6 @@ public class AgoraMicStateReconcileTaskRtcProviderTest { null, null, null, - null, null); } diff --git a/rc-service/rc-service-live/live-infrastructure/src/main/java/com/red/circle/live/infra/database/cache/key/LiveMicKey.java b/rc-service/rc-service-live/live-infrastructure/src/main/java/com/red/circle/live/infra/database/cache/key/LiveMicKey.java index 30187ae1..7dd1ac32 100644 --- a/rc-service/rc-service-live/live-infrastructure/src/main/java/com/red/circle/live/infra/database/cache/key/LiveMicKey.java +++ b/rc-service/rc-service-live/live-infrastructure/src/main/java/com/red/circle/live/infra/database/cache/key/LiveMicKey.java @@ -14,10 +14,18 @@ public enum LiveMicKey implements RedisKeys { */ LOCK, - /** - * 位置. - */ - SEAT; + /** + * 位置. + */ + SEAT, + + /** + * 当前存在麦位数据、需要参与 RTC 对账的房间集合. + * + *

它是 {@link #SEAT} Hash 的有界辅助索引，不是第二份麦位状态。成员只保存 roomId， + * 对账时仍以对应 SEAT Hash 为事实；索引残留会由原子 compare-prune 收敛。 + */ + SEAT_ROOM_INDEX; @Override diff --git a/rc-service/rc-service-live/live-infrastructure/src/main/java/com/red/circle/live/infra/database/cache/service/LiveMicCacheService.java b/rc-service/rc-service-live/live-infrastructure/src/main/java/com/red/circle/live/infra/database/cache/service/LiveMicCacheService.java index 57a5c002..537d4842 100644 --- a/rc-service/rc-service-live/live-infrastructure/src/main/java/com/red/circle/live/infra/database/cache/service/LiveMicCacheService.java +++ b/rc-service/rc-service-live/live-infrastructure/src/main/java/com/red/circle/live/infra/database/cache/service/LiveMicCacheService.java @@ -1,8 +1,9 @@ package com.red.circle.live.infra.database.cache.service; import com.red.circle.live.domain.live.LiveMicrophone; -import com.red.circle.tool.core.func.VoidConsumer; -import java.util.List; +import com.red.circle.tool.core.func.VoidConsumer; +import java.util.List; +import java.util.Set; /** * 麦克风用户缓存服务. @@ -60,7 +61,24 @@ public interface LiveMicCacheService { /** * 下掉所有. */ - void goDownAll(Long roomId); + void goDownAll(Long roomId); + + /** + * 返回当前需要参与 RTC 对账的房间 id. + * + *

这里只读取 LiveMicCacheService 在写麦位时维护的有界 Set，禁止通过 + * {@code SCAN LiveMick:SEAT:*} 从共享 Redis DB 反查房间。 + */ + Set listReconcileRoomIds(); + + /** + * 仅当房间 SEAT Hash 已为空时从对账索引移除 roomId. + * + *

判断和删除必须在一条 Lua 内完成，避免并发 goUp 已写入新麦位后被旧清理误删索引。 + * + * @return true 表示本次确实移除了残留索引成员 + */ + boolean pruneReconcileRoomIndexIfEmpty(Long roomId); /** * 验证是否是自己在麦克风. diff --git a/rc-service/rc-service-live/live-infrastructure/src/main/java/com/red/circle/live/infra/database/cache/service/impl/LiveMicCacheServiceImpl.java b/rc-service/rc-service-live/live-infrastructure/src/main/java/com/red/circle/live/infra/database/cache/service/impl/LiveMicCacheServiceImpl.java index 2b0dad6e..64d85c44 100644 --- a/rc-service/rc-service-live/live-infrastructure/src/main/java/com/red/circle/live/infra/database/cache/service/impl/LiveMicCacheServiceImpl.java +++ b/rc-service/rc-service-live/live-infrastructure/src/main/java/com/red/circle/live/infra/database/cache/service/impl/LiveMicCacheServiceImpl.java @@ -5,11 +5,13 @@ import com.red.circle.component.redis.service.RedisService; import com.red.circle.live.domain.live.LiveMicrophone; import com.red.circle.live.infra.database.cache.key.LiveMicKey; import com.red.circle.live.infra.database.cache.service.LiveMicCacheService; -import com.red.circle.tool.core.collection.CollectionUtils; -import com.red.circle.tool.core.func.VoidConsumer; -import java.util.List; -import java.util.Objects; -import java.util.concurrent.TimeUnit; +import com.red.circle.tool.core.collection.CollectionUtils; +import com.red.circle.tool.core.func.VoidConsumer; +import java.util.LinkedHashSet; +import java.util.List; +import java.util.Objects; +import java.util.Set; +import java.util.concurrent.TimeUnit; import lombok.RequiredArgsConstructor; import lombok.extern.slf4j.Slf4j; import org.springframework.stereotype.Service; @@ -22,9 +24,22 @@ import org.springframework.stereotype.Service; @Slf4j @Service @RequiredArgsConstructor -public class LiveMicCacheServiceImpl implements LiveMicCacheService { - - private final RedisService redisService; +public class LiveMicCacheServiceImpl implements LiveMicCacheService { + + /** + * 删除索引前在 Redis 内重新检查 SEAT Hash，保证判断与 SREM 不留并发窗口。 + * + *

goUp 使用 HSET -> SADD。若 goUp 先于本脚本执行，HLEN 会阻止误删；若 goUp 后执行， + * 它最终的 SADD 会恢复索引。因此无论命令如何串行化，都不会留下“有麦位但无索引”的稳态。 + */ + private static final String PRUNE_ROOM_INDEX_IF_EMPTY_SCRIPT = """ + if redis.call('hlen', KEYS[1]) == 0 then + return redis.call('srem', KEYS[2], ARGV[1]) + end + return 0 + """; + + private final RedisService redisService; @Override public boolean checkLockMic(Long roomId, Integer index) { @@ -67,23 +82,70 @@ public class LiveMicCacheServiceImpl implements LiveMicCacheService { redisService.hashPut(LiveMicKey.SEAT.getKey(roomId), Objects.toString(index), microphone); - } + // HSET 必须先于 SADD：并发下麦的 compare-prune 与 HSET 在 Redis 内有确定顺序， + // 随后的幂等 SADD 会保证任何成功写入的麦位最终都可被定时对账发现。 + addReconcileRoomIndex(roomId); + } @Override - public void goDown(Long roomId, Integer index) { - redisService.hashDelete(LiveMicKey.SEAT.getKey(roomId), Objects.toString(index)); - } + public void goDown(Long roomId, Integer index) { + redisService.hashDelete(LiveMicKey.SEAT.getKey(roomId), Objects.toString(index)); + pruneReconcileRoomIndexIfEmpty(roomId); + } @Override - public void goDown(Long roomId, List index) { - redisService.hashDelete(LiveMicKey.SEAT.getKey(roomId), - index.stream().map(Objects::toString).toArray(String[]::new)); - } + public void goDown(Long roomId, List index) { + redisService.hashDelete(LiveMicKey.SEAT.getKey(roomId), + index.stream().map(Objects::toString).toArray(String[]::new)); + pruneReconcileRoomIndexIfEmpty(roomId); + } @Override - public void goDownAll(Long roomId) { - redisService.delete(LiveMicKey.SEAT.getKey(roomId)); - } + public void goDownAll(Long roomId) { + redisService.delete(LiveMicKey.SEAT.getKey(roomId)); + // 即使与新 goUp 并发，也必须复用 HLEN compare-prune，不能无条件 SREM。 + pruneReconcileRoomIndexIfEmpty(roomId); + } + + @Override + public Set listReconcileRoomIds() { + Set members = redisService.setGet(reconcileRoomIndexKey()); + if (CollectionUtils.isEmpty(members)) { + return new LinkedHashSet<>(); + } + + Set roomIds = new LinkedHashSet<>(); + for (Object member : members) { + Long roomId = parseRoomId(member); + if (Objects.isNull(roomId)) { + // 只有合法 roomId 才能定位 SEAT Hash；损坏成员无法 compare-prune，直接从辅助索引剔除。 + redisService.setRemove(reconcileRoomIndexKey(), member); + continue; + } + roomIds.add(roomId); + } + return roomIds; + } + + @Override + public boolean pruneReconcileRoomIndexIfEmpty(Long roomId) { + if (Objects.isNull(roomId)) { + return false; + } + try { + Long removed = redisService.execute( + PRUNE_ROOM_INDEX_IF_EMPTY_SCRIPT, + Long.class, + List.of(LiveMicKey.SEAT.getKey(roomId), reconcileRoomIndexKey()), + roomId.toString() + ); + return Objects.nonNull(removed) && removed > 0L; + } catch (RuntimeException e) { + // 麦位 Hash 已先完成主操作；辅助索引失败只能造成可重试的残留，不能让客户端误以为上下麦失败。 + log.error("原子清理麦位房间索引失败, roomId={}", roomId, e); + return false; + } + } @Override public boolean checkEqSelf(Long roomId, Integer index, Long userId) { @@ -130,6 +192,8 @@ public class LiveMicCacheServiceImpl implements LiveMicCacheService { List liveMicrophone = listLiveMicrophone(roomId); if (CollectionUtils.isEmpty(liveMicrophone)) { + // Hash 可能已经因 120 秒 TTL 自然过期；这里同步收敛 Set 中的残留成员。 + pruneReconcileRoomIndexIfEmpty(roomId); return CollectionUtils.newArrayList(); } @@ -148,6 +212,9 @@ public class LiveMicCacheServiceImpl implements LiveMicCacheService { Long size = redisService.hashDelete(LiveMicKey.SEAT.getKey(roomId), index); log.info("清除不活跃的用户结果 = {} ", size); + // 删除最后一个不活跃麦位时立即移除索引；若并发上麦，Lua 的 HLEN 会保留新房间成员。 + pruneReconcileRoomIndexIfEmpty(roomId); + if (Objects.isNull(size) || size <= 0) { return CollectionUtils.newArrayList(); } @@ -155,8 +222,34 @@ public class LiveMicCacheServiceImpl implements LiveMicCacheService { } - private void setExpire(Long roomId) { - redisService.expire(LiveMicKey.SEAT.getKey(roomId), 120, TimeUnit.SECONDS); - } - -} + private void setExpire(Long roomId) { + redisService.expire(LiveMicKey.SEAT.getKey(roomId), 120, TimeUnit.SECONDS); + } + + private String reconcileRoomIndexKey() { + return LiveMicKey.SEAT_ROOM_INDEX.getKey(); + } + + private void addReconcileRoomIndex(Long roomId) { + try { + // Set 成员和 Lua ARGV 统一使用十进制字符串，避免不同 Redis serializer 对 Long/String + // 产生不同字节，导致 compare-prune 看似成功却删不到实际成员。 + redisService.setAdd(reconcileRoomIndexKey(), roomId.toString()); + } catch (RuntimeException e) { + // HSET 已经成功时不回滚真实麦位；活跃用户下一次心跳 goUp 会继续幂等补写索引。 + log.error("维护麦位房间索引失败，等待心跳重试, roomId={}", roomId, e); + } + } + + private Long parseRoomId(Object member) { + if (Objects.isNull(member)) { + return null; + } + try { + return Long.valueOf(member.toString()); + } catch (NumberFormatException ignored) { + return null; + } + } + +} diff --git a/rc-service/rc-service-live/live-infrastructure/src/test/java/com/red/circle/live/infra/database/cache/service/impl/LiveMicCacheServiceImplTest.java b/rc-service/rc-service-live/live-infrastructure/src/test/java/com/red/circle/live/infra/database/cache/service/impl/LiveMicCacheServiceImplTest.java new file mode 100644 index 00000000..bf5e0bdc --- /dev/null +++ b/rc-service/rc-service-live/live-infrastructure/src/test/java/com/red/circle/live/infra/database/cache/service/impl/LiveMicCacheServiceImplTest.java @@ -0,0 +1,323 @@ +package com.red.circle.live.infra.database.cache.service.impl; + +import static org.junit.Assert.assertEquals; +import static org.junit.Assert.assertFalse; +import static org.junit.Assert.assertTrue; + +import com.red.circle.component.redis.service.RedisService; +import com.red.circle.live.domain.live.LiveMicrophone; +import com.red.circle.live.infra.database.cache.key.LiveMicKey; +import com.red.circle.tool.core.json.JacksonUtils; +import java.lang.reflect.Proxy; +import java.util.ArrayList; +import java.util.LinkedHashMap; +import java.util.LinkedHashSet; +import java.util.List; +import java.util.Map; +import java.util.Objects; +import java.util.Set; +import java.util.concurrent.TimeUnit; +import org.junit.Test; +import org.springframework.data.redis.core.HashOperations; +import org.springframework.data.redis.core.RedisTemplate; +import org.springframework.data.redis.core.SetOperations; +import org.springframework.data.redis.core.script.RedisScript; + +public class LiveMicCacheServiceImplTest { + + private static final Long ROOM_ID = 1001L; + private static final String SEAT_KEY = "LiveMick:SEAT:1001"; + private static final String ROOM_INDEX_KEY = "LiveMick:SEAT_ROOM_INDEX"; + + @Test + public void goUpShouldWriteSeatBeforeIndexAndHeartbeatShouldBackfillLegacyRoom() { + RecordingRedisTemplate redisTemplate = new RecordingRedisTemplate(); + LiveMicCacheServiceImpl service = service(redisTemplate); + + // 模拟滚动升级前已存在、但尚未进入新 Set 的麦位 Hash。下一次正常心跳仍会调用 goUp。 + redisTemplate.seedSeat(ROOM_ID, 0, microphone(0, System.currentTimeMillis())); + assertTrue(redisTemplate.indexMembers().isEmpty()); + + service.goUp(ROOM_ID, 0, microphone(0, System.currentTimeMillis())); + + assertEquals(List.of("HSET " + SEAT_KEY + " 0", "SADD " + ROOM_INDEX_KEY + " 1001"), + redisTemplate.events.subList(0, 2)); + assertEquals(Set.of(ROOM_ID.toString()), redisTemplate.indexMembers()); + assertEquals(Set.of(ROOM_ID), service.listReconcileRoomIds()); + } + + @Test + public void goDownShouldKeepIndexUntilLastSeatIsRemoved() { + RecordingRedisTemplate redisTemplate = new RecordingRedisTemplate(); + LiveMicCacheServiceImpl service = service(redisTemplate); + service.goUp(ROOM_ID, 0, microphone(0, System.currentTimeMillis())); + service.goUp(ROOM_ID, 1, microphone(1, System.currentTimeMillis())); + + service.goDown(ROOM_ID, 0); + assertEquals(Set.of(ROOM_ID), service.listReconcileRoomIds()); + + service.goDown(ROOM_ID, 1); + assertTrue(service.listReconcileRoomIds().isEmpty()); + assertEquals(0L, redisTemplate.hashSize(SEAT_KEY)); + } + + @Test + public void comparePruneShouldNotRemoveIndexWhenConcurrentGoUpWritesNewSeat() { + RecordingRedisTemplate redisTemplate = new RecordingRedisTemplate(); + LiveMicCacheServiceImpl service = service(redisTemplate); + service.goUp(ROOM_ID, 0, microphone(0, System.currentTimeMillis())); + + // 精确模拟 HDEL 已完成、Lua 尚未执行时另一个请求上麦。Lua 内的 HLEN 必须看到新 Hash。 + redisTemplate.beforeNextScript = () -> + service.goUp(ROOM_ID, 1, microphone(1, System.currentTimeMillis())); + service.goDown(ROOM_ID, 0); + + assertEquals(Set.of(ROOM_ID), service.listReconcileRoomIds()); + assertEquals(1L, redisTemplate.hashSize(SEAT_KEY)); + assertTrue(redisTemplate.scriptCalls.get(0).script().contains("hlen")); + assertTrue(redisTemplate.scriptCalls.get(0).script().contains("srem")); + assertEquals(List.of(SEAT_KEY, ROOM_INDEX_KEY), redisTemplate.scriptCalls.get(0).keys()); + assertEquals(List.of(ROOM_ID.toString()), redisTemplate.scriptCalls.get(0).args()); + } + + @Test + public void batchAndAllGoDownShouldPruneIndexOnlyThroughCompareScript() { + RecordingRedisTemplate redisTemplate = new RecordingRedisTemplate(); + LiveMicCacheServiceImpl service = service(redisTemplate); + service.goUp(ROOM_ID, 0, microphone(0, System.currentTimeMillis())); + service.goUp(ROOM_ID, 1, microphone(1, System.currentTimeMillis())); + + service.goDown(ROOM_ID, List.of(0, 1)); + assertTrue(service.listReconcileRoomIds().isEmpty()); + + service.goUp(ROOM_ID, 2, microphone(2, System.currentTimeMillis())); + service.goDownAll(ROOM_ID); + assertTrue(service.listReconcileRoomIds().isEmpty()); + assertEquals(2, redisTemplate.scriptCalls.size()); + } + + @Test + public void inactiveCleanupShouldKeepLegacyTtlAndRemoveLastRoomIndex() { + RecordingRedisTemplate redisTemplate = new RecordingRedisTemplate(); + LiveMicCacheServiceImpl service = service(redisTemplate); + long expiredRefreshTime = System.currentTimeMillis() - TimeUnit.SECONDS.toMillis(121L); + service.goUp(ROOM_ID, 0, microphone(0, expiredRefreshTime)); + + List removed = service.refreshNotActiveUserAndReturn(ROOM_ID); + + assertEquals(1, removed.size()); + assertEquals(List.of(new ExpireCall(SEAT_KEY, 120L, TimeUnit.SECONDS)), + redisTemplate.expireCalls); + assertTrue(service.listReconcileRoomIds().isEmpty()); + } + + @Test + public void listReconcileRoomIdsShouldDiscardMalformedMembersWithoutScanningKeyspace() { + RecordingRedisTemplate redisTemplate = new RecordingRedisTemplate(); + LiveMicCacheServiceImpl service = service(redisTemplate); + redisTemplate.setMembers.computeIfAbsent(ROOM_INDEX_KEY, ignored -> new LinkedHashSet<>()) + .addAll(List.of("bad-room-id", ROOM_ID.toString())); + + assertEquals(Set.of(ROOM_ID), service.listReconcileRoomIds()); + assertEquals(Set.of(ROOM_ID.toString()), redisTemplate.indexMembers()); + // RecordingRedisTemplate 只实现 Hash/Set/expire/script；若生产代码重新引入 SCAN，测试会直接失败。 + assertFalse(redisTemplate.events.stream().anyMatch(event -> event.startsWith("SCAN"))); + } + + @Test + public void auxiliaryIndexFailureShouldNotReverseCompletedSeatMutation() { + RecordingRedisTemplate redisTemplate = new RecordingRedisTemplate(); + LiveMicCacheServiceImpl service = service(redisTemplate); + redisTemplate.setFailure = new IllegalStateException("set unavailable"); + + service.goUp(ROOM_ID, 0, microphone(0, System.currentTimeMillis())); + assertEquals(1L, redisTemplate.hashSize(SEAT_KEY)); + + redisTemplate.setFailure = null; + service.goUp(ROOM_ID, 0, microphone(0, System.currentTimeMillis())); + redisTemplate.scriptFailure = new IllegalStateException("script unavailable"); + service.goDown(ROOM_ID, 0); + assertEquals(0L, redisTemplate.hashSize(SEAT_KEY)); + } + + private static LiveMicCacheServiceImpl service(RecordingRedisTemplate redisTemplate) { + return new LiveMicCacheServiceImpl(new RedisService(redisTemplate)); + } + + private static LiveMicrophone microphone(int index, long refreshTime) { + return new LiveMicrophone().setMicIndex(index).setRefreshTime(refreshTime); + } + + private record ExpireCall(String key, long ttl, TimeUnit unit) { + } + + private record ScriptCall(String script, List keys, List args) { + } + + /** + * 用最小内存模型执行 RedisService 会使用的命令，并真实模拟 compare-prune Lua 的原子结果。 + * 未实现的命令一律抛错，防止测试悄悄放过新的全库遍历或非预期读写。 + */ + private static final class RecordingRedisTemplate extends RedisTemplate { + + private final Map> hashes = new LinkedHashMap<>(); + private final Map> setMembers = new LinkedHashMap<>(); + private final List events = new ArrayList<>(); + private final List expireCalls = new ArrayList<>(); + private final List scriptCalls = new ArrayList<>(); + private Runnable beforeNextScript; + private RuntimeException setFailure; + private RuntimeException scriptFailure; + + @Override + @SuppressWarnings("unchecked") + public HashOperations opsForHash() { + return (HashOperations) proxy(HashOperations.class, (method, args) -> { + String key = Objects.toString(args[0]); + LinkedHashMap hash = hashes.computeIfAbsent(key, + ignored -> new LinkedHashMap<>()); + return switch (method) { + case "put" -> { + hash.put(args[1], args[2]); + events.add("HSET " + key + " " + args[1]); + yield null; + } + case "delete" -> { + long removed = 0L; + Object[] fields = (Object[]) args[1]; + for (Object field : fields) { + if (hash.remove(field) != null) { + removed++; + } + } + events.add("HDEL " + key); + yield removed; + } + case "size" -> (long) hash.size(); + case "values" -> new ArrayList<>(hash.values()); + case "get" -> hash.get(args[1]); + case "hasKey" -> hash.containsKey(args[1]); + default -> throw new UnsupportedOperationException(method); + }; + }); + } + + @Override + @SuppressWarnings("unchecked") + public SetOperations opsForSet() { + return proxy(SetOperations.class, (method, args) -> { + String key = Objects.toString(args[0]); + LinkedHashSet members = setMembers.computeIfAbsent(key, + ignored -> new LinkedHashSet<>()); + return switch (method) { + case "add" -> { + if (setFailure != null) { + throw setFailure; + } + long added = 0L; + Object[] values = (Object[]) args[1]; + for (Object value : values) { + if (members.add(value)) { + added++; + } + events.add("SADD " + key + " " + value); + } + yield added; + } + case "members" -> new LinkedHashSet<>(members); + case "remove" -> { + long removed = 0L; + Object[] values = (Object[]) args[1]; + for (Object value : values) { + if (members.remove(value)) { + removed++; + } + } + events.add("SREM " + key); + yield removed; + } + default -> throw new UnsupportedOperationException(method); + }; + }); + } + + @Override + public Boolean delete(String key) { + events.add("DEL " + key); + return hashes.remove(key) != null || setMembers.remove(key) != null; + } + + @Override + public Boolean expire(String key, long timeout, TimeUnit unit) { + expireCalls.add(new ExpireCall(key, timeout, unit)); + return Boolean.TRUE; + } + + @Override + @SuppressWarnings("unchecked") + public T execute(RedisScript script, List keys, Object... args) { + if (scriptFailure != null) { + throw scriptFailure; + } + Runnable hook = beforeNextScript; + beforeNextScript = null; + if (hook != null) { + hook.run(); + } + scriptCalls.add(new ScriptCall( + script.getScriptAsString(), + List.copyOf(keys), + List.of(args) + )); + + String seatKey = keys.get(0); + String indexKey = keys.get(1); + long removed = 0L; + if (hashSize(seatKey) == 0L) { + LinkedHashSet members = setMembers.computeIfAbsent(indexKey, + ignored -> new LinkedHashSet<>()); + if (members.remove(args[0])) { + removed = 1L; + } + } + events.add("LUA compare-prune " + seatKey); + return (T) Long.valueOf(removed); + } + + private void seedSeat(Long roomId, int index, LiveMicrophone microphone) { + String key = LiveMicKey.SEAT.getKey(roomId); + hashes.computeIfAbsent(key, ignored -> new LinkedHashMap<>()) + .put(Integer.toString(index), JacksonUtils.toJson(microphone)); + } + + private long hashSize(String key) { + return hashes.getOrDefault(key, new LinkedHashMap<>()).size(); + } + + private Set indexMembers() { + return new LinkedHashSet<>(setMembers.getOrDefault(ROOM_INDEX_KEY, new LinkedHashSet<>())); + } + + @SuppressWarnings("unchecked") + private static T proxy(Class type, Operation operation) { + return (T) Proxy.newProxyInstance(type.getClassLoader(), new Class[] {type}, + (proxy, method, args) -> { + if (method.getDeclaringClass().equals(Object.class)) { + return switch (method.getName()) { + case "toString" -> type.getSimpleName() + "RecordingProxy"; + case "hashCode" -> System.identityHashCode(proxy); + case "equals" -> proxy == args[0]; + default -> throw new UnsupportedOperationException(method.getName()); + }; + } + return operation.invoke(method.getName(), args == null ? new Object[0] : args); + }); + } + } + + @FunctionalInterface + private interface Operation { + + Object invoke(String method, Object[] args); + } +}