Get to the point of establishing subscriber peer connection. (#3821)

Proper message id checks.

Some refactoring for common code in rtcv2service handlers.

Instantiate signalling based on synchronous local candidates flag, but
needs better defined code.

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-20250519131108-fb90f5acfded
-	github.com/livekit/protocol v1.39.4-0.20250724172257-afef3ef967cc
+	github.com/livekit/protocol v1.39.4-0.20250725083335-7313a8195a4b
 	github.com/livekit/psrpc v0.6.1-0.20250724161801-262a822e7cd7
 	github.com/mackerelio/go-osstat v0.2.5
 	github.com/magefile/mage v1.15.0

go.sum

@@ -167,8 +167,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-20250519131108-fb90f5acfded h1:ylZPdnlX1RW9Z15SD4mp87vT2D2shsk0hpLJwSPcq3g=
 github.com/livekit/mediatransportutil v0.0.0-20250519131108-fb90f5acfded/go.mod h1:mSNtYzSf6iY9xM3UX42VEI+STHvMgHmrYzEHPcdhB8A=
-github.com/livekit/protocol v1.39.4-0.20250724172257-afef3ef967cc h1:kF9ju/WEqtzMhI3P9hCUcgoh1mFEuq7DuWPkji6iq/M=
-github.com/livekit/protocol v1.39.4-0.20250724172257-afef3ef967cc/go.mod h1:YlgUxAegtU8jZ0tVXoIV/4fHeHqqLvS+6JnPKDbpFPU=
+github.com/livekit/protocol v1.39.4-0.20250725083335-7313a8195a4b h1:uxvoeGd0vmGDIL0JyLLV9h2o97tpt3rR9s4ikuLVz/g=
+github.com/livekit/protocol v1.39.4-0.20250725083335-7313a8195a4b/go.mod h1:YlgUxAegtU8jZ0tVXoIV/4fHeHqqLvS+6JnPKDbpFPU=
 github.com/livekit/psrpc v0.6.1-0.20250724161801-262a822e7cd7 h1:x50axcjXwfwnII7sMhJPyZ6f5LpPapZtsp75KJX8nIQ=
 github.com/livekit/psrpc v0.6.1-0.20250724161801-262a822e7cd7/go.mod h1:kmD+AZPkWu0MaXIMv57jhNlbiSZZ/Jx4bzlxBDVmJes=
 github.com/mackerelio/go-osstat v0.2.5 h1:+MqTbZUhoIt4m8qzkVoXUJg1EuifwlAJSk4Yl2GXh+o=

pkg/rtc/participant.go

@@ -313,8 +313,8 @@ type ParticipantImpl struct {
 	metricsCollector  *metric.MetricsCollector
 	metricsReporter   *metric.MetricsReporter
 
-	signalhandler signalling.ParticipantSignalHandler
 	signalling    signalling.ParticipantSignalling
+	signalhandler signalling.ParticipantSignalHandler
 	signaller     signalling.ParticipantSignaller
 
 	// loggers for publisher and subscriber
@@ -353,30 +353,7 @@ func NewParticipant(params ParticipantParams) (*ParticipantImpl, error) {
 			joiningMessageLastWrittenSeqs: make(map[livekit.ParticipantID]uint32),
 		},
 	}
-	p.signalhandler = signalling.NewSignalHandler(signalling.SignalHandlerParams{
-		Logger:      params.Logger,
-		Participant: p,
-	})
-	p.signalling = signalling.NewSignalling(signalling.SignallingParams{
-		Logger: params.Logger,
-	})
-	p.signaller = signalling.NewSignallerAsync(signalling.SignallerAsyncParams{
-		Logger:      params.Logger,
-		Participant: p,
-	})
-	/* SIGNALLING-V2-TODO: instantiate these based on signalling version and signal transport
-	p.signalhandler = signalling.NewSignalHandlerv2(signalling.SignalHandlerv2Params{
-		Logger:      params.Logger,
-		Participant: p,
-	})
-	p.signalling = signalling.NewSignallingv2(signalling.Signallingv2Params{
-		Logger: params.Logger,
-	})
-	p.signaller = signalling.NewSignallerv2Async(signalling.Signallerv2AsyncParams{
-		Logger:      params.Logger,
-		Participant: p,
-	})
-	*/
+	p.setupSignalling()
 
 	p.id.Store(params.SID)
 	p.dataChannelStats = telemetry.NewBytesTrackStats(
@@ -1841,7 +1818,35 @@ func (h PrimaryTransportHandler) OnFullyEstablished() {
 	h.p.onPrimaryTransportFullyEstablished()
 }
 
-// ----------------------------------------------------------
+func (p *ParticipantImpl) setupSignalling() {
+	// SIGNALLING-V2-TODO: do proper types to decide which signalling components to instantiate
+	if !p.params.SynchronousLocalCandidatesMode {
+		p.signalling = signalling.NewSignalling(signalling.SignallingParams{
+			Logger: p.params.Logger,
+		})
+		p.signalhandler = signalling.NewSignalHandler(signalling.SignalHandlerParams{
+			Logger:      p.params.Logger,
+			Participant: p,
+		})
+		p.signaller = signalling.NewSignallerAsync(signalling.SignallerAsyncParams{
+			Logger:      p.params.Logger,
+			Participant: p,
+		})
+	} else {
+		p.signalling = signalling.NewSignallingv2(signalling.Signallingv2Params{
+			Logger: p.params.Logger,
+		})
+		p.signalhandler = signalling.NewSignalHandlerv2(signalling.SignalHandlerv2Params{
+			Logger:      p.params.Logger,
+			Participant: p,
+			Signalling:  p.signalling,
+		})
+		p.signaller = signalling.NewSignallerv2Async(signalling.Signallerv2AsyncParams{
+			Logger:      p.params.Logger,
+			Participant: p,
+		})
+	}
+}
 
 func (p *ParticipantImpl) setupTransportManager() error {
 	p.twcc = twcc.NewTransportWideCCResponder()

pkg/rtc/signalling/signalhandlerv2.go

@@ -70,7 +70,6 @@ func (s *signalhandlerv2) HandleRequest(msg proto.Message) error {
 	switch msg := req.GetMessage().(type) {
 	case *livekit.Signalv2WireMessage_Envelope:
 		for _, clientMessage := range msg.Envelope.ClientMessages {
-			/* SIGNALLING-V2-TODO: uncommment once remote side sends proper messageId
 			sequencer := clientMessage.GetSequencer()
 			if sequencer == nil || sequencer.MessageId == 0 {
 				s.params.Logger.Warnw(
@@ -90,6 +89,7 @@ func (s *signalhandlerv2) HandleRequest(msg proto.Message) error {
 				continue
 			}
 
+			// SIGNALLING-V2-TODO: ask for replay if there are gaps
 			if lprmi != 0 && sequencer.MessageId != lprmi+1 {
 				s.params.Logger.Infow(
 					"gap in message stream",
@@ -97,7 +97,6 @@ func (s *signalhandlerv2) HandleRequest(msg proto.Message) error {
 					"current", clientMessage.Sequencer.MessageId,
 				)
 			}
-			*/
 
 			switch payload := clientMessage.GetMessage().(type) {
 			case *livekit.Signalv2ClientMessage_PublisherSdp:
@@ -115,11 +114,9 @@ func (s *signalhandlerv2) HandleRequest(msg proto.Message) error {
 				s.params.Participant.AddICECandidate(candidateInit, payload.Trickle.Target)
 			}
 
-			/* SIGNALLING-V2-TODO: uncomment once sequencer is implemented on both sides
 			s.lastProcessedRemoteMessageId.Store(sequencer.MessageId)
 			s.params.Signalling.AckMessageId(sequencer.LastProcessedRemoteMessageId)
 			s.params.Signalling.SetLastProcessedRemoteMessageId(sequencer.MessageId)
-			*/
 		}
 
 	case *livekit.Signalv2WireMessage_Fragment:

pkg/service/rtcv2service.go

@@ -73,7 +73,7 @@ func NewRTCv2Service(
 
 func (s *RTCv2Service) SetupRoutes(mux *http.ServeMux) {
 	mux.HandleFunc("POST "+cRTCv2Path, s.handlePost)
-	mux.HandleFunc("GET "+cRTCv2Path, s.validate)
+	mux.HandleFunc("GET "+cRTCv2ValidatePath, s.validate)
 	mux.HandleFunc("PATCH "+cRTCv2ParticipantIDPath, s.handleParticipantPatch)
 	mux.HandleFunc("DELETE "+cRTCv2ParticipantIDPath, s.handleParticipantDelete)
 }
@@ -212,31 +212,9 @@ func (s *RTCv2Service) handleParticipantPatch(w http.ResponseWriter, r *http.Req
 		return
 	}
 
-	claims := GetGrants(r.Context())
-	if claims == nil || claims.Video == nil {
-		HandleErrorJson(w, r, http.StatusUnauthorized, rtc.ErrPermissionDenied)
-		return
-	}
-
-	roomName, err := EnsureJoinPermission(r.Context())
+	roomName, participantIdentity, pID, code, err := getParams(r)
 	if err != nil {
-		HandleErrorJson(w, r, http.StatusUnauthorized, err)
-		return
-	}
-	if roomName == "" {
-		HandleErrorJson(w, r, http.StatusUnauthorized, ErrNoRoomName)
-		return
-	}
-
-	participantIdentity := livekit.ParticipantIdentity(claims.Identity)
-	if participantIdentity == "" {
-		HandleErrorJson(w, r, http.StatusUnauthorized, ErrIdentityEmpty)
-		return
-	}
-
-	pID := livekit.ParticipantID(r.PathValue("participant_id"))
-	if pID == "" {
-		HandleErrorJson(w, r, http.StatusBadRequest, ErrParticipantSidEmpty)
+		HandleErrorJson(w, r, code, err)
 		return
 	}
 
@@ -301,25 +279,9 @@ func (s *RTCv2Service) handleParticipantDelete(w http.ResponseWriter, r *http.Re
 		return
 	}
 
-	roomName, err := EnsureJoinPermission(r.Context())
+	roomName, participantIdentity, pID, code, err := getParams(r)
 	if err != nil {
-		HandleErrorJson(w, r, http.StatusUnauthorized, err)
-		return
-	}
-	if roomName == "" {
-		HandleErrorJson(w, r, http.StatusUnauthorized, ErrNoRoomName)
-		return
-	}
-
-	participantIdentity := livekit.ParticipantIdentity(claims.Identity)
-	if participantIdentity == "" {
-		HandleErrorJson(w, r, http.StatusUnauthorized, ErrIdentityEmpty)
-		return
-	}
-
-	pID := livekit.ParticipantID(r.PathValue("participant_id"))
-	if pID == "" {
-		HandleErrorJson(w, r, http.StatusBadRequest, ErrParticipantSidEmpty)
+		HandleErrorJson(w, r, code, err)
 		return
 	}
 
@@ -363,3 +325,30 @@ func getWireMessage(r *http.Request) (*livekit.Signalv2WireMessage, error) {
 
 	return wireMessage, nil
 }
+
+func getParams(r *http.Request) (livekit.RoomName, livekit.ParticipantIdentity, livekit.ParticipantID, int, error) {
+	claims := GetGrants(r.Context())
+	if claims == nil || claims.Video == nil {
+		return "", "", "", http.StatusUnauthorized, rtc.ErrPermissionDenied
+	}
+
+	roomName, err := EnsureJoinPermission(r.Context())
+	if err != nil {
+		return "", "", "", http.StatusUnauthorized, err
+	}
+	if roomName == "" {
+		return "", "", "", http.StatusUnauthorized, ErrNoRoomName
+	}
+
+	participantIdentity := livekit.ParticipantIdentity(claims.Identity)
+	if participantIdentity == "" {
+		return "", "", "", http.StatusUnauthorized, ErrIdentityEmpty
+	}
+
+	pID := livekit.ParticipantID(r.PathValue("participant_id"))
+	if pID == "" {
+		return "", "", "", http.StatusBadRequest, ErrParticipantSidEmpty
+	}
+
+	return roomName, participantIdentity, pID, http.StatusOK, nil
+}

pkg/service/utils.go

@@ -59,6 +59,7 @@ func HandleErrorJson(w http.ResponseWriter, r *http.Request, status int, err err
 	}{
 		Error: err.Error(),
 	})
+	w.Header().Add("Content-type", "application/json")
 }
 
 func boolValue(s string) bool {