moq experiment

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

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,
 	},
 }
 

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,

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
+}

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
+}

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)

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...)
+}

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
+}

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

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
+)

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())