diff --git a/server/internal/session/session.go b/server/internal/session/session.go index 686c9611a..eb5e31ce3 100644 --- a/server/internal/session/session.go +++ b/server/internal/session/session.go @@ -212,6 +212,28 @@ func (session *SessionCtx) DisconnectWebSocketPeer(websocketPeer types.WebSocket session.websocketPeer = nil } session.websocketMu.Unlock() + + // A WebRTC peer that has not connected yet needs this websocket to do so: + // it carries the answer and the candidates. Without it the peer would stay + // half-open for good and keep the capture pipelines it subscribed to + // running with nobody watching. + session.destroyPendingWebRTCPeer() +} + +// Destroy the WebRTC peer if it exists but has not connected. A connected peer +// is left alone: it does not need the websocket to keep streaming. +// +// The websocket peer must be cleared before this is called, and a WebRTC peer +// must be set before SetWebRTCPeer looks at the websocket peer. That order +// makes sure one of the two sees the other, whichever happens first. +func (session *SessionCtx) destroyPendingWebRTCPeer() { + if session.state.IsWatching { + return + } + + if webrtcPeer := session.GetWebRTCPeer(); webrtcPeer != nil { + webrtcPeer.Destroy() + } } // Destroy WebSocket peer disconnects the peer and destroys it. It ensures that the peer is @@ -256,6 +278,16 @@ func (session *SessionCtx) SetWebRTCPeer(webrtcPeer types.WebRTCPeer) { if webrtcPeer != nil && webrtcPeer != session.webrtcPeer { webrtcPeer.Destroy() } + + // The websocket can go away while a peer is being created for it. Its + // disconnect found no peer to destroy then, so it is done here. + session.websocketMu.Lock() + hasWebSocket := session.websocketPeer != nil + session.websocketMu.Unlock() + + if !hasWebSocket { + session.destroyPendingWebRTCPeer() + } } // Set if current webrtc peer is connected or not. Since there might be lefover calls from diff --git a/server/internal/session/session_test.go b/server/internal/session/session_test.go index 2a2dddac6..d685d9066 100644 --- a/server/internal/session/session_test.go +++ b/server/internal/session/session_test.go @@ -39,3 +39,175 @@ func TestLegacyIsHost(t *testing.T) { }) } } + +type fakeWebSocketPeer struct{} + +func (fakeWebSocketPeer) Send(event string, payload any) {} +func (fakeWebSocketPeer) Ping() error { return nil } +func (fakeWebSocketPeer) Destroy(reason string) {} + +// fakeWebRTCPeer counts how often it was destroyed. The embedded interface is +// nil: these tests must not reach any other method. +type fakeWebRTCPeer struct { + types.WebRTCPeer + destroyed int +} + +func (p *fakeWebRTCPeer) Destroy() { p.destroyed++ } + +// A client can ask for media and vanish before it answers the offer. Its peer +// never connects, so nothing but the session can end it; left alone it keeps +// the capture pipelines it subscribed to running with nobody watching. +func TestWebRTCPeerThatNeverConnectedIsDestroyedWithItsSession(t *testing.T) { + profile := types.MemberProfile{CanLogin: true, CanConnect: true, CanWatch: true} + + tests := []struct { + name string + connected bool + end func(manager *SessionManagerCtx, session types.Session, websocket types.WebSocketPeer) error + wantDestroyed int + }{ + { + name: "websocket closes before the peer connects", + end: func(_ *SessionManagerCtx, session types.Session, websocket types.WebSocketPeer) error { + session.DisconnectWebSocketPeer(websocket, false) + return nil + }, + wantDestroyed: 1, + }, + { + name: "websocket closes while the peer is streaming", + connected: true, + end: func(_ *SessionManagerCtx, session types.Session, websocket types.WebSocketPeer) error { + session.DisconnectWebSocketPeer(websocket, false) + return nil + }, + wantDestroyed: 0, + }, + { + name: "websocket drops and may still reconnect", + end: func(_ *SessionManagerCtx, session types.Session, websocket types.WebSocketPeer) error { + session.DisconnectWebSocketPeer(websocket, true) + return nil + }, + wantDestroyed: 0, + }, + { + name: "session is deleted before the peer connects", + end: func(manager *SessionManagerCtx, session types.Session, _ types.WebSocketPeer) error { + return manager.Delete(session.ID()) + }, + wantDestroyed: 1, + }, + { + name: "session is disconnected before the peer connects", + end: func(manager *SessionManagerCtx, session types.Session, _ types.WebSocketPeer) error { + return manager.Disconnect(session.ID()) + }, + wantDestroyed: 1, + }, + { + name: "session is deleted while the peer is streaming", + connected: true, + end: func(manager *SessionManagerCtx, session types.Session, _ types.WebSocketPeer) error { + return manager.Delete(session.ID()) + }, + wantDestroyed: 1, + }, + } + + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + manager := New(&config.Session{}) + session, _, err := manager.Create("member", profile) + if err != nil { + t.Fatalf("create session: %v", err) + } + websocket := &fakeWebSocketPeer{} + session.ConnectWebSocketPeer(websocket) + // stops the timer of a delayed disconnect + t.Cleanup(func() { session.DisconnectWebSocketPeer(websocket, false) }) + peer := &fakeWebRTCPeer{} + session.SetWebRTCPeer(peer) + if tt.connected { + session.SetWebRTCConnected(peer, true) + } + + if err := tt.end(manager, session, websocket); err != nil { + t.Fatalf("end session: %v", err) + } + + if peer.destroyed != tt.wantDestroyed { + t.Errorf("peer destroyed %d times, want %d", peer.destroyed, tt.wantDestroyed) + } + }) + } +} + +// Creating a peer takes a while, and the websocket that asked for it can be +// gone by the time the peer is attached to the session. Its disconnect could +// not destroy a peer that did not exist yet. +func TestWebRTCPeerSetAfterItsWebSocketLeftIsDestroyed(t *testing.T) { + profile := types.MemberProfile{CanLogin: true, CanConnect: true, CanWatch: true} + + tests := []struct { + name string + leave func(manager *SessionManagerCtx, session types.Session, websocket types.WebSocketPeer) error + wantDestroyed int + }{ + { + name: "websocket still connected", + leave: func(*SessionManagerCtx, types.Session, types.WebSocketPeer) error { + return nil + }, + wantDestroyed: 0, + }, + { + name: "websocket dropped and may still reconnect", + leave: func(_ *SessionManagerCtx, session types.Session, websocket types.WebSocketPeer) error { + session.DisconnectWebSocketPeer(websocket, true) + return nil + }, + wantDestroyed: 0, + }, + { + name: "websocket closed", + leave: func(_ *SessionManagerCtx, session types.Session, websocket types.WebSocketPeer) error { + session.DisconnectWebSocketPeer(websocket, false) + return nil + }, + wantDestroyed: 1, + }, + { + name: "session deleted", + leave: func(manager *SessionManagerCtx, session types.Session, _ types.WebSocketPeer) error { + return manager.Delete(session.ID()) + }, + wantDestroyed: 1, + }, + } + + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + manager := New(&config.Session{}) + session, _, err := manager.Create("member", profile) + if err != nil { + t.Fatalf("create session: %v", err) + } + websocket := &fakeWebSocketPeer{} + session.ConnectWebSocketPeer(websocket) + // stops the timer of a delayed disconnect + t.Cleanup(func() { session.DisconnectWebSocketPeer(websocket, false) }) + + if err := tt.leave(manager, session, websocket); err != nil { + t.Fatalf("leave: %v", err) + } + peer := &fakeWebRTCPeer{} + session.SetWebRTCPeer(peer) + + if peer.destroyed != tt.wantDestroyed { + t.Errorf("peer destroyed %d times, want %d", peer.destroyed, tt.wantDestroyed) + } + }) + } +} diff --git a/server/internal/webrtc/track.go b/server/internal/webrtc/track.go index 8ddc7ea29..1fc1b557a 100644 --- a/server/internal/webrtc/track.go +++ b/server/internal/webrtc/track.go @@ -22,10 +22,13 @@ type Track struct { sample chan types.Sample paused bool + closed bool stream types.StreamSinkManager streamMu sync.Mutex } +var ErrTrackClosed = errors.New("track is closed") + type trackOption func(*Track) func WithRtcpChan(rtcp chan []rtcp.Packet) trackOption { @@ -64,6 +67,10 @@ func NewTrack(logger zerolog.Logger, codec codec.RTPCodec, connection *webrtc.Pe } func (t *Track) Shutdown() { + t.streamMu.Lock() + t.closed = true + t.streamMu.Unlock() + t.RemoveStream() close(t.sample) } @@ -119,6 +126,13 @@ func (t *Track) SetStream(stream types.StreamSinkManager) (bool, error) { t.streamMu.Lock() defer t.streamMu.Unlock() + // A peer can be destroyed while it is still being set up. A listener added + // after that would never be removed, and its samples would be written to a + // closed channel. + if t.closed { + return false, ErrTrackClosed + } + // if we already listen to the stream, do nothing if t.stream == stream { return false, nil diff --git a/server/internal/webrtc/track_test.go b/server/internal/webrtc/track_test.go new file mode 100644 index 000000000..f7c6f6e7b --- /dev/null +++ b/server/internal/webrtc/track_test.go @@ -0,0 +1,79 @@ +package webrtc + +import ( + "errors" + "testing" + + "github.com/pion/webrtc/v3" + "github.com/rs/zerolog" + + "github.com/m1k1o/neko/server/pkg/types" + "github.com/m1k1o/neko/server/pkg/types/codec" +) + +// listenerCountingStream counts its listeners. The embedded interface is nil: +// these tests must not reach any other method. +type listenerCountingStream struct { + types.StreamSinkManager + added, removed int +} + +func (s *listenerCountingStream) AddListener(types.SampleListener) error { + s.added++ + return nil +} + +func (s *listenerCountingStream) RemoveListener(types.SampleListener) error { + s.removed++ + return nil +} + +func newTestTrack(t *testing.T) *Track { + t.Helper() + connection, err := webrtc.NewPeerConnection(webrtc.Configuration{}) + if err != nil { + t.Fatalf("create peer connection: %v", err) + } + t.Cleanup(func() { connection.Close() }) + + track, err := NewTrack(zerolog.Nop(), codec.VP8(), connection) + if err != nil { + t.Fatalf("create track: %v", err) + } + return track +} + +func TestTrackStopsListeningOnShutdown(t *testing.T) { + track := newTestTrack(t) + stream := &listenerCountingStream{} + + if _, err := track.SetStream(stream); err != nil { + t.Fatalf("set stream: %v", err) + } + track.Shutdown() + + if stream.added != 1 || stream.removed != 1 { + t.Errorf("listener added %d times and removed %d times, want 1 and 1", stream.added, stream.removed) + } +} + +// A peer can be destroyed between its creation and the request that selects +// its video. The track is shut down by then; listening to a stream after that +// would keep the pipeline running for a track nobody removes. +func TestTrackDoesNotListenAfterShutdown(t *testing.T) { + track := newTestTrack(t) + stream := &listenerCountingStream{} + + track.Shutdown() + + if _, err := track.SetStream(stream); !errors.Is(err, ErrTrackClosed) { + t.Errorf("set stream after shutdown returned %v, want %v", err, ErrTrackClosed) + } + // unpausing is what starts audio + track.SetPaused(true) + track.SetPaused(false) + + if stream.added != 0 { + t.Errorf("listener added %d times after shutdown, want 0", stream.added) + } +}