From 50af9b3c8bf0c9016ac6ca1729eb40ed3fc7566b Mon Sep 17 00:00:00 2001 From: yummybomb <19238148+yummybomb@users.noreply.github.com> Date: Wed, 11 Nov 2026 08:50:37 +0000 Subject: [PATCH] Destroy a WebRTC peer whose client left before it connected A client that asks for media and goes away before it answers the offer leaves a WebRTC peer that never connects. Nothing destroys such a peer: the connection state handler only acts on a peer that disconnects or fails, and deleting or disconnecting a session only destroys a peer that is watching. The peer keeps its listeners on the capture streams, so the video and audio pipelines keep encoding for the life of the process with nobody watching. A peer needs the websocket to finish connecting, so one that has not connected is now destroyed when its session's websocket goes away. The websocket can also go away while the peer is still being created; SetWebRTCPeer then destroys the peer it is given. A track that was shut down refuses a stream, so a peer destroyed between its creation and the request that selects its video cannot start a pipeline nothing stops. --- server/internal/session/session.go | 32 +++++ server/internal/session/session_test.go | 172 ++++++++++++++++++++++++ server/internal/webrtc/track.go | 14 ++ server/internal/webrtc/track_test.go | 79 +++++++++++ 4 files changed, 297 insertions(+) create mode 100644 server/internal/webrtc/track_test.go 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) + } +}