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) <noreply@anthropic.com>

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

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")
+}