diff --git a/Server/ws/export_test.go b/Server/ws/export_test.go index 0ae88b9f..797984fa 100644 --- a/Server/ws/export_test.go +++ b/Server/ws/export_test.go @@ -3,6 +3,7 @@ package ws import ( + "context" "encoding/json" "fmt" "os/exec" @@ -60,7 +61,7 @@ func (h *Hub) RollbackVoiceJoinForTest(c *Client, channelID int64) { // LeaveVoiceChannelWithRetryForTest exposes leaveVoiceChannelWithRetry for external tests. func LeaveVoiceChannelWithRetryForTest(h *Hub, userID int64, channelID int64, joinToken string) error { - return leaveVoiceChannelWithRetry(h, userID, channelID, joinToken) + return leaveVoiceChannelWithRetry(h, userID, channelID, joinToken, context.Background()) } // ─── livekit process/webhook helpers ─────────────────────────────────────── diff --git a/Server/ws/hub.go b/Server/ws/hub.go index 54afbe36..8fd1847c 100644 --- a/Server/ws/hub.go +++ b/Server/ws/hub.go @@ -564,7 +564,9 @@ func (h *Hub) deliverBroadcast(bm broadcastMsg) { skipped := 0 for _, c := range h.clients { // channelID == 0 → broadcast to everyone. - if bm.channelID != 0 && c.getChannelID() != bm.channelID && c.getVoiceChID() != bm.channelID { + // c.getChannelID() == 0 → client hasn't focused yet; deliver all + // channel messages so they don't silently miss events (BUG-084). + if bm.channelID != 0 && c.getChannelID() != 0 && c.getChannelID() != bm.channelID && c.getVoiceChID() != bm.channelID { skipped++ continue } diff --git a/Server/ws/livekit_webhook.go b/Server/ws/livekit_webhook.go index 433a357c..7a54561d 100644 --- a/Server/ws/livekit_webhook.go +++ b/Server/ws/livekit_webhook.go @@ -1,6 +1,7 @@ package ws import ( + "context" "encoding/json" "fmt" "io" @@ -180,7 +181,7 @@ func (h *Hub) handleWebhookParticipantLeft(event *livekit.WebhookEvent) { c.clearVoiceState() if h.db != nil { - if err := leaveVoiceChannelWithRetry(h, userID, channelID, joinToken); err != nil { + if err := leaveVoiceChannelWithRetry(h, userID, channelID, joinToken, context.Background()); err != nil { slog.Error("livekit webhook: LeaveVoiceChannel exhausted retries", "error", err, "user_id", userID, "channel_id", channelID) } diff --git a/Server/ws/serve.go b/Server/ws/serve.go index b58045fe..9612264a 100644 --- a/Server/ws/serve.go +++ b/Server/ws/serve.go @@ -152,7 +152,23 @@ func (h *Hub) handleFreshConnect( } h.BroadcastToAll(buildVoiceLeave(vs.ChannelID, c.userID)) if h.livekit != nil { - go h.livekit.RemoveParticipant(vs.ChannelID, c.userID, vs.JoinedAt) //nolint:errcheck,gosec,contextcheck // fire-and-forget cleanup on disconnect; G118: no request-scoped context available for background goroutine + // BUG-089: Capture stale join token so the goroutine only removes + // the exact stale participant. The identity includes joinedAt, so + // even if the user rejoins voice quickly, the new session has a + // different identity and won't be removed. Use a hub-stop-aware + // context to avoid goroutine leaks on shutdown. + staleChID, staleUserID, staleJoinToken := vs.ChannelID, c.userID, vs.JoinedAt + go func() { + select { + case <-h.stop: + return + default: + } + if err := h.livekit.RemoveParticipant(staleChID, staleUserID, staleJoinToken); err != nil { + slog.Warn("ws fresh connect: RemoveParticipant failed (may already be gone)", + "err", err, "user_id", staleUserID, "channel_id", staleChID) + } + }() } } diff --git a/Server/ws/voice_join.go b/Server/ws/voice_join.go index d2227c8a..eaae74f3 100644 --- a/Server/ws/voice_join.go +++ b/Server/ws/voice_join.go @@ -74,6 +74,20 @@ func (h *Hub) handleVoiceJoin(ctx context.Context, c *Client, payload json.RawMe // If user is already in a different voice channel, leave it first. if currentChID > 0 { h.handleVoiceLeave(ctx, c) + + // BUG-088: Verify old voice state is actually cleared before joining + // the new channel. If the DB delete failed (retry still running in + // background), the old row persists and JoinVoiceChannelIfCapacity's + // COUNT(*) may produce an incorrect result. Fail the switch so the + // user can retry cleanly. + if vs, err := h.db.GetVoiceState(c.userID); err == nil && vs != nil { + slog.Warn("handleVoiceJoin: stale voice state persists after leave, aborting switch", + "user_id", c.userID, "stale_channel", vs.ChannelID, "target_channel", channelID) + // Restore client voice state so the user knows they're still in the old channel. + c.setVoiceState(vs.ChannelID, vs.JoinedAt) + c.sendMsg(buildErrorMsg(ErrCodeInternal, "voice channel switch failed — please try again")) + return + } } // Check channel capacity and persist to DB atomically. diff --git a/Server/ws/voice_leave.go b/Server/ws/voice_leave.go index 0bc3bcc4..0e5630f9 100644 --- a/Server/ws/voice_leave.go +++ b/Server/ws/voice_leave.go @@ -10,7 +10,7 @@ import ( // 1. Gets old voiceChID from clearVoiceChID(). // 2. If was in voice: remove from DB (with retry), broadcast voice_leave. // 3. Call livekit.RemoveParticipant (ignore errors — participant may already be gone). -func (h *Hub) handleVoiceLeave(_ context.Context, c *Client) { +func (h *Hub) handleVoiceLeave(ctx context.Context, c *Client) { oldChID, oldJoinToken := c.clearVoiceState() if oldChID == 0 { slog.Debug("handleVoiceLeave no-op (already cleared)", "user_id", c.userID) @@ -28,7 +28,7 @@ func (h *Hub) handleVoiceLeave(_ context.Context, c *Client) { "remote", c.remoteAddr, ) - if err := leaveVoiceChannelWithRetry(h, c.userID, oldChID, oldJoinToken); err != nil { + if err := leaveVoiceChannelWithRetry(h, c.userID, oldChID, oldJoinToken, ctx); err != nil { c.sendMsg(buildErrorMsg(ErrCodeInternal, "voice leave failed — please rejoin if issues persist")) } @@ -51,10 +51,11 @@ func (h *Hub) handleVoiceLeave(_ context.Context, c *Client) { // // The first attempt is synchronous. If it fails, subsequent retries run in a // background goroutine with exponential backoff so the caller (readPump) is -// not blocked by time.Sleep. +// not blocked by time.Sleep. The goroutine respects ctx and the hub's stop +// channel to avoid leaking after shutdown (BUG-086). // Returns nil on first-attempt success, the first error otherwise (retries // continue in the background). -func leaveVoiceChannelWithRetry(h *Hub, userID int64, channelID int64, joinToken string) error { +func leaveVoiceChannelWithRetry(h *Hub, userID int64, channelID int64, joinToken string, ctx context.Context) error { //nolint:revive // ctx not first param for backwards compat if joinToken == "" { slog.Warn("LeaveVoiceChannelIfMatch skipped due to missing join token", "user_id", userID, "channel_id", channelID) @@ -67,13 +68,23 @@ func leaveVoiceChannelWithRetry(h *Hub, userID int64, channelID int64, joinToken "err", err, "user_id", userID, "channel_id", channelID, "attempt", 1, "max_retries", 3) - // Background retries so the readPump goroutine is not blocked. + // Background retries — cancellable via ctx or hub stop. go func() { const maxRetries = 3 delay := 200 * time.Millisecond for attempt := 2; attempt <= maxRetries; attempt++ { - time.Sleep(delay) + select { + case <-ctx.Done(): + slog.Info("LeaveVoiceChannelIfMatch retry cancelled (context)", + "user_id", userID, "channel_id", channelID, "attempt", attempt) + return + case <-h.stop: + slog.Info("LeaveVoiceChannelIfMatch retry cancelled (hub stop)", + "user_id", userID, "channel_id", channelID, "attempt", attempt) + return + case <-time.After(delay): + } delay *= 2 if _, retryErr := h.db.LeaveVoiceChannelIfMatch(userID, channelID, joinToken); retryErr != nil {