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 1/2] 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) + } +} From a84c7136afba9dc077016298bec95bc5cdfef2e7 Mon Sep 17 00:00:00 2001 From: yummybomb <19238148+yummybomb@users.noreply.github.com> Date: Fri, 13 Nov 2026 23:34:58 +0000 Subject: [PATCH 2/2] Track whether the session's WebRTC peer has connected per peer Whether a peer had connected was read from the session's watching state. That state is left over from a peer that was replaced: the closing of a replaced peer is ignored, so the session goes on counting as watching until the new peer connects. A client that was watching, asked for media again and left before it answered therefore still left a peer that nothing destroyed, with both pipelines running. The session now records under the peer's lock whether the current peer has connected, and resets that when the peer is replaced. --- server/internal/session/session.go | 21 ++++++--- server/internal/session/session_test.go | 58 +++++++++++++++++++++++++ 2 files changed, 74 insertions(+), 5 deletions(-) diff --git a/server/internal/session/session.go b/server/internal/session/session.go index eb5e31ce3..1872914df 100644 --- a/server/internal/session/session.go +++ b/server/internal/session/session.go @@ -30,7 +30,10 @@ type SessionCtx struct { wsDelayedTimer *time.Timer webrtcPeer types.WebRTCPeer - webrtcMu sync.Mutex + // whether webrtcPeer has connected; unlike state.IsWatching this is never + // left over from a peer that was replaced + webrtcConnected bool + webrtcMu sync.Mutex } func (session *SessionCtx) ID() string { @@ -149,6 +152,8 @@ func (session *SessionCtx) ConnectWebSocketPeer(websocketPeer types.WebSocketPee // but after a delay. If the peer is connected again before the delay, the disconnect will be // cancelled. // +// A WebRTC peer that has not connected is destroyed with the websocket peer. +// // If the peer is not the current peer or the peer is nil, it will be ignored. func (session *SessionCtx) DisconnectWebSocketPeer(websocketPeer types.WebSocketPeer, delayed bool) { session.websocketMu.Lock() @@ -227,11 +232,11 @@ func (session *SessionCtx) DisconnectWebSocketPeer(websocketPeer types.WebSocket // 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 - } + session.webrtcMu.Lock() + webrtcPeer, connected := session.webrtcPeer, session.webrtcConnected + session.webrtcMu.Unlock() - if webrtcPeer := session.GetWebRTCPeer(); webrtcPeer != nil { + if webrtcPeer != nil && !connected { webrtcPeer.Destroy() } } @@ -272,6 +277,9 @@ func (session *SessionCtx) Send(event string, payload any) { // Set webrtc peer and destroy the old one, if there is old one. func (session *SessionCtx) SetWebRTCPeer(webrtcPeer types.WebRTCPeer) { session.webrtcMu.Lock() + if webrtcPeer != session.webrtcPeer { + session.webrtcConnected = false + } session.webrtcPeer, webrtcPeer = webrtcPeer, session.webrtcPeer session.webrtcMu.Unlock() @@ -299,6 +307,9 @@ func (session *SessionCtx) SetWebRTCPeer(webrtcPeer types.WebRTCPeer) { func (session *SessionCtx) SetWebRTCConnected(webrtcPeer types.WebRTCPeer, connected bool) { session.webrtcMu.Lock() isCurrentPeer := webrtcPeer == session.webrtcPeer + if isCurrentPeer { + session.webrtcConnected = connected + } session.webrtcMu.Unlock() if !isCurrentPeer { diff --git a/server/internal/session/session_test.go b/server/internal/session/session_test.go index d685d9066..53231952b 100644 --- a/server/internal/session/session_test.go +++ b/server/internal/session/session_test.go @@ -84,6 +84,16 @@ func TestWebRTCPeerThatNeverConnectedIsDestroyedWithItsSession(t *testing.T) { }, wantDestroyed: 0, }, + { + name: "websocket closes while the peer is streaming and was set again", + connected: true, + end: func(_ *SessionManagerCtx, session types.Session, websocket types.WebSocketPeer) error { + session.SetWebRTCPeer(session.GetWebRTCPeer()) + 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 { @@ -211,3 +221,51 @@ func TestWebRTCPeerSetAfterItsWebSocketLeftIsDestroyed(t *testing.T) { }) } } + +// A client that is watching can ask for media again, which replaces its peer. +// The session goes on counting as watching, because the closing of a replaced +// peer is ignored, but the new peer has not connected. +func TestWebRTCPeerThatReplacedAStreamingOneIsDestroyedUnlessItConnected(t *testing.T) { + profile := types.MemberProfile{CanLogin: true, CanConnect: true, CanWatch: true} + + tests := []struct { + name string + connected bool + wantDestroyed int + }{ + {name: "websocket closes before the new peer connects", wantDestroyed: 1}, + {name: "websocket closes while the new peer is streaming", connected: true, wantDestroyed: 0}, + } + + 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) + replaced := &fakeWebRTCPeer{} + session.SetWebRTCPeer(replaced) + session.SetWebRTCConnected(replaced, true) + + peer := &fakeWebRTCPeer{} + session.SetWebRTCPeer(peer) + if tt.connected { + session.SetWebRTCConnected(peer, true) + } + // the replaced peer reports its closing whenever it gets to it + session.SetWebRTCConnected(replaced, false) + + session.DisconnectWebSocketPeer(websocket, false) + + if replaced.destroyed != 1 { + t.Errorf("replaced peer destroyed %d times, want 1", replaced.destroyed) + } + if peer.destroyed != tt.wantDestroyed { + t.Errorf("peer destroyed %d times, want %d", peer.destroyed, tt.wantDestroyed) + } + }) + } +}