diff --git a/config-sample.yaml b/config-sample.yaml index eba55b100..a930446b9 100644 --- a/config-sample.yaml +++ b/config-sample.yaml @@ -16,18 +16,25 @@ # for production setups, this port should be placed behind a load balancer with TLS port: 7880 -# Experimental receive-only Media over QUIC/WebTransport downstream. Supports -# the LiveKit H264 probe protocol and moq-lite-04 for @moq/net receive testing. +# Experimental Media over QUIC/WebTransport. Supports the LiveKit H264 probe +# protocol and moq-lite-04 for @moq/net receive testing. The ingest path is +# experimental and accepts one H264 Annex-B track per MoQ publisher session. # WebTransport uses HTTP/3 over UDP and requires TLS. In production, expose this # UDP port directly or through an HTTP/3-capable load balancer. # moq: # enabled: false +# ingest_enabled: false # port: 7883 # path: /moq/v1 # cert_file: /path/to/cert.pem # key_file: /path/to/key.pem # track_queue_size: 256 # cache_max_bytes: 2097152 +# ingest_queue_size: 256 +# ingest_max_frame_bytes: 2097152 +# ingest_read_timeout: 10s +# ingest_resume_grace: 30s +# ingest_max_resume_sessions: 1024 # write_timeout: 2s # when redis is set, LiveKit will automatically operate in a fully distributed fashion diff --git a/pkg/config/config.go b/pkg/config/config.go index a49740da9..49975ee7d 100644 --- a/pkg/config/config.go +++ b/pkg/config/config.go @@ -191,8 +191,18 @@ type MoQConfig struct { // dropped to keep latency bounded. TrackQueueSize int `yaml:"track_queue_size,omitempty"` // Max bytes retained for the cached keyframe access unit per track/layer. - CacheMaxBytes int `yaml:"cache_max_bytes,omitempty"` - WriteTimeout time.Duration `yaml:"write_timeout,omitempty"` + CacheMaxBytes int `yaml:"cache_max_bytes,omitempty"` + + // Experimental MoQ ingest path. When enabled, publishers can send H264 + // Annex-B access units over moq-lite and appear as normal LiveKit tracks. + IngestEnabled bool `yaml:"ingest_enabled,omitempty"` + IngestQueueSize int `yaml:"ingest_queue_size,omitempty"` + IngestMaxFrameBytes int `yaml:"ingest_max_frame_bytes,omitempty"` + IngestReadTimeout time.Duration `yaml:"ingest_read_timeout,omitempty"` + IngestResumeGrace time.Duration `yaml:"ingest_resume_grace,omitempty"` + IngestMaxResumeSessions int `yaml:"ingest_max_resume_sessions,omitempty"` + + WriteTimeout time.Duration `yaml:"write_timeout,omitempty"` } func (c MoQConfig) WithDefaults(development bool, bindAddresses []string) MoQConfig { @@ -205,6 +215,21 @@ func (c MoQConfig) WithDefaults(development bool, bindAddresses []string) MoQCon if c.CacheMaxBytes <= 0 { c.CacheMaxBytes = 2 * 1024 * 1024 } + if c.IngestQueueSize <= 0 { + c.IngestQueueSize = 256 + } + if c.IngestMaxFrameBytes <= 0 { + c.IngestMaxFrameBytes = 2 * 1024 * 1024 + } + if c.IngestReadTimeout <= 0 { + c.IngestReadTimeout = 10 * time.Second + } + if c.IngestResumeGrace <= 0 { + c.IngestResumeGrace = 30 * time.Second + } + if c.IngestMaxResumeSessions <= 0 { + c.IngestMaxResumeSessions = 1024 + } if c.WriteTimeout <= 0 { c.WriteTimeout = 2 * time.Second } @@ -529,10 +554,15 @@ var DefaultConfig = Config{ API: DefaultAPIConfig(), EnableDataTracks: true, MoQ: MoQConfig{ - Path: "/moq/v1", - TrackQueueSize: 256, - CacheMaxBytes: 2 * 1024 * 1024, - WriteTimeout: 2 * time.Second, + Path: "/moq/v1", + TrackQueueSize: 256, + CacheMaxBytes: 2 * 1024 * 1024, + IngestQueueSize: 256, + IngestMaxFrameBytes: 2 * 1024 * 1024, + IngestReadTimeout: 10 * time.Second, + IngestResumeGrace: 30 * time.Second, + IngestMaxResumeSessions: 1024, + WriteTimeout: 2 * time.Second, }, } diff --git a/pkg/rtc/participant.go b/pkg/rtc/participant.go index f5f459e4e..fcba2201c 100644 --- a/pkg/rtc/participant.go +++ b/pkg/rtc/participant.go @@ -1344,6 +1344,55 @@ func (p *ParticipantImpl) AddTrack(req *livekit.AddTrackRequest) { p.handlePendingRemoteTracks() } +// PublishSyntheticTrack creates a published media track from a non-WebRTC +// receiver. It is used by server-side ingest paths that already have encoded +// RTP packets and need normal LiveKit SFU fanout. +func (p *ParticipantImpl) PublishSyntheticTrack(req *livekit.AddTrackRequest, receiver sfu.TrackReceiver) (types.MediaTrack, error) { + if req == nil { + return nil, errors.New("add track request is nil") + } + if receiver == nil { + return nil, errors.New("track receiver is nil") + } + if !p.CanPublishSource(req.Source) { + return nil, errors.New("participant is not allowed to publish requested source") + } + if req.Type != livekit.TrackType_VIDEO { + return nil, errors.New("synthetic publish currently supports video tracks only") + } + if req.Cid == "" { + req = utils.CloneProto(req) + req.Cid = string(receiver.TrackID()) + } + + p.pendingTracksLock.Lock() + ti := p.addPendingTrackLocked(req) + if ti == nil { + p.pendingTracksLock.Unlock() + return nil, errors.New("could not create pending synthetic track") + } + ti.MimeType = receiver.Mime().String() + if len(ti.Codecs) == 1 && ti.Codecs[0].MimeType == "" { + ti.Codecs[0].MimeType = receiver.Mime().String() + } + for _, codec := range ti.Codecs { + if codec.MimeType == "" { + codec.MimeType = receiver.Mime().String() + } + } + if utils.TimedVersionFromProto(ti.Version).IsZero() { + ti.Version = p.params.VersionGenerator.Next().ToProto() + } + mt := p.addMediaTrack(req.Cid, ti) + p.dirty.Store(true) + p.pendingTracksLock.Unlock() + + mt.SetPotentialCodecs([]webrtc.RTPCodecParameters{receiver.Codec()}, receiver.HeaderExtensions()) + mt.SetupReceiver(receiver, 0, "") + p.handleTrackPublished(mt, false) + return mt, nil +} + func (p *ParticipantImpl) SetMigrateInfo( previousOffer, previousAnswer *webrtc.SessionDescription, mediaTracks []*livekit.TrackPublishedResponse, diff --git a/pkg/service/moq_ingest.go b/pkg/service/moq_ingest.go new file mode 100644 index 000000000..5f0959cce --- /dev/null +++ b/pkg/service/moq_ingest.go @@ -0,0 +1,387 @@ +// Copyright 2026 LiveKit, Inc. +// +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. + +package service + +import ( + "context" + "crypto/rand" + "encoding/base64" + "encoding/json" + "errors" + "net/http" + "sync" + "time" + + "github.com/livekit/protocol/auth" + "github.com/livekit/protocol/livekit" + "github.com/livekit/protocol/logger" + "github.com/livekit/protocol/utils" + "github.com/livekit/protocol/utils/guid" + "github.com/quic-go/quic-go/quicvarint" + webtransport "github.com/quic-go/webtransport-go" + + "github.com/livekit/livekit-server/pkg/config" + "github.com/livekit/livekit-server/pkg/routing" + "github.com/livekit/livekit-server/pkg/rtc" + "github.com/livekit/livekit-server/pkg/rtc/types" +) + +const moqLiteStreamPublishControl uint64 = 6 + +type moqIngestRegistryParams struct { + Config config.MoQConfig + Logger logger.Logger + RoomManager *RoomManager +} + +type moqIngestRegistry struct { + params moqIngestRegistryParams + + lock sync.Mutex + sessions map[string]*moqIngestSession +} + +type moqIngestSession struct { + registry *moqIngestRegistry + + sessionID string + resumeToken string + roomName livekit.RoomName + identity livekit.ParticipantIdentity + trackName string + trackID livekit.TrackID + + participant types.LocalParticipant + track types.MediaTrack + receiver *moqIngestReceiver + + lock sync.Mutex + generation uint64 + resumeDeadline time.Time + expireTimer *time.Timer + closed bool +} + +type moqIngestControlResponse struct { + PublishSessionID string `json:"publish_session_id"` + ResumeToken string `json:"resume_token"` + TrackSID string `json:"track_sid"` + NextSequence uint64 `json:"next_sequence"` + ResumeDeadlineMs int64 `json:"resume_deadline_ms"` +} + +func newMoQIngestRegistry(params moqIngestRegistryParams) *moqIngestRegistry { + return &moqIngestRegistry{ + params: params, + sessions: make(map[string]*moqIngestSession), + } +} + +func (s *MoQService) serveMoQLitePublishSession( + sess *webtransport.Session, + r *http.Request, + roomName livekit.RoomName, + claims *auth.ClaimGrants, +) { + params, err := parseMoQIngestPublishParams(r) + if err != nil { + _ = sess.CloseWithError(1, err.Error()) + return + } + + ingestSession, generation, err := s.ingest.Acquire(sess.Context(), roomName, livekit.ParticipantIdentity(claims.Identity), claims, params, r) + if err != nil { + _ = sess.CloseWithError(1, err.Error()) + return + } + defer ingestSession.ReleaseGeneration(generation) + + if err := writeMoQIngestControl(sess.Context(), sess, s.config.WriteTimeout, ingestSession.ControlResponse()); err != nil { + s.logger.Debugw("could not write moq ingest control response", "error", err) + return + } + + for { + stream, err := sess.AcceptUniStream(sess.Context()) + if err != nil { + if sess.Context().Err() == nil { + s.logger.Debugw("could not accept moq ingest stream", "error", err) + } + return + } + group, err := readMoQLitePublishGroup(stream, s.config.IngestMaxFrameBytes) + if err != nil { + s.logger.Debugw("could not read moq ingest group", "error", err) + continue + } + if !ingestSession.Enqueue(generation, group.sequence, group.payload) { + s.logger.Debugw("dropped moq ingest group", "trackID", ingestSession.trackID, "sequence", group.sequence) + } + } +} + +func (r *moqIngestRegistry) Acquire( + ctx context.Context, + roomName livekit.RoomName, + identity livekit.ParticipantIdentity, + claims *auth.ClaimGrants, + params moqIngestPublishParams, + req *http.Request, +) (*moqIngestSession, uint64, error) { + if params.ResumeToken != "" { + return r.resume(roomName, identity, params) + } + return r.create(ctx, roomName, identity, claims, params, req) +} + +func (r *moqIngestRegistry) create( + ctx context.Context, + roomName livekit.RoomName, + identity livekit.ParticipantIdentity, + claims *auth.ClaimGrants, + params moqIngestPublishParams, + req *http.Request, +) (*moqIngestSession, uint64, error) { + r.lock.Lock() + if len(r.sessions) >= r.params.Config.IngestMaxResumeSessions { + r.lock.Unlock() + return nil, 0, errors.New("too many moq ingest resume sessions") + } + r.lock.Unlock() + + connID := livekit.ConnectionID(guid.New("CO_")) + pi := routing.ParticipantInit{ + Identity: identity, + Name: livekit.ParticipantName(claims.Name), + AutoSubscribe: false, + Client: ParseClientInfo(req), + Grants: claims, + CreateRoom: &livekit.CreateRoomRequest{ + Name: string(roomName), + RoomPreset: claims.RoomPreset, + }, + AdaptiveStream: false, + DisableICELite: true, + } + if pi.Client.Protocol == 0 { + pi.Client.Protocol = types.CurrentProtocol + } + SetRoomConfiguration(pi.CreateRoom, claims.GetRoomConfiguration()) + + source := routing.NewNullMessageSource(connID) + sink := routing.NewNullMessageSink(connID) + if err := r.params.RoomManager.StartSession(ctx, pi, source, sink, true); err != nil { + return nil, 0, err + } + + room := r.params.RoomManager.GetRoom(ctx, roomName) + if room == nil { + return nil, 0, ErrRoomNotFound + } + participant := room.GetParticipant(identity) + if participant == nil { + return nil, 0, ErrParticipantNotFound + } + + trackID := livekit.TrackID(guid.New(utils.TrackPrefix)) + receiver, err := newMoQIngestReceiver(trackID, params.TrackName, params.Width, params.Height, params.FPS, r.params.Config, r.params.Logger) + if err != nil { + _ = participant.Close(false, types.ParticipantCloseReasonPublicationError, false) + return nil, 0, err + } + + impl, ok := participant.(*rtc.ParticipantImpl) + if !ok { + receiver.Close() + _ = participant.Close(false, types.ParticipantCloseReasonPublicationError, false) + return nil, 0, errors.New("moq ingest requires local rtc participant") + } + track, err := impl.PublishSyntheticTrack(&livekit.AddTrackRequest{ + Cid: string(trackID), + Name: params.TrackName, + Type: livekit.TrackType_VIDEO, + Width: params.Width, + Height: params.Height, + Source: livekit.TrackSource_CAMERA, + Stream: "camera", + Muted: false, + }, receiver) + if err != nil { + receiver.Close() + _ = participant.Close(false, types.ParticipantCloseReasonPublicationError, false) + return nil, 0, err + } + + resumeToken, err := randomResumeToken() + if err != nil { + track.Close(false) + receiver.Close() + _ = participant.Close(false, types.ParticipantCloseReasonPublicationError, false) + return nil, 0, err + } + + session := &moqIngestSession{ + registry: r, + sessionID: guid.New("MQ_"), + resumeToken: resumeToken, + roomName: roomName, + identity: identity, + trackName: params.TrackName, + trackID: track.ID(), + participant: participant, + track: track, + receiver: receiver, + } + + r.lock.Lock() + r.sessions[resumeToken] = session + r.lock.Unlock() + + return session, session.Bind(params.NextSequence), nil +} + +func (r *moqIngestRegistry) resume( + roomName livekit.RoomName, + identity livekit.ParticipantIdentity, + params moqIngestPublishParams, +) (*moqIngestSession, uint64, error) { + r.lock.Lock() + session := r.sessions[params.ResumeToken] + r.lock.Unlock() + if session == nil { + return nil, 0, errors.New("moq ingest resume token not found") + } + if session.roomName != roomName || session.identity != identity || session.trackName != params.TrackName { + return nil, 0, errors.New("moq ingest resume metadata mismatch") + } + if params.TrackID != "" && params.TrackID != session.trackID { + return nil, 0, errors.New("moq ingest resume track mismatch") + } + if !session.CanResume() { + return nil, 0, errors.New("moq ingest resume window expired") + } + return session, session.Bind(params.NextSequence), nil +} + +func (s *moqIngestSession) Bind(nextSequence uint64) uint64 { + s.lock.Lock() + defer s.lock.Unlock() + if s.expireTimer != nil { + s.expireTimer.Stop() + s.expireTimer = nil + } + s.generation++ + s.resumeDeadline = time.Time{} + s.receiver.SetNextSequence(nextSequence) + return s.generation +} + +func (s *moqIngestSession) ReleaseGeneration(generation uint64) { + s.lock.Lock() + defer s.lock.Unlock() + if s.closed || generation != s.generation { + return + } + s.resumeDeadline = time.Now().Add(s.registry.params.Config.IngestResumeGrace) + if s.expireTimer != nil { + s.expireTimer.Stop() + } + s.expireTimer = time.AfterFunc(s.registry.params.Config.IngestResumeGrace, func() { + s.Expire() + }) +} + +func (s *moqIngestSession) CanResume() bool { + s.lock.Lock() + defer s.lock.Unlock() + return !s.closed && (s.resumeDeadline.IsZero() || time.Now().Before(s.resumeDeadline)) +} + +func (s *moqIngestSession) Enqueue(generation uint64, sequence uint64, payload []byte) bool { + s.lock.Lock() + active := !s.closed && generation == s.generation + s.lock.Unlock() + if !active { + return false + } + return s.receiver.EnqueueAccessUnit(sequence, payload) +} + +func (s *moqIngestSession) ControlResponse() moqIngestControlResponse { + s.lock.Lock() + defer s.lock.Unlock() + deadline := time.Now().Add(s.registry.params.Config.IngestResumeGrace) + if !s.resumeDeadline.IsZero() { + deadline = s.resumeDeadline + } + return moqIngestControlResponse{ + PublishSessionID: s.sessionID, + ResumeToken: s.resumeToken, + TrackSID: string(s.trackID), + NextSequence: s.receiver.NextSequence(), + ResumeDeadlineMs: max(int64(time.Until(deadline)/time.Millisecond), 0), + } +} + +func (s *moqIngestSession) Expire() { + s.lock.Lock() + if s.closed || s.resumeDeadline.IsZero() || time.Now().Before(s.resumeDeadline) { + s.lock.Unlock() + return + } + s.closed = true + s.lock.Unlock() + + s.registry.lock.Lock() + if s.registry.sessions[s.resumeToken] == s { + delete(s.registry.sessions, s.resumeToken) + } + s.registry.lock.Unlock() + + s.track.Close(false) + s.receiver.Close() + _ = s.participant.Close(false, types.ParticipantCloseReasonSignalSourceClose, false) +} + +func writeMoQIngestControl( + ctx context.Context, + sess *webtransport.Session, + timeout time.Duration, + response moqIngestControlResponse, +) error { + stream, err := sess.OpenUniStreamSync(ctx) + if err != nil { + return err + } + defer func() { + _ = stream.Close() + }() + payload, err := json.Marshal(response) + if err != nil { + return err + } + var data []byte + data = quicvarint.Append(data, moqLiteStreamPublishControl) + data = appendMoQLiteMessage(data, payload) + return writeMoQLiteBytes(stream, timeout, data) +} + +func randomResumeToken() (string, error) { + var b [32]byte + if _, err := rand.Read(b[:]); err != nil { + return "", err + } + return base64.RawURLEncoding.EncodeToString(b[:]), nil +} diff --git a/pkg/service/moq_ingest_params.go b/pkg/service/moq_ingest_params.go new file mode 100644 index 000000000..0f9959942 --- /dev/null +++ b/pkg/service/moq_ingest_params.go @@ -0,0 +1,77 @@ +// Copyright 2026 LiveKit, Inc. +// +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. + +package service + +import ( + "fmt" + "net/http" + "strconv" + + "github.com/livekit/protocol/livekit" +) + +type moqIngestPublishParams struct { + TrackName string + Width uint32 + Height uint32 + FPS uint32 + ResumeToken string + TrackID livekit.TrackID + NextSequence uint64 +} + +func parseMoQIngestPublishParams(r *http.Request) (moqIngestPublishParams, error) { + q := r.URL.Query() + params := moqIngestPublishParams{ + TrackName: q.Get("track"), + Width: 640, + Height: 480, + FPS: 30, + } + if params.TrackName == "" { + params.TrackName = "camera" + } + if width := q.Get("width"); width != "" { + parsed, err := strconv.ParseUint(width, 10, 32) + if err != nil || parsed == 0 { + return params, fmt.Errorf("invalid width: %q", width) + } + params.Width = uint32(parsed) + } + if height := q.Get("height"); height != "" { + parsed, err := strconv.ParseUint(height, 10, 32) + if err != nil || parsed == 0 { + return params, fmt.Errorf("invalid height: %q", height) + } + params.Height = uint32(parsed) + } + if fps := q.Get("fps"); fps != "" { + parsed, err := strconv.ParseUint(fps, 10, 32) + if err != nil || parsed == 0 { + return params, fmt.Errorf("invalid fps: %q", fps) + } + params.FPS = uint32(parsed) + } + params.ResumeToken = q.Get("resume_token") + params.TrackID = livekit.TrackID(q.Get("track_sid")) + if next := q.Get("next_sequence"); next != "" { + parsed, err := strconv.ParseUint(next, 10, 64) + if err != nil { + return params, fmt.Errorf("invalid next_sequence: %q", next) + } + params.NextSequence = parsed + } + return params, nil +} diff --git a/pkg/service/moq_ingest_receiver.go b/pkg/service/moq_ingest_receiver.go new file mode 100644 index 000000000..477125b00 --- /dev/null +++ b/pkg/service/moq_ingest_receiver.go @@ -0,0 +1,329 @@ +// Copyright 2026 LiveKit, Inc. +// +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. + +package service + +import ( + "crypto/rand" + "encoding/binary" + "errors" + "io" + "sync" + "sync/atomic" + + "github.com/livekit/protocol/codecs/mime" + "github.com/livekit/protocol/livekit" + "github.com/livekit/protocol/logger" + "github.com/pion/rtcp" + "github.com/pion/rtp" + "github.com/pion/rtp/codecs" + "github.com/pion/webrtc/v4" + + "github.com/livekit/livekit-server/pkg/config" + "github.com/livekit/livekit-server/pkg/sfu" + "github.com/livekit/livekit-server/pkg/sfu/buffer" +) + +const ( + moqIngestClockRate = 90000 + moqIngestPayloadType = 102 + moqIngestRTPMTU = 1188 +) + +type moqIngestFrame struct { + sequence uint64 + payload []byte + keyFrame bool +} + +type moqIngestReceiver struct { + *sfu.ReceiverBase + + config config.MoQConfig + logger logger.Logger + buff *buffer.Buffer + + payloader codecs.H264Payloader + + ssrc uint32 + rtpSequence uint16 + rtpTimestamp uint32 + timestampStep uint32 + + frames chan moqIngestFrame + done chan struct{} + + lock sync.Mutex + closed bool + nextSequence uint64 + dropUntilIDR bool + + pliRequests atomic.Uint64 +} + +func newMoQIngestReceiver( + trackID livekit.TrackID, + trackName string, + width uint32, + height uint32, + fps uint32, + config config.MoQConfig, + lgr logger.Logger, +) (*moqIngestReceiver, error) { + if fps == 0 { + fps = 30 + } + ssrc, err := randomUint32() + if err != nil { + return nil, err + } + if ssrc == 0 { + ssrc = 1 + } + + codec := moqIngestH264Codec() + ti := &livekit.TrackInfo{ + Sid: string(trackID), + Type: livekit.TrackType_VIDEO, + Name: trackName, + Width: width, + Height: height, + Source: livekit.TrackSource_CAMERA, + Stream: "camera", + MimeType: mime.MimeTypeH264.String(), + Codecs: []*livekit.SimulcastCodecInfo{ + { + Cid: string(trackID), + MimeType: mime.MimeTypeH264.String(), + VideoLayerMode: livekit.VideoLayer_ONE_SPATIAL_LAYER_PER_STREAM, + Layers: []*livekit.VideoLayer{ + { + Quality: livekit.VideoQuality_HIGH, + Width: width, + Height: height, + Bitrate: 1_500_000, + SpatialLayer: 0, + }, + }, + }, + }, + } + + r := &moqIngestReceiver{ + config: config, + logger: lgr.WithValues("trackID", trackID, "trackName", trackName), + ssrc: ssrc, + timestampStep: uint32(moqIngestClockRate / fps), + frames: make(chan moqIngestFrame, config.IngestQueueSize), + done: make(chan struct{}), + dropUntilIDR: true, + } + r.ReceiverBase = sfu.NewReceiverBase( + sfu.ReceiverBaseParams{ + TrackID: trackID, + StreamID: "camera", + Kind: webrtc.RTPCodecTypeVideo, + Codec: codec, + Logger: r.logger, + IsSelfClosing: false, + }, + ti, + sfu.ReceiverCodecStateNormal, + ) + + r.buff = buffer.NewBuffer(ssrc, config.IngestQueueSize, 0) + r.buff.SetLogger(r.logger) + if err := r.buff.Bind(webrtc.RTPParameters{Codecs: []webrtc.RTPCodecParameters{codec}}, codec.RTPCodecCapability, 0); err != nil { + return nil, err + } + r.buff.OnRtcpFeedback(func(pkts []rtcp.Packet) { + for _, pkt := range pkts { + switch pkt.(type) { + case *rtcp.PictureLossIndication, *rtcp.FullIntraRequest: + r.pliRequests.Add(1) + r.logger.Debugw("moq ingest keyframe requested") + } + } + }) + r.AddBuffer(r.buff, 0) + r.StartBuffer(r.buff, 0) + + go r.run() + return r, nil +} + +func moqIngestH264Codec() webrtc.RTPCodecParameters { + return webrtc.RTPCodecParameters{ + RTPCodecCapability: webrtc.RTPCodecCapability{ + MimeType: webrtc.MimeTypeH264, + ClockRate: moqIngestClockRate, + SDPFmtpLine: "level-asymmetry-allowed=1;packetization-mode=1;profile-level-id=42e01f", + RTCPFeedback: []webrtc.RTCPFeedback{{Type: "nack"}, {Type: "nack", Parameter: "pli"}, {Type: "goog-remb"}}, + }, + PayloadType: moqIngestPayloadType, + } +} + +func (r *moqIngestReceiver) EnqueueAccessUnit(sequence uint64, payload []byte) bool { + if len(payload) == 0 || len(payload) > r.config.IngestMaxFrameBytes { + return false + } + frame := moqIngestFrame{ + sequence: sequence, + payload: cloneBytes(payload), + keyFrame: h264AnnexBHasIDR(payload), + } + + select { + case <-r.done: + return false + case r.frames <- frame: + return true + default: + } + + if !frame.keyFrame { + return false + } + select { + case <-r.done: + return false + case <-r.frames: + default: + } + select { + case <-r.done: + return false + case r.frames <- frame: + return true + default: + return false + } +} + +func (r *moqIngestReceiver) SetNextSequence(next uint64) { + r.lock.Lock() + if next > r.nextSequence { + r.nextSequence = next + } + r.dropUntilIDR = true + r.lock.Unlock() +} + +func (r *moqIngestReceiver) NextSequence() uint64 { + r.lock.Lock() + defer r.lock.Unlock() + return r.nextSequence +} + +func (r *moqIngestReceiver) Close() { + r.lock.Lock() + if r.closed { + r.lock.Unlock() + return + } + r.closed = true + close(r.done) + r.lock.Unlock() + r.ReceiverBase.Close("moq-ingest-close", true) +} + +func (r *moqIngestReceiver) run() { + for { + select { + case <-r.done: + return + case frame := <-r.frames: + if err := r.writeAccessUnit(frame); err != nil && !errors.Is(err, io.EOF) { + r.logger.Debugw("could not write moq ingest access unit", "error", err) + } + } + } +} + +func (r *moqIngestReceiver) writeAccessUnit(frame moqIngestFrame) error { + r.lock.Lock() + if r.closed { + r.lock.Unlock() + return io.EOF + } + if frame.sequence < r.nextSequence { + r.lock.Unlock() + return nil + } + if r.dropUntilIDR && !frame.keyFrame { + r.nextSequence = frame.sequence + 1 + r.lock.Unlock() + return nil + } + if frame.sequence > r.nextSequence { + r.nextSequence = frame.sequence + } + r.nextSequence = frame.sequence + 1 + if frame.keyFrame { + r.dropUntilIDR = false + } + + rtpTime := r.rtpTimestamp + r.rtpTimestamp += r.timestampStep + payloads := r.payloader.Payload(moqIngestRTPMTU, frame.payload) + packets := make([][]byte, 0, len(payloads)) + for i, payload := range payloads { + pkt := &rtp.Packet{ + Header: rtp.Header{ + Version: 2, + PayloadType: moqIngestPayloadType, + SequenceNumber: r.rtpSequence, + Timestamp: rtpTime, + SSRC: r.ssrc, + Marker: i == len(payloads)-1, + }, + Payload: payload, + } + r.rtpSequence++ + raw, err := pkt.Marshal() + if err != nil { + r.lock.Unlock() + return err + } + packets = append(packets, raw) + } + r.lock.Unlock() + + for _, raw := range packets { + if _, err := r.buff.Write(raw); err != nil { + return err + } + } + return nil +} + +func h264AnnexBHasIDR(payload []byte) bool { + for _, nalu := range splitAnnexBNALUs(payload) { + if len(nalu) != 0 && nalu[0]&0x1f == 5 { + return true + } + } + return false +} + +func randomUint32() (uint32, error) { + var b [4]byte + if _, err := rand.Read(b[:]); err != nil { + return 0, err + } + return binary.BigEndian.Uint32(b[:]), nil +} + +var _ sfu.TrackReceiver = (*moqIngestReceiver)(nil) diff --git a/pkg/service/moq_ingest_test.go b/pkg/service/moq_ingest_test.go new file mode 100644 index 000000000..25e111e86 --- /dev/null +++ b/pkg/service/moq_ingest_test.go @@ -0,0 +1,86 @@ +// Copyright 2026 LiveKit, Inc. +// +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. + +package service + +import ( + "bytes" + "net/http" + "testing" + + "github.com/quic-go/quic-go/quicvarint" + "github.com/stretchr/testify/require" +) + +func TestReadMoQLitePublishGroup(t *testing.T) { + payload := []byte{0, 0, 0, 1, 0x65, 0x88} + + var data []byte + data = quicvarint.Append(data, uint64(moqLiteStreamGroup)) + var group []byte + group = quicvarint.Append(group, 0) + group = quicvarint.Append(group, 42) + data = appendTestMoQLiteMessage(data, group) + data = quicvarint.Append(data, uint64(len(payload))) + data = append(data, payload...) + + result, err := readMoQLitePublishGroup(bytes.NewReader(data), 1024) + require.NoError(t, err) + require.Equal(t, uint64(42), result.sequence) + require.Equal(t, payload, result.payload) +} + +func TestReadMoQLitePublishGroupRejectsOversizePayload(t *testing.T) { + var data []byte + data = quicvarint.Append(data, uint64(moqLiteStreamGroup)) + var group []byte + group = quicvarint.Append(group, 0) + group = quicvarint.Append(group, 42) + data = appendTestMoQLiteMessage(data, group) + data = quicvarint.Append(data, 3) + data = append(data, []byte{1, 2, 3}...) + + _, err := readMoQLitePublishGroup(bytes.NewReader(data), 2) + require.ErrorContains(t, err, "payload too large") +} + +func TestParseMoQIngestPublishParamsDefaults(t *testing.T) { + req, err := http.NewRequest(http.MethodGet, "/moq/v1?role=publish", nil) + require.NoError(t, err) + + params, err := parseMoQIngestPublishParams(req) + require.NoError(t, err) + require.Equal(t, "camera", params.TrackName) + require.Equal(t, uint32(640), params.Width) + require.Equal(t, uint32(480), params.Height) + require.Equal(t, uint32(30), params.FPS) +} + +func TestParseMoQIngestPublishParamsRejectsMalformedMetadata(t *testing.T) { + req, err := http.NewRequest(http.MethodGet, "/moq/v1?width=wide", nil) + require.NoError(t, err) + + _, err = parseMoQIngestPublishParams(req) + require.ErrorContains(t, err, "invalid width") +} + +func TestH264AnnexBHasIDR(t *testing.T) { + require.True(t, h264AnnexBHasIDR([]byte{0, 0, 0, 1, 0x65, 0x88})) + require.False(t, h264AnnexBHasIDR([]byte{0, 0, 0, 1, 0x61, 0x88})) +} + +func appendTestMoQLiteMessage(dst []byte, payload []byte) []byte { + dst = quicvarint.Append(dst, uint64(len(payload))) + return append(dst, payload...) +} diff --git a/pkg/service/moq_ingest_wire.go b/pkg/service/moq_ingest_wire.go new file mode 100644 index 000000000..94c985d2c --- /dev/null +++ b/pkg/service/moq_ingest_wire.go @@ -0,0 +1,83 @@ +// Copyright 2026 LiveKit, Inc. +// +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. + +package service + +import ( + "bytes" + "fmt" + "io" + + "github.com/quic-go/quic-go/quicvarint" +) + +type moqLitePublishGroup struct { + sequence uint64 + payload []byte +} + +func readMoQLitePublishGroup(r io.Reader, maxPayloadBytes int) (moqLitePublishGroup, error) { + var group moqLitePublishGroup + reader := quicvarint.NewReader(r) + streamType, err := quicvarint.Read(reader) + if err != nil { + return group, err + } + if streamType != uint64(moqLiteStreamGroup) { + return group, fmt.Errorf("unexpected moq-lite publish stream type: %d", streamType) + } + + groupMeta, err := readMoQLiteMessageLimited(reader, moqLiteMaxMessageSize) + if err != nil { + return group, err + } + br := bytes.NewReader(groupMeta) + metaReader := quicvarint.NewReader(br) + if _, err = quicvarint.Read(metaReader); err != nil { + return group, err + } + group.sequence, err = quicvarint.Read(metaReader) + if err != nil { + return group, err + } + if br.Len() != 0 { + return group, fmt.Errorf("moq-lite publish group had %d trailing bytes", br.Len()) + } + + payloadSize, err := quicvarint.Read(reader) + if err != nil { + return group, err + } + if payloadSize > uint64(maxPayloadBytes) { + return group, fmt.Errorf("moq-lite publish payload too large: %d", payloadSize) + } + group.payload = make([]byte, int(payloadSize)) + _, err = io.ReadFull(reader, group.payload) + return group, err +} + +func readMoQLiteMessageLimited(r quicvarint.Reader, maxSize int) ([]byte, error) { + size, err := quicvarint.Read(r) + if err != nil { + return nil, err + } + if size > uint64(maxSize) { + return nil, fmt.Errorf("moq-lite message too large: %d", size) + } + data := make([]byte, int(size)) + if _, err := io.ReadFull(r, data); err != nil { + return nil, err + } + return data, nil +} diff --git a/pkg/service/moq_lite.go b/pkg/service/moq_lite.go index 2ff35c4db..d4056a46d 100644 --- a/pkg/service/moq_lite.go +++ b/pkg/service/moq_lite.go @@ -31,19 +31,6 @@ import ( webtransport "github.com/quic-go/webtransport-go" ) -const ( - moqLiteProtocol = "moq-lite-04" - - moqLiteStreamGroup byte = 0 - moqLiteStreamSubscribe uint64 = 2 - moqLiteStreamProbe uint64 = 4 - - moqLiteSubscribeOK uint64 = 0 - moqLiteSubscribeDrop uint64 = 1 - - moqLiteMaxMessageSize = 1024 * 1024 -) - type moqLiteSubscribe struct { ID uint64 Broadcast string diff --git a/pkg/service/moq_lite_wire.go b/pkg/service/moq_lite_wire.go new file mode 100644 index 000000000..2c8438d87 --- /dev/null +++ b/pkg/service/moq_lite_wire.go @@ -0,0 +1,28 @@ +// Copyright 2026 LiveKit, Inc. +// +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. + +package service + +const ( + moqLiteProtocol = "moq-lite-04" + + moqLiteStreamGroup byte = 0 + moqLiteStreamSubscribe uint64 = 2 + moqLiteStreamProbe uint64 = 4 + + moqLiteSubscribeOK uint64 = 0 + moqLiteSubscribeDrop uint64 = 1 + + moqLiteMaxMessageSize = 1024 * 1024 +) diff --git a/pkg/service/moq_service.go b/pkg/service/moq_service.go index bcdeb7866..78071749c 100644 --- a/pkg/service/moq_service.go +++ b/pkg/service/moq_service.go @@ -52,6 +52,7 @@ type MoQService struct { keyProvider auth.KeyProvider roomManager *RoomManager tracks *moqTrackRegistry + ingest *moqIngestRegistry logger logger.Logger lock sync.Mutex @@ -87,6 +88,11 @@ func NewMoQService(conf *config.Config, keyProvider auth.KeyProvider, roomManage Config: s.config, Logger: lgr, }) + s.ingest = newMoQIngestRegistry(moqIngestRegistryParams{ + Config: s.config, + Logger: lgr, + RoomManager: roomManager, + }) roomManager.AddOnRoomCreated(s.tracks.AttachRoom) return s, nil } @@ -202,7 +208,17 @@ func (s *MoQService) handleWebTransport(w http.ResponseWriter, r *http.Request, http.Error(w, err.Error(), http.StatusUnauthorized) return } - if claims.Video == nil || !claims.Video.RoomJoin || !claims.Video.GetCanSubscribe() { + role := r.URL.Query().Get("role") + if claims.Video == nil || !claims.Video.RoomJoin { + http.Error(w, ErrPermissionDenied.Error(), http.StatusForbidden) + return + } + if role == "publish" { + if !s.config.IngestEnabled || !claims.Video.GetCanPublishSource(livekit.TrackSource_CAMERA) { + http.Error(w, ErrPermissionDenied.Error(), http.StatusForbidden) + return + } + } else if !claims.Video.GetCanSubscribe() { http.Error(w, ErrPermissionDenied.Error(), http.StatusForbidden) return } @@ -237,8 +253,24 @@ func (s *MoQService) handleWebTransport(w http.ResponseWriter, r *http.Request, switch protocol := sess.SessionState().ApplicationProtocol; protocol { case moqLiteProtocol: + if role == "publish" { + if !s.config.IngestEnabled { + _ = sess.CloseWithError(1, "moq ingest is disabled") + return + } + s.serveMoQLitePublishSession(sess, r, roomName, claims) + return + } + if claims.Video == nil || !claims.Video.GetCanSubscribe() { + _ = sess.CloseWithError(1, ErrPermissionDenied.Error()) + return + } s.serveMoQLiteSession(sess, roomName, claims, layer) case "", moqWireProtocol: + if claims.Video == nil || !claims.Video.GetCanSubscribe() { + _ = sess.CloseWithError(1, ErrPermissionDenied.Error()) + return + } resolved, catalog, _, err := s.resolveTrack(sess.Context(), roomName, livekit.TrackID(r.URL.Query().Get("track_id")), claims) if err != nil { _ = sess.CloseWithError(1, err.Error())