From 02d9351b1a014ad8919193faeb225c492624bf72 Mon Sep 17 00:00:00 2001 From: zhx Date: Tue, 21 Jul 2026 13:50:45 +0800 Subject: [PATCH] fix(room): reconcile RTC mute from latest state --- .../internal/room/service/mic_test.go | 165 +++++++++++++++++- .../internal/room/service/moderation.go | 9 +- .../internal/room/service/outbox_worker.go | 26 ++- .../service/outbox_worker_internal_test.go | 34 ++++ .../internal/room/service/rtc_event.go | 7 +- .../internal/room/service/rtc_moderation.go | 147 +++++++++++++++- .../internal/room/service/service.go | 9 + 7 files changed, 372 insertions(+), 25 deletions(-) diff --git a/services/room-service/internal/room/service/mic_test.go b/services/room-service/internal/room/service/mic_test.go index 178c6350..f539f36d 100644 --- a/services/room-service/internal/room/service/mic_test.go +++ b/services/room-service/internal/room/service/mic_test.go @@ -2,6 +2,7 @@ package service_test import ( "context" + "errors" "fmt" "sync" "testing" @@ -20,13 +21,14 @@ import ( type recordingRTCUserAudioBlocker struct { mu sync.Mutex calls []bool + err error } func (b *recordingRTCUserAudioBlocker) SetUserAudioBlockedByStrRoomID(_ context.Context, _ string, _ int64, blocked bool) error { b.mu.Lock() defer b.mu.Unlock() b.calls = append(b.calls, blocked) - return nil + return b.err } func (b *recordingRTCUserAudioBlocker) snapshot() []bool { @@ -100,6 +102,18 @@ func TestMutedUserCannotSpeakOrMicUpUntilUnmuted(t *testing.T) { if !unmutedSpeak.GetAllowed() || unmutedSpeak.GetReason() != "" { t.Fatalf("unmuted user must recover public chat permission: %+v", unmutedSpeak) } + if _, err := svc.MuteUser(ctx, &roomv1.MuteUserRequest{ + Meta: rocketMeta(roomID, ownerID, "mute-user-guard"), + TargetUserId: mutedUserID, + Muted: true, + }); err != nil { + t.Fatalf("old mute command replay failed: %v", err) + } + if got := rtcBlocker.snapshot(); len(got) != 3 || got[2] { + // pipeline 会向重放返回第一次 mute 的历史 result.snapshot；RTC 投影 + // 必须忽略它并重读 newer unmute，否则重试会把云端永久回退为禁音。 + t.Fatalf("old mute replay must reconcile latest unmuted state: %+v", got) + } upResp, err := svc.MicUp(ctx, &roomv1.MicUpRequest{ Meta: rocketMeta(roomID, mutedUserID, "unmuted-mic-up"), SeatNo: 2, @@ -174,8 +188,9 @@ func TestMutedUserCannotSpeakOrMicUpUntilUnmuted(t *testing.T) { } directCalls := rtcBlocker.snapshot() - if len(directCalls) != 5 || !directCalls[0] || directCalls[1] || !directCalls[2] || !directCalls[3] || directCalls[4] { - // 四次 MuteUser 分别投影 true/false/true/false；中间被篡改客户端触发的 + if len(directCalls) != 6 || !directCalls[0] || directCalls[1] || directCalls[2] || !directCalls[3] || !directCalls[4] || directCalls[5] { + // 四次新 MuteUser 分别投影 true/false/true/false，旧 mute 重放必须仍投影 false； + // 中间被篡改客户端触发的 // audio_started 必须额外 re-block=true，因此按调用顺序断言服务端防线完整。 t.Fatalf("unexpected direct RTC audio block sequence: %+v", directCalls) } @@ -196,6 +211,150 @@ func TestMutedUserCannotSpeakOrMicUpUntilUnmuted(t *testing.T) { } } +type blockingRTCUserAudioBlocker struct { + mu sync.Mutex + calls []bool + blocked bool + firstStarted chan struct{} + releaseFirst chan struct{} +} + +func newBlockingRTCUserAudioBlocker() *blockingRTCUserAudioBlocker { + return &blockingRTCUserAudioBlocker{firstStarted: make(chan struct{}), releaseFirst: make(chan struct{})} +} + +func (b *blockingRTCUserAudioBlocker) SetUserAudioBlockedByStrRoomID(ctx context.Context, _ string, _ int64, blocked bool) error { + b.mu.Lock() + b.calls = append(b.calls, blocked) + first := len(b.calls) == 1 + b.mu.Unlock() + if first { + close(b.firstStarted) + select { + case <-ctx.Done(): + return ctx.Err() + case <-b.releaseFirst: + } + } + b.mu.Lock() + b.blocked = blocked + b.mu.Unlock() + return nil +} + +func (b *blockingRTCUserAudioBlocker) snapshot() ([]bool, bool) { + b.mu.Lock() + defer b.mu.Unlock() + return append([]bool(nil), b.calls...), b.blocked +} + +func TestRTCUserAudioReconcileConvergesSlowMuteAndNewerUnmute(t *testing.T) { + ctx := context.Background() + repository := mysqltest.NewRepository(t) + now := &fixedRoomRocketClock{now: time.Date(2026, 7, 21, 9, 0, 0, 0, time.UTC)} + rtcBlocker := newBlockingRTCUserAudioBlocker() + svc := roomservice.New(roomservice.Config{ + NodeID: "node-muted-rtc-race-test", + LeaseTTL: 10 * time.Second, + SnapshotEveryN: 1, + Clock: now, + RTCUserAudioBlocker: rtcBlocker, + }, router.NewMemoryDirectory(), repository, &rocketTestWallet{}, integration.NewNoopRoomEventPublisher(), integration.NewNoopOutboxPublisher()) + + roomID := "room-muted-rtc-race" + ownerID := int64(8491) + targetUserID := int64(8492) + createRocketRoom(t, ctx, svc, roomID, ownerID, 9101) + joinRocketRoom(t, ctx, svc, roomID, targetUserID) + + muteDone := make(chan error, 1) + go func() { + _, err := svc.MuteUser(ctx, &roomv1.MuteUserRequest{ + Meta: rocketMeta(roomID, ownerID, "slow-mute"), + TargetUserId: targetUserID, + Muted: true, + }) + muteDone <- err + }() + select { + case <-rtcBlocker.firstStarted: + case <-time.After(2 * time.Second): + t.Fatal("slow mute never reached RTC blocker") + } + + unmuteDone := make(chan error, 1) + go func() { + _, err := svc.MuteUser(ctx, &roomv1.MuteUserRequest{ + Meta: rocketMeta(roomID, ownerID, "fast-unmute"), + TargetUserId: targetUserID, + Muted: false, + }) + unmuteDone <- err + }() + deadline := time.Now().Add(2 * time.Second) + for { + permission, err := svc.CheckSpeakPermission(ctx, &roomv1.CheckSpeakPermissionRequest{RoomId: roomID, UserId: targetUserID, AppCode: appcode.Default}) + if err == nil && permission.GetAllowed() { + break + } + if time.Now().After(deadline) { + t.Fatalf("newer unmute did not commit while slow RTC mute was in flight: permission=%+v err=%v", permission, err) + } + time.Sleep(5 * time.Millisecond) + } + close(rtcBlocker.releaseFirst) + if err := <-muteDone; err != nil { + t.Fatalf("slow mute request failed: %v", err) + } + if err := <-unmuteDone; err != nil { + t.Fatalf("newer unmute request failed: %v", err) + } + calls, blocked := rtcBlocker.snapshot() + if blocked || len(calls) < 2 || !calls[0] || calls[len(calls)-1] { + t.Fatalf("RTC projection must finish at newer unmuted state: calls=%+v blocked=%t", calls, blocked) + } +} + +func TestRTCUserAudioFailureDoesNotBlockRoomMuteOutboxPublisher(t *testing.T) { + ctx := context.Background() + repository := mysqltest.NewRepository(t) + now := &fixedRoomRocketClock{now: time.Date(2026, 7, 21, 10, 0, 0, 0, time.UTC)} + rtcBlocker := &recordingRTCUserAudioBlocker{err: errors.New("rtc unavailable")} + outboxPublisher := newRecordingRoomDirectIMPublisher() + svc := roomservice.New(roomservice.Config{ + NodeID: "node-muted-rtc-outbox-test", + LeaseTTL: 10 * time.Second, + SnapshotEveryN: 1, + Clock: now, + RTCUserAudioBlocker: rtcBlocker, + }, router.NewMemoryDirectory(), repository, &rocketTestWallet{}, integration.NewNoopRoomEventPublisher(), outboxPublisher) + + roomID := "room-muted-rtc-outbox" + ownerID := int64(8493) + targetUserID := int64(8494) + createRocketRoom(t, ctx, svc, roomID, ownerID, 9101) + joinRocketRoom(t, ctx, svc, roomID, targetUserID) + if _, err := svc.MuteUser(ctx, &roomv1.MuteUserRequest{ + Meta: rocketMeta(roomID, ownerID, "mute-with-rtc-failure"), + TargetUserId: targetUserID, + Muted: true, + }); err != nil { + t.Fatalf("durable room mute must succeed despite direct RTC failure: %v", err) + } + if err := svc.ProcessPendingOutbox(ctx, roomservice.OutboxWorkerOptions{ + PublishTimeout: time.Second, + BatchSize: 100, + MaxRetryCount: 3, + InitialBackoff: time.Second, + MaxBackoff: time.Second, + }); err != nil { + t.Fatalf("process room outbox with RTC failure: %v", err) + } + if outboxPublisher.counts()["RoomUserMuted"] != 1 { + t.Fatalf("RTC failure must not prevent generic RoomUserMuted publication: counts=%+v", outboxPublisher.counts()) + } +} + func TestMicUpReturnsExistingSeatForDuplicateUserCommand(t *testing.T) { ctx := context.Background() repository := mysqltest.NewRepository(t) diff --git a/services/room-service/internal/room/service/moderation.go b/services/room-service/internal/room/service/moderation.go index c1e27634..fe8c3b6c 100644 --- a/services/room-service/internal/room/service/moderation.go +++ b/services/room-service/internal/room/service/moderation.go @@ -121,14 +121,13 @@ func (s *Service) MuteUser(ctx context.Context, req *roomv1.MuteUserRequest) (*r } if s.rtcUserAudioBlocker != nil { - // 低延迟路径在 Room Cell 提交后立即更新 RTC；失败不能回滚已经落盘的禁言事实， - // 同一条 RoomUserMuted outbox 会按最新快照继续补偿，避免客户端暂时断线造成永久漏禁。 - blocked := containsID(result.snapshot.GetMuteUserIds(), cmd.TargetUserID) - if err := s.rtcUserAudioBlocker.SetUserAudioBlockedByStrRoomID(ctx, cmd.RoomID(), cmd.TargetUserID, blocked); err != nil { + // 低延迟路径在 Room Cell 提交后立即收敛 RTC；必须重读当前状态， + // 不能使用 result.snapshot，因为旧 command_id 重放会返回历史结果， + // 在 newer unmute 后照旧投影就会把云端状态永久回退。 + if err := s.reconcileRTCUserAudio(ctx, cmd.RoomID(), cmd.TargetUserID, false); err != nil { logx.Warn(ctx, "rtc_user_audio_block_sync_failed", slog.String("room_id", cmd.RoomID()), slog.Int64("target_user_id", cmd.TargetUserID), - slog.Bool("blocked", blocked), slog.String("error", err.Error()), ) } diff --git a/services/room-service/internal/room/service/outbox_worker.go b/services/room-service/internal/room/service/outbox_worker.go index ec2de989..8535bd9a 100644 --- a/services/room-service/internal/room/service/outbox_worker.go +++ b/services/room-service/internal/room/service/outbox_worker.go @@ -413,19 +413,29 @@ func (s *Service) publishOutboxRecord(ctx context.Context, record outbox.Record) } return s.roomDisplayPublisher.PublishOutboxEvent(ctx, record.Envelope) } - // RTC 禁言是 RoomUserMuted 的外部媒体投影。先按当前 Room Cell 状态收敛，再发布通用 - // MQ/活动消费者；任何一侧失败都会保留同一 outbox 记录重试，且 RTC 调用本身幂等。 - if err := s.syncRTCUserMuteFromOutbox(ctx, record.Envelope); err != nil { - return err - } // Ignited 在 outbox worker 内安排延迟发射，保证 MQ 调度失败能通过 room_outbox retry；客户端 IM 展示已在 Room Cell 提交后直发。 if err := s.scheduleRoomRocketLaunchFromEnvelope(ctx, record.Envelope); err != nil { return err } - if s.outboxPublisher == nil { - return nil + var publishErr error + if s.outboxPublisher != nil { + // 通用事实通道必须先尝试发布；RTC 外部 API 慢或失败不能让 + // RoomUserMuted 在进入 MQ/活动消费者之前就被截断。RTC 失败时本记录仍会 + // retry，MQ 可能重放由既有 event_id 幂等收敛。 + publishErr = s.outboxPublisher.PublishOutboxEvent(ctx, record.Envelope) } - return s.outboxPublisher.PublishOutboxEvent(ctx, record.Envelope) + + var rtcErr error + if record.Envelope != nil && record.Envelope.GetEventType() == "RoomUserMuted" { + // RTC 使用独立有界预算，不消耗通用 publisher 的 deadline；两侧都尝试后 + // 任一失败都让 outbox 保持 retryable，避免丢失媒体补偿或业务事实。 + rtcCtx, cancel := context.WithTimeout(appcode.WithContext(context.Background(), appcode.FromContext(ctx)), rtcMuteProjectionTimeout) + rtcErr = s.syncRTCUserMuteFromOutbox(rtcCtx, record.Envelope) + cancel() + } + // Join 保留两个独立通道的排障语义；只要任一不为 nil，调用方就会 + // 将记录改为 retryable，下次同时重放 MQ event_id 和 latest-state RTC 投影。 + return errors.Join(publishErr, rtcErr) } // outboxBackoff 根据失败次数计算指数退避，且永远不超过 MaxBackoff。 diff --git a/services/room-service/internal/room/service/outbox_worker_internal_test.go b/services/room-service/internal/room/service/outbox_worker_internal_test.go index 1e76d122..e59fa8dd 100644 --- a/services/room-service/internal/room/service/outbox_worker_internal_test.go +++ b/services/room-service/internal/room/service/outbox_worker_internal_test.go @@ -79,6 +79,40 @@ func TestOutboxSingleRecordLeaseCoversPublishAndStateWrite(t *testing.T) { } } +func TestRTCMuteProjectionGateWaitHonorsContextAndReleasesRefs(t *testing.T) { + svc := &Service{} + key := rtcMuteProjectionKey{appCode: "lalu", roomID: "room-gate-timeout", userID: 1001} + releaseOwner, err := svc.acquireRTCMuteProjection(context.Background(), key) + if err != nil { + t.Fatalf("acquire gate owner: %v", err) + } + + waitCtx, cancel := context.WithTimeout(context.Background(), 20*time.Millisecond) + defer cancel() + started := time.Now() + if _, err := svc.acquireRTCMuteProjection(waitCtx, key); err == nil { + t.Fatal("waiting gate acquisition must honor context timeout") + } + if elapsed := time.Since(started); elapsed > 250*time.Millisecond { + t.Fatalf("gate waiter exceeded bounded context: %s", elapsed) + } + svc.rtcMuteGateMu.Lock() + gate := svc.rtcMuteGates[key] + if gate == nil || gate.refs != 1 { + svc.rtcMuteGateMu.Unlock() + t.Fatalf("cancelled waiter leaked gate ref: gate=%+v", gate) + } + svc.rtcMuteGateMu.Unlock() + + releaseOwner() + svc.rtcMuteGateMu.Lock() + remaining := len(svc.rtcMuteGates) + svc.rtcMuteGateMu.Unlock() + if remaining != 0 { + t.Fatalf("last gate owner release must delete keyed gate, remaining=%d", remaining) + } +} + func TestDirectAndRobotPublishersScopeBuiltRecordsFromContext(t *testing.T) { ctx := appcode.WithContext(context.Background(), "fami") directPublisher := newScopedOutboxPublisher() diff --git a/services/room-service/internal/room/service/rtc_event.go b/services/room-service/internal/room/service/rtc_event.go index b4dc4ab7..c0d11fa4 100644 --- a/services/room-service/internal/room/service/rtc_event.go +++ b/services/room-service/internal/room/service/rtc_event.go @@ -66,10 +66,11 @@ func (s *Service) ApplyRTCEvent(ctx context.Context, req *roomv1.ApplyRTCEventRe if err != nil { return nil, err } - if cmd.EventType == rtcEventAudioStarted && s.rtcUserAudioBlocker != nil && containsID(result.snapshot.GetMuteUserIds(), cmd.TargetUserID) { + if cmd.EventType == rtcEventAudioStarted && s.rtcUserAudioBlocker != nil { // 被禁言用户绕过客户端直接发流时，腾讯 RTC webhook 是第二道服务端防线。 - // 失败向 gateway 返回错误让腾讯重试；幂等 command replay 仍会再次执行本段外部投影。 - if err := s.rtcUserAudioBlocker.SetUserAudioBlockedByStrRoomID(ctx, cmd.RoomID(), cmd.TargetUserID, true); err != nil { + // 这里不能根据 result.snapshot 判断：腾讯重放旧 external_event_id 时会拿到 + // 历史 command result；reconciler 重读最新 Room Cell，未禁言时不发多余 unblock。 + if err := s.reconcileRTCUserAudio(ctx, cmd.RoomID(), cmd.TargetUserID, true); err != nil { return nil, err } } diff --git a/services/room-service/internal/room/service/rtc_moderation.go b/services/room-service/internal/room/service/rtc_moderation.go index 74b7e17a..0d3b430b 100644 --- a/services/room-service/internal/room/service/rtc_moderation.go +++ b/services/room-service/internal/room/service/rtc_moderation.go @@ -3,11 +3,37 @@ package service import ( "context" "fmt" + "strings" + "time" "google.golang.org/protobuf/proto" roomeventsv1 "hyapp.local/api/proto/events/room/v1" + "hyapp/pkg/appcode" ) +const ( + // SetUserBlockedByStrRoomId 官方限频为 20 QPS；全进程每 50ms 最多发起一次。 + rtcMuteProjectionInterval = 50 * time.Millisecond + // 一次入口最多追赶三个并发状态版本；持续颠簸时返回可重试错误， + // 交给新命令的 post-commit 调用或 durable outbox 继续收敛，不无界占用 worker。 + rtcMuteProjectionMaxAttempts = 3 + // REST client 单次默认上限为 5s；总预算只多留 1s 给门闩、限速和快照复核。 + rtcMuteProjectionTimeout = 6 * time.Second +) + +type rtcMuteProjectionKey struct { + appCode string + roomID string + userID int64 +} + +// rtcMuteProjectionGate 用 channel 而不是 sync.Mutex，让等待同一用户投影的请求 +// 能遵守 context deadline；refs 由 Service.rtcMuteGateMu 保护，最后一个使用者退出即删除门闩。 +type rtcMuteProjectionGate struct { + token chan struct{} + refs int +} + // syncRTCUserMuteFromOutbox 把持久化的房间禁言事实补偿到腾讯 RTC 媒体平面。 // 重试时必须读取 Room Cell 的最新禁言集合，而不能直接重放旧事件里的 muted；否则旧 mute // 在 newer unmute 之后重试会重新关掉用户音频，形成跨系统状态回退。 @@ -24,15 +50,124 @@ func (s *Service) syncRTCUserMuteFromOutbox(ctx context.Context, envelope *roome return fmt.Errorf("rtc mute outbox event is incomplete") } - snapshot, err := s.currentSnapshot(ctx, envelope.GetRoomId()) + return s.reconcileRTCUserAudio(ctx, envelope.GetRoomId(), event.GetTargetUserId(), false) +} + +// reconcileRTCUserAudio 只把调用时的最新 Room Cell 状态投影到 RTC，从不信任历史 +// command result 或 outbox payload 中的 muted 快照。onlyIfBlocked 只用于 audio_started +// webhook：用户本来未被禁言时无需多发一次 unmute，但如果本轮已经 block 后 +// 又观测到新 unmute，仍必须追赶并发出 unblock，否则会留下本轮自己的旧副作用。 +func (s *Service) reconcileRTCUserAudio(parent context.Context, roomID string, userID int64, onlyIfBlocked bool) error { + if s == nil || s.rtcUserAudioBlocker == nil { + return nil + } + roomID = strings.TrimSpace(roomID) + if roomID == "" || userID <= 0 { + return fmt.Errorf("rtc mute reconcile target is incomplete") + } + + ctx, cancel := context.WithTimeout(parent, rtcMuteProjectionTimeout) + defer cancel() + release, err := s.acquireRTCMuteProjection(ctx, rtcMuteProjectionKey{ + appCode: appcode.FromContext(ctx), + roomID: roomID, + userID: userID, + }) if err != nil { return err } - if snapshot == nil { - // 房间已被物理清理时 RTC 连接也不再存在；把历史禁言事件视为终态成功， - // 避免一条不可执行的外部副作用永久阻塞同一 outbox worker。 + defer release() + + projected := false + for attempt := 1; attempt <= rtcMuteProjectionMaxAttempts; attempt++ { + before, err := s.currentSnapshot(ctx, roomID) + if err != nil { + return err + } + if before == nil { + // 房间已被物理清理时 RTC 连接也不再存在，历史禁言投影可直接收敛。 + return nil + } + desiredBlocked := containsID(before.GetMuteUserIds(), userID) + if onlyIfBlocked && !projected && !desiredBlocked { + return nil + } + if err := s.waitRTCMuteProjectionRate(ctx); err != nil { + return err + } + if err := s.rtcUserAudioBlocker.SetUserAudioBlockedByStrRoomID(ctx, roomID, userID, desiredBlocked); err != nil { + return err + } + projected = true + + after, err := s.currentSnapshot(ctx, roomID) + if err != nil { + return err + } + if after == nil || containsID(after.GetMuteUserIds(), userID) == desiredBlocked { + return nil + } + // 状态在云 API 请求期间变化；保持同一 keyed gate 并重读最新快照， + // 让本轮亲自纠正可能已落地的旧云端状态。 + } + return fmt.Errorf("rtc mute state kept changing during %d reconcile attempts", rtcMuteProjectionMaxAttempts) +} + +func (s *Service) acquireRTCMuteProjection(ctx context.Context, key rtcMuteProjectionKey) (func(), error) { + s.rtcMuteGateMu.Lock() + if s.rtcMuteGates == nil { + s.rtcMuteGates = make(map[rtcMuteProjectionKey]*rtcMuteProjectionGate) + } + gate := s.rtcMuteGates[key] + if gate == nil { + gate = &rtcMuteProjectionGate{token: make(chan struct{}, 1)} + gate.token <- struct{}{} + s.rtcMuteGates[key] = gate + } + gate.refs++ + s.rtcMuteGateMu.Unlock() + + select { + case <-ctx.Done(): + s.releaseRTCMuteProjectionRef(key, gate) + return nil, ctx.Err() + case <-gate.token: + return func() { + gate.token <- struct{}{} + s.releaseRTCMuteProjectionRef(key, gate) + }, nil + } +} + +func (s *Service) releaseRTCMuteProjectionRef(key rtcMuteProjectionKey, gate *rtcMuteProjectionGate) { + s.rtcMuteGateMu.Lock() + defer s.rtcMuteGateMu.Unlock() + gate.refs-- + if gate.refs == 0 && s.rtcMuteGates[key] == gate { + delete(s.rtcMuteGates, key) + } +} + +func (s *Service) waitRTCMuteProjectionRate(ctx context.Context) error { + s.rtcMuteRateMu.Lock() + now := time.Now() + slot := now + if s.rtcMuteNextCall.After(slot) { + slot = s.rtcMuteNextCall + } + s.rtcMuteNextCall = slot.Add(rtcMuteProjectionInterval) + s.rtcMuteRateMu.Unlock() + + wait := time.Until(slot) + if wait <= 0 { + return nil + } + timer := time.NewTimer(wait) + defer timer.Stop() + select { + case <-ctx.Done(): + return ctx.Err() + case <-timer.C: return nil } - blocked := containsID(snapshot.GetMuteUserIds(), event.GetTargetUserId()) - return s.rtcUserAudioBlocker.SetUserAudioBlockedByStrRoomID(ctx, envelope.GetRoomId(), event.GetTargetUserId(), blocked) } diff --git a/services/room-service/internal/room/service/service.go b/services/room-service/internal/room/service/service.go index b2955256..ec7a7afc 100644 --- a/services/room-service/internal/room/service/service.go +++ b/services/room-service/internal/room/service/service.go @@ -121,6 +121,14 @@ type Service struct { rtcUserRemover integration.RTCUserRemover // rtcUserAudioBlocker 在房间禁言提交后关闭目标用户的 RTC 上行音频；状态仍只保存在 Room Cell。 rtcUserAudioBlocker integration.RTCUserAudioBlocker + // rtcMuteGateMu 保护按 app/room/user 划分的 RTC 投影门闩；门闩只串行化本进程请求， + // 跨实例竞态由每次云 API 返回后的 latest-state recheck 继续收敛。 + rtcMuteGateMu sync.Mutex + rtcMuteGates map[rtcMuteProjectionKey]*rtcMuteProjectionGate + // rtcMuteRateMu 和 nextCall 在进程内把 SetUserBlockedByStrRoomId 压到官方 20 QPS 以内， + // 避免 HTTP 直调与 outbox 补偿叠加后在故障时放大限流。 + rtcMuteRateMu sync.Mutex + rtcMuteNextCall time.Time // draining 表示当前节点正在下线，只允许已经进入执行链路的命令收尾。 draining atomic.Bool @@ -304,6 +312,7 @@ func New(cfg Config, directory router.Directory, repository Repository, wallet i luckyGiftSendLockTTL: luckyGiftSendLockTTL, rtcUserRemover: cfg.RTCUserRemover, rtcUserAudioBlocker: cfg.RTCUserAudioBlocker, + rtcMuteGates: make(map[rtcMuteProjectionKey]*rtcMuteProjectionGate), cells: make(map[string]*loadedCell), robotRuntimes: make(map[string]robotRoomRuntime), humanRobotRuntimes: make(map[string]robotRoomRuntime),