From d1b031a9dd4a8e511348ad4f657960edd49f221d Mon Sep 17 00:00:00 2001 From: Benjamin Pracht Date: Thu, 2 Jul 2026 14:07:21 +0100 Subject: [PATCH] Stop WHIP session notifier when participant leaves (#4637) The WHIP connection-notify loop kept issuing RPCs after the participant had left the room. Guard sendConnectionNotify against a closed participant (returning ErrParticipantNotFound) and treat that error as a clean loop exit. Also reorder DeleteSession so the participant is removed before its WHIP OnClose entry is cleared. Adds tests covering loop termination on participant leave, context cancellation, and the closed-participant guard. Co-authored-by: Claude Opus 4.8 (1M context) --- pkg/service/roommanager_service.go | 13 ++- pkg/service/roommanager_service_test.go | 125 ++++++++++++++++++++++++ 2 files changed, 133 insertions(+), 5 deletions(-) create mode 100644 pkg/service/roommanager_service_test.go diff --git a/pkg/service/roommanager_service.go b/pkg/service/roommanager_service.go index 5986fa997..d8ea05433 100644 --- a/pkg/service/roommanager_service.go +++ b/pkg/service/roommanager_service.go @@ -20,9 +20,8 @@ import ( "github.com/livekit/psrpc" ) -const ( - whipSessionNotifyInterval = 10 * time.Second -) +// whipSessionNotifyInterval is a var (rather than a const) so tests can shorten it. +var whipSessionNotifyInterval = 10 * time.Second type whipService struct { *RoomManager @@ -181,7 +180,7 @@ func (s whipService) notifySession(ctx context.Context, participant types.Partic case <-ticker.C: err := s.sendConnectionNotify(ctx, participant) if err != nil { - if errors.Is(err, context.Canceled) { + if errors.Is(err, context.Canceled) || errors.Is(err, ErrParticipantNotFound) { return nil } } @@ -193,6 +192,10 @@ func (s whipService) notifySession(ctx context.Context, participant types.Partic } func (s whipService) sendConnectionNotify(ctx context.Context, participant types.Participant) error { + if participant.IsClosed() { + return ErrParticipantNotFound + } + video, audio := getMediaStateForParticipant(participant) _, err := s.ingressRpcCli.WHIPRTCConnectionNotify(ctx, string(participant.ID()), &rpc.WHIPRTCConnectionNotifyRequest{ @@ -345,12 +348,12 @@ func (r whipParticipantService) DeleteSession(ctx context.Context, req *rpc.WHIP lp := room.GetParticipantByID(livekit.ParticipantID(req.ParticipantId)) if lp != nil { - lp.AddOnClose(types.ParticipantCloseKeyWHIP, nil) room.RemoveParticipant( lp.Identity(), lp.ID(), types.ParticipantCloseReasonClientRequestLeave, ) + lp.AddOnClose(types.ParticipantCloseKeyWHIP, nil) } return &emptypb.Empty{}, nil diff --git a/pkg/service/roommanager_service_test.go b/pkg/service/roommanager_service_test.go new file mode 100644 index 000000000..e0d588080 --- /dev/null +++ b/pkg/service/roommanager_service_test.go @@ -0,0 +1,125 @@ +package service + +import ( + "context" + "testing" + "time" + + "github.com/stretchr/testify/require" + "go.uber.org/atomic" + "google.golang.org/protobuf/types/known/emptypb" + + "github.com/livekit/protocol/livekit" + "github.com/livekit/protocol/rpc" + "github.com/livekit/psrpc" + + "github.com/livekit/livekit-server/pkg/rtc/types/typesfakes" +) + +// fakeIngressHandlerClient records WHIPRTCConnectionNotify calls. It embeds the +// interface so only the method under test needs to be implemented; any other +// call would panic (and we assert none happen). +type fakeIngressHandlerClient struct { + rpc.IngressHandlerClient + notifyCount atomic.Int32 +} + +func (f *fakeIngressHandlerClient) WHIPRTCConnectionNotify( + _ context.Context, + _ string, + _ *rpc.WHIPRTCConnectionNotifyRequest, + _ ...psrpc.RequestOption, +) (*emptypb.Empty, error) { + f.notifyCount.Inc() + return &emptypb.Empty{}, nil +} + +// TestWhipNotifySessionStopsWhenParticipantLeaves verifies the notifier loop +// terminates once the WHIP participant leaves the room (i.e. IsClosed becomes +// true), and stops issuing further connection notifications. +func TestWhipNotifySessionStopsWhenParticipantLeaves(t *testing.T) { + origInterval := whipSessionNotifyInterval + whipSessionNotifyInterval = 5 * time.Millisecond + t.Cleanup(func() { whipSessionNotifyInterval = origInterval }) + + var closed atomic.Bool + participant := &typesfakes.FakeParticipant{} + participant.IsClosedStub = func() bool { return closed.Load() } + participant.IDReturns(livekit.ParticipantID("PA_test")) + participant.ToProtoReturns(&livekit.ParticipantInfo{}) + + cli := &fakeIngressHandlerClient{} + s := whipService{ingressRpcCli: cli} + + done := make(chan error, 1) + go func() { + done <- s.notifySession(context.Background(), participant) + }() + + // while the participant is connected the loop should keep notifying + require.Eventually(t, func() bool { + return cli.notifyCount.Load() > 0 + }, time.Second, time.Millisecond, "expected notifications while participant is connected") + + // the participant leaves the room + closed.Store(true) + + select { + case err := <-done: + require.NoError(t, err) + case <-time.After(time.Second): + t.Fatal("notifySession did not stop after the participant left the room") + } + + // no further notifications should be attempted after it stops + countAtStop := cli.notifyCount.Load() + time.Sleep(50 * time.Millisecond) + require.Equal(t, countAtStop, cli.notifyCount.Load(), "should not notify after the participant left") +} + +// TestWhipNotifySessionStopsOnContextCancel verifies the loop exits when the +// aliveCtx (cancelled from the participant's OnClose callback) is done. +func TestWhipNotifySessionStopsOnContextCancel(t *testing.T) { + origInterval := whipSessionNotifyInterval + whipSessionNotifyInterval = 5 * time.Millisecond + t.Cleanup(func() { whipSessionNotifyInterval = origInterval }) + + participant := &typesfakes.FakeParticipant{} + participant.IsClosedReturns(false) + participant.IDReturns(livekit.ParticipantID("PA_test")) + participant.ToProtoReturns(&livekit.ParticipantInfo{}) + + cli := &fakeIngressHandlerClient{} + s := whipService{ingressRpcCli: cli} + + ctx, cancel := context.WithCancel(context.Background()) + done := make(chan error, 1) + go func() { + done <- s.notifySession(ctx, participant) + }() + + cancel() + + select { + case err := <-done: + require.NoError(t, err) + case <-time.After(time.Second): + t.Fatal("notifySession did not stop after context was cancelled") + } +} + +// TestWhipSendConnectionNotifySkipsClosedParticipant verifies the guard that +// short-circuits the RPC (and drives loop termination) for a closed participant. +func TestWhipSendConnectionNotifySkipsClosedParticipant(t *testing.T) { + participant := &typesfakes.FakeParticipant{} + participant.IsClosedReturns(true) + participant.IDReturns(livekit.ParticipantID("PA_test")) + participant.ToProtoReturns(&livekit.ParticipantInfo{}) + + cli := &fakeIngressHandlerClient{} + s := whipService{ingressRpcCli: cli} + + err := s.sendConnectionNotify(context.Background(), participant) + require.ErrorIs(t, err, ErrParticipantNotFound) + require.Zero(t, cli.notifyCount.Load(), "should not issue an RPC for a closed participant") +}