[pull] master from livekit:master

go.mod

@@ -23,7 +23,7 @@ require (
 	github.com/jxskiss/base62 v1.1.0
 	github.com/livekit/mageutil v0.0.0-20250511045019-0f1ff63f7731
 	github.com/livekit/mediatransportutil v0.0.0-20250825135402-7bc31f107ade
-	github.com/livekit/protocol v1.41.1-0.20250905101817-7b3b388b3292
+	github.com/livekit/protocol v1.41.1-0.20250909050443-48ed04737846
 	github.com/livekit/psrpc v0.6.1-0.20250828235857-3fafdbbcbe55
 	github.com/mackerelio/go-osstat v0.2.6
 	github.com/magefile/mage v1.15.0

go.sum

@@ -171,8 +171,8 @@ github.com/livekit/mageutil v0.0.0-20250511045019-0f1ff63f7731 h1:9x+U2HGLrSw5AT
 github.com/livekit/mageutil v0.0.0-20250511045019-0f1ff63f7731/go.mod h1:Rs3MhFwutWhGwmY1VQsygw28z5bWcnEYmS1OG9OxjOQ=
 github.com/livekit/mediatransportutil v0.0.0-20250825135402-7bc31f107ade h1:lpxPcglwzUWNB4J0S2qZuyMehzmR7vW9whzSwV4IGoI=
 github.com/livekit/mediatransportutil v0.0.0-20250825135402-7bc31f107ade/go.mod h1:mSNtYzSf6iY9xM3UX42VEI+STHvMgHmrYzEHPcdhB8A=
-github.com/livekit/protocol v1.41.1-0.20250905101817-7b3b388b3292 h1:JKs2f+btdWXaHG67BBzbKeEeNXVAdlaqJarGLFFa6+Q=
-github.com/livekit/protocol v1.41.1-0.20250905101817-7b3b388b3292/go.mod h1:Scx8arfj5y65w6EYA3ZIKJafoN2xBuV8pauvyrvI4eg=
+github.com/livekit/protocol v1.41.1-0.20250909050443-48ed04737846 h1:J/Ry54lAQShsUZQ74DLxuvVyPl92TCaYi/47eLLKem0=
+github.com/livekit/protocol v1.41.1-0.20250909050443-48ed04737846/go.mod h1:Scx8arfj5y65w6EYA3ZIKJafoN2xBuV8pauvyrvI4eg=
 github.com/livekit/psrpc v0.6.1-0.20250828235857-3fafdbbcbe55 h1:6/iy4APnZZDmtDOxoqv3/eo5hxhpaA/M0ND75XqL7aA=
 github.com/livekit/psrpc v0.6.1-0.20250828235857-3fafdbbcbe55/go.mod h1:AuDC5uOoEjQJEc69v4Li3t77Ocz0e0NdjQEuFfO+vfk=
 github.com/mackerelio/go-osstat v0.2.6 h1:gs4U8BZeS1tjrL08tt5VUliVvSWP26Ai2Ob8Lr7f2i0=

pkg/rtc/clientinfo.go

@@ -45,6 +45,10 @@ func (c ClientInfo) isAndroid() bool {
 	return c.ClientInfo != nil && strings.EqualFold(c.ClientInfo.Os, "android")
 }
 
+func (c ClientInfo) isOBS() bool {
+	return c.ClientInfo != nil && strings.Contains(c.ClientInfo.Browser, "OBS")
+}
+
 func (c ClientInfo) SupportsAudioRED() bool {
 	return !c.isFirefox() && !c.isSafari()
 }

pkg/rtc/participant.go

@@ -32,6 +32,7 @@ import (
 	"github.com/pkg/errors"
 	"go.uber.org/atomic"
 	"go.uber.org/zap/zapcore"
+	"golang.org/x/exp/maps"
 	"google.golang.org/protobuf/proto"
 
 	"github.com/livekit/mediatransportutil/pkg/twcc"
@@ -305,7 +306,7 @@ type ParticipantImpl struct {
 	migrateState                atomic.Value // types.MigrateState
 	migratedTracksPublishedFuse core.Fuse
 
-	onClose            func(types.LocalParticipant)
+	onClose            map[string]func(types.LocalParticipant)
 	onClaimsChanged    func(participant types.LocalParticipant)
 	onICEConfigChanged func(participant types.LocalParticipant, iceConfig *livekit.ICEConfig)
 
@@ -359,6 +360,7 @@ func NewParticipant(params ParticipantParams) (*ParticipantImpl, error) {
 			joiningMessageFirstSeqs:       make(map[livekit.ParticipantID]uint32),
 			joiningMessageLastWrittenSeqs: make(map[livekit.ParticipantID]uint32),
 		},
+		onClose: make(map[string]func(types.LocalParticipant)),
 	}
 	p.setupSignalling()
 
@@ -1048,14 +1050,18 @@ func (p *ParticipantImpl) getOnLeave() func(types.LocalParticipant, types.Partic
 	return p.onLeave
 }
 
-func (p *ParticipantImpl) OnClose(callback func(types.LocalParticipant)) {
+func (p *ParticipantImpl) AddOnClose(key string, callback func(types.LocalParticipant)) {
 	if p.isClosed.Load() {
 		go callback(p)
 		return
 	}
 
 	p.lock.Lock()
-	p.onClose = callback
+	if callback == nil {
+		delete(p.onClose, key)
+	} else {
+		p.onClose[key] = callback
+	}
 	p.lock.Unlock()
 }
 
@@ -1505,10 +1511,10 @@ func (p *ParticipantImpl) Close(sendLeave bool, reason types.ParticipantCloseRea
 	// ensure this is synchronized
 	p.CloseSignalConnection(types.SignallingCloseReasonParticipantClose)
 	p.lock.RLock()
-	onClose := p.onClose
+	onClose := maps.Values(p.onClose)
 	p.lock.RUnlock()
-	if onClose != nil {
-		onClose(p)
+	for _, cb := range onClose {
+		cb(p)
 	}
 
 	// Close peer connections without blocking participant Close. If peer connections are gathering candidates
@@ -2851,9 +2857,14 @@ func (p *ParticipantImpl) addPendingTrackLocked(req *livekit.AddTrackRequest) *l
 
 	if len(req.SimulcastCodecs) == 0 {
 		// clients not supporting simulcast codecs, synthesise a codec
+		videoLayerMode := livekit.VideoLayer_MODE_UNUSED
+		if p.params.ClientInfo.isOBS() {
+			videoLayerMode = livekit.VideoLayer_ONE_SPATIAL_LAYER_PER_STREAM_INCOMPLETE_RTCP_SR
+		}
 		ti.Codecs = append(ti.Codecs, &livekit.SimulcastCodecInfo{
-			Cid:    req.Cid,
-			Layers: cloneLayers(req.Layers),
+			Cid:            req.Cid,
+			Layers:         cloneLayers(req.Layers),
+			VideoLayerMode: videoLayerMode,
 		})
 	} else {
 		seenCodecs := make(map[string]struct{})
@@ -2888,7 +2899,11 @@ func (p *ParticipantImpl) addPendingTrackLocked(req *livekit.AddTrackRequest) *l
 					if mime.IsMimeTypeStringSVC(mimeType) {
 						videoLayerMode = livekit.VideoLayer_MULTIPLE_SPATIAL_LAYERS_PER_STREAM
 					} else {
-						videoLayerMode = livekit.VideoLayer_ONE_SPATIAL_LAYER_PER_STREAM
+						if p.params.ClientInfo.isOBS() {
+							videoLayerMode = livekit.VideoLayer_ONE_SPATIAL_LAYER_PER_STREAM_INCOMPLETE_RTCP_SR
+						} else {
+							videoLayerMode = livekit.VideoLayer_ONE_SPATIAL_LAYER_PER_STREAM
+						}
 					}
 				}
 			} else if req.Type == livekit.TrackType_AUDIO {
@@ -3966,8 +3981,8 @@ func (p *ParticipantImpl) SupportsMoving() error {
 		return ErrMoveOldClientVersion
 	}
 
-	if kind := p.Kind(); kind == livekit.ParticipantInfo_EGRESS || kind == livekit.ParticipantInfo_AGENT {
-		return fmt.Errorf("%s participants cannot be moved", kind.String())
+	if kind := p.Kind(); kind == livekit.ParticipantInfo_EGRESS || kind == livekit.ParticipantInfo_AGENT || p.params.UseOneShotSignallingMode {
+		return fmt.Errorf("%s participants cannot be moved, one-shot signaling mode: %t", kind.String(), p.params.UseOneShotSignallingMode)
 	}
 
 	return nil
@@ -3977,10 +3992,10 @@ func (p *ParticipantImpl) MoveToRoom(params types.MoveToRoomParams) {
 	// fire onClose callback for original room
 	p.lock.Lock()
 	onClose := p.onClose
-	p.onClose = nil
+	p.onClose = make(map[string]func(types.LocalParticipant))
 	p.lock.Unlock()
-	if onClose != nil {
-		onClose(p)
+	for _, cb := range onClose {
+		cb(p)
 	}
 
 	for _, track := range p.GetPublishedTracks() {

pkg/rtc/types/interfaces.go

@@ -265,6 +265,12 @@ func (s SignallingCloseReason) String() string {
 	}
 }
 
+// ---------------------------------------------
+const (
+	ParticipantCloseKeyNormal = "normal"
+	ParticipantCloseKeyWHIP   = "whip"
+)
+
 // ---------------------------------------------
 
 //counterfeiter:generate . Participant
@@ -470,7 +476,7 @@ type LocalParticipant interface {
 	OnDataPacket(callback func(LocalParticipant, livekit.DataPacket_Kind, *livekit.DataPacket))
 	OnDataMessage(callback func(LocalParticipant, []byte))
 	OnSubscribeStatusChanged(fn func(publisherID livekit.ParticipantID, subscribed bool))
-	OnClose(callback func(LocalParticipant))
+	AddOnClose(key string, callback func(LocalParticipant))
 	OnClaimsChanged(callback func(LocalParticipant))
 	OnUpdateSubscriptions(func(
 		LocalParticipant,

pkg/service/roommanager.go

@@ -154,7 +154,11 @@ func NewLocalRoomManager(
 		return nil, err
 	}
 
-	r.whipServer, err = rpc.NewWHIPServer[livekit.NodeID](whipService{r}, bus, rpc.WithDefaultServerOptions(conf.PSRPC, logger.GetLogger()))
+	whipService, err := newWhipService(r)
+	if err != nil {
+		return nil, err
+	}
+	r.whipServer, err = rpc.NewWHIPServer[livekit.NodeID](whipService, bus, rpc.WithDefaultServerOptions(conf.PSRPC, logger.GetLogger()))
 	if err != nil {
 		return nil, err
 	}
@@ -548,7 +552,7 @@ func (r *RoomManager) StartSession(
 
 	clientMeta := &livekit.AnalyticsClientMeta{Region: r.currentNode.Region(), Node: string(r.currentNode.NodeID())}
 	r.telemetry.ParticipantJoined(ctx, protoRoom, participant.ToProto(), pi.Client, clientMeta, true)
-	participant.OnClose(func(p types.LocalParticipant) {
+	participant.AddOnClose(types.ParticipantCloseKeyNormal, func(p types.LocalParticipant) {
 		participantServerClosers.Close()
 
 		if err := r.roomStore.DeleteParticipant(ctx, room.Name(), p.Identity()); err != nil {

pkg/service/roommanager_service.go

@@ -5,20 +5,34 @@ import (
 	"fmt"
 	"time"
 
+	"github.com/pion/webrtc/v4"
+	"golang.org/x/sync/errgroup"
+	"google.golang.org/protobuf/types/known/emptypb"
+
 	"github.com/livekit/livekit-server/pkg/routing"
 	"github.com/livekit/livekit-server/pkg/rtc/types"
 	"github.com/livekit/livekit-server/pkg/telemetry/prometheus"
 	"github.com/livekit/protocol/livekit"
 	"github.com/livekit/protocol/logger"
 	"github.com/livekit/protocol/rpc"
 	"github.com/livekit/psrpc"
-	"github.com/pion/webrtc/v4"
-	"golang.org/x/sync/errgroup"
-	"google.golang.org/protobuf/types/known/emptypb"
 )
 
 type whipService struct {
 	*RoomManager
+
+	ingressRpcCli rpc.IngressHandlerClient
+}
+
+func newWhipService(rm *RoomManager) (*whipService, error) {
+	cli, err := rpc.NewIngressHandlerClient(rm.bus, rpc.WithDefaultClientOptions(logger.GetLogger()))
+	if err != nil {
+		return nil, err
+	}
+	return &whipService{
+		RoomManager:   rm,
+		ingressRpcCli: cli,
+	}, nil
 }
 
 func (s whipService) Create(ctx context.Context, req *rpc.WHIPCreateRequest) (*rpc.WHIPCreateResponse, error) {
@@ -97,6 +111,21 @@ func (s whipService) Create(ctx context.Context, req *rpc.WHIPCreateRequest) (*r
 		return nil, err
 	}
 
+	if req.FromIngress {
+		lp.AddOnClose(types.ParticipantCloseKeyWHIP, func(lp types.LocalParticipant) {
+			go func() {
+				lp.GetLogger().Debugw("whip service: notify participant closed")
+				_, err := s.ingressRpcCli.WHIPRTCConnectionNotify(context.Background(), string(lp.ID()), &rpc.WHIPRTCConnectionNotifyRequest{
+					ParticipantId: string(lp.ID()),
+					Closed:        true,
+				}, psrpc.WithRequestTimeout(rpc.DefaultPSRPCConfig.Timeout))
+				if err != nil {
+					lp.GetLogger().Warnw("whip service: could not notify ingress of participant closed", err)
+				}
+			}()
+		})
+	}
+
 	var iceServers []*livekit.ICEServer
 	apiKey, _, err := s.RoomManager.getFirstKeyPair()
 	if err != nil {
@@ -185,12 +214,19 @@ func (r whipParticipantService) DeleteSession(ctx context.Context, req *rpc.WHIP
 		return nil, ErrRoomNotFound
 	}
 
+	reason := types.ParticipantCloseReasonClientRequestLeave
 	lp := room.GetParticipantByID(livekit.ParticipantID(req.ParticipantId))
+	if lp == nil && req.FromSweeper && req.ParticipantId == "" {
+		lp = room.GetParticipant(livekit.ParticipantIdentity(req.ParticipantIdentity))
+		reason = types.ParticipantCloseReasonStale
+	}
+
 	if lp != nil {
+		lp.AddOnClose(types.ParticipantCloseKeyWHIP, nil)
 		room.RemoveParticipant(
 			lp.Identity(),
 			lp.ID(),
-			types.ParticipantCloseReasonClientRequestLeave,
+			reason,
 		)
 	}