Merge pull request #295 from kerberos-io/feature/optimise-hls-upload

feature/optimise-hls-upload

machinery/src/cloud/livehls.go

@@ -1,6 +1,7 @@
 package cloud
 
 import (
+	"os"
 	"time"
 
 	mqtt "github.com/eclipse/paho.mqtt.golang"
@@ -29,7 +30,6 @@ const hlsViewerTimeoutSeconds = 8
 // spamming the control plane.
 const hlsReadyReannounceSeconds = 2
 
-
 // HandleLiveStreamHLS drives the live HLS producer. It mirrors HandleLiveStreamSD:
 // it reads the camera's packet stream from a Latest() cursor, and while a viewer
 // is active (kept alive via communication.HandleLiveHLS) it muxes the packets
@@ -38,6 +38,14 @@ const hlsReadyReannounceSeconds = 2
 //
 // A session is created lazily on the first keyframe seen while a viewer is active
 // and torn down once viewers go away, so an idle camera produces no live traffic.
+//
+// By default (AGENT_LIVE_HLS_PREWARM unset or != "false") the agent instead keeps
+// one long-lived session muxing continuously into a small in-memory ring buffer
+// while idle (uploading nothing) and, the moment a viewer arrives, flushes the
+// already-encoded init + most-recent segment(s) and starts uploading live. This
+// trades a little idle CPU for a near-instant "requesting stream", so viewers no
+// longer wait a full GOP for the first segment to be cut. Set
+// AGENT_LIVE_HLS_PREWARM=false to fall back to the lazy on-demand path above.
 func HandleLiveStreamHLS(livestreamCursor *packets.QueueCursor, configuration *models.Configuration, communication *models.Communication, mqttClient mqtt.Client, _ capture.RTSPClient) {
 
 	log.Log.Debug("cloud.HandleLiveStreamHLS(): started")
@@ -78,6 +86,29 @@ func HandleLiveStreamHLS(livestreamCursor *packets.QueueCursor, configuration *m
 	width := uint16(config.Capture.IPCamera.Width)
 	height := uint16(config.Capture.IPCamera.Height)
 
+	// prewarm keeps a single long-lived session muxing into an in-memory ring
+	// buffer while idle and flushes it the instant a viewer arrives, eliminating
+	// the per-request GOP wait. Enabled by default; set AGENT_LIVE_HLS_PREWARM=false
+	// to fall back to the lazy on-demand path.
+	prewarm := os.Getenv("AGENT_LIVE_HLS_PREWARM") != "false"
+	if prewarm {
+		log.Log.Info("cloud.HandleLiveStreamHLS(): live HLS prewarm ENABLED (set AGENT_LIVE_HLS_PREWARM=false to disable)")
+	} else {
+		log.Log.Info("cloud.HandleLiveStreamHLS(): live HLS prewarm DISABLED (AGENT_LIVE_HLS_PREWARM=false)")
+	}
+
+	// lowLatency enables LL-HLS: each segment is sliced into CMAF parts shipped the
+	// instant they close and advertised via #EXT-X-PART, taking glass-to-glass HLS
+	// latency from ~4-6s down to ~1-2s. Enabled by default; set
+	// AGENT_LIVE_HLS_LOW_LATENCY=false to fall back to whole-segment HLS.
+	partTargetMs := uint64(0)
+	if os.Getenv("AGENT_LIVE_HLS_LOW_LATENCY") != "false" {
+		partTargetMs = livehls.DefaultPartTargetMs
+		log.Log.Info("cloud.HandleLiveStreamHLS(): live HLS low-latency (LL-HLS) ENABLED (set AGENT_LIVE_HLS_LOW_LATENCY=false to disable)")
+	} else {
+		log.Log.Info("cloud.HandleLiveStreamHLS(): live HLS low-latency (LL-HLS) DISABLED (AGENT_LIVE_HLS_LOW_LATENCY=false)")
+	}
+
 	var session *livehls.Session
 	lastViewerRequest := int64(0)
 	lastReadyAnnounce := int64(0)
@@ -97,7 +128,10 @@ func HandleLiveStreamHLS(livestreamCursor *packets.QueueCursor, configuration *m
 			// fired when this session's first segment landed. Re-announce (throttled)
 			// so late/refreshed viewers learn the active session id; the frontend
 			// dedupes by session id, so this is a no-op for viewers already playing.
-			if session != nil && session.IsReady() && now-lastReadyAnnounce >= hlsReadyReannounceSeconds {
+			// UploadsActive() is always true for the on-demand path; for prewarm it
+			// suppresses a stale re-announce while idle (the flush-on-arrival path
+			// below announces once the buffer has actually been shipped).
+			if session != nil && session.IsReady() && session.UploadsActive() && now-lastReadyAnnounce >= hlsReadyReannounceSeconds {
 				publishHLSReady(configuration, mqttClient, hubKey, deviceId, session.SessionID())
 				lastReadyAnnounce = now
 			}
@@ -106,6 +140,55 @@ func HandleLiveStreamHLS(livestreamCursor *packets.QueueCursor, configuration *m
 
 		viewerActive := now-lastViewerRequest <= hlsViewerTimeoutSeconds
 
+		if prewarm {
+			// Keep one long-lived session muxing into the ring buffer. Create it on
+			// the first keyframe (so the buffer opens on a random-access point) and
+			// never tear it down for idleness; uploads, not muxing, are what we gate
+			// on viewer presence.
+			if session == nil {
+				if len(pkt.Data) == 0 || !pkt.IsVideo || !pkt.IsKeyFrame {
+					continue
+				}
+				session = livehls.NewSession(publisher, livehls.SessionOptions{
+					Codec:          pkt.Codec,
+					SPSNALUs:       config.Capture.IPCamera.SPSNALUs,
+					PPSNALUs:       config.Capture.IPCamera.PPSNALUs,
+					VPSNALUs:       config.Capture.IPCamera.VPSNALUs,
+					Width:          width,
+					Height:         height,
+					PartTargetMs:   partTargetMs,
+					StartBuffering: true,
+				})
+				session.SetOnReady(func(sessionID string) {
+					log.Log.Info("cloud.HandleLiveStreamHLS(): live HLS session ready, announcing " + sessionID)
+					publishHLSReady(configuration, mqttClient, hubKey, deviceId, sessionID)
+					lastReadyAnnounce = time.Now().Unix()
+				})
+				log.Log.Info("cloud.HandleLiveStreamHLS(): prewarming live HLS session " + session.SessionID())
+			}
+
+			if viewerActive {
+				// Activating flushes the cached init + buffered segment(s). onReady
+				// announces the first-ever readiness; on a later re-activation it has
+				// already fired, so announce here (throttled, so the first activation
+				// does not double up) once the buffer has actually been shipped.
+				if session.SetUploadsActive(true) && session.IsReady() && now-lastReadyAnnounce >= hlsReadyReannounceSeconds {
+					publishHLSReady(configuration, mqttClient, hubKey, deviceId, session.SessionID())
+					lastReadyAnnounce = now
+				}
+			} else {
+				// No viewer: keep muxing into the buffer but stop uploading.
+				session.SetUploadsActive(false)
+			}
+
+			if len(pkt.Data) > 0 && pkt.IsVideo {
+				if err := session.WritePacket(pkt); err != nil {
+					log.Log.Error("cloud.HandleLiveStreamHLS(): " + err.Error())
+				}
+			}
+			continue
+		}
+
 		if !viewerActive {
 			// No viewer: stop and discard the session so we stop shipping segments.
 			if session != nil {
@@ -127,12 +210,13 @@ func HandleLiveStreamHLS(livestreamCursor *packets.QueueCursor, configuration *m
 				continue
 			}
 			session = livehls.NewSession(publisher, livehls.SessionOptions{
-				Codec:    pkt.Codec,
-				SPSNALUs: config.Capture.IPCamera.SPSNALUs,
-				PPSNALUs: config.Capture.IPCamera.PPSNALUs,
-				VPSNALUs: config.Capture.IPCamera.VPSNALUs,
-				Width:    width,
-				Height:   height,
+				Codec:        pkt.Codec,
+				SPSNALUs:     config.Capture.IPCamera.SPSNALUs,
+				PPSNALUs:     config.Capture.IPCamera.PPSNALUs,
+				VPSNALUs:     config.Capture.IPCamera.VPSNALUs,
+				Width:        width,
+				Height:       height,
+				PartTargetMs: partTargetMs,
 			})
 			session.SetOnReady(func(sessionID string) {
 				log.Log.Info("cloud.HandleLiveStreamHLS(): live HLS session ready, announcing " + sessionID)

machinery/src/cloud/livehls/publisher.go

@@ -57,6 +57,11 @@ const (
 	headerLiveName      = "X-Kerberos-Live-Name"
 	headerLiveSequence  = "X-Kerberos-Live-Sequence"
 	headerLiveDuration  = "X-Kerberos-Live-Duration"
+	// Low-latency (LL-HLS) part headers. A part belongs to media segment
+	// X-Kerberos-Live-Sequence and is the X-Kerberos-Live-Part-th chunk within it;
+	// X-Kerberos-Live-Part-Independent flags a part that starts on a keyframe.
+	headerLivePart            = "X-Kerberos-Live-Part"
+	headerLivePartIndependent = "X-Kerberos-Live-Part-Independent"
 
 	// defaultPublishTimeout bounds a single segment upload. A live segment that
 	// cannot be delivered within roughly its own duration is stale, so the upload
@@ -136,12 +141,33 @@ func (p *Publisher) PublishSegment(ctx context.Context, sessionID string, seg vi
 	})
 }
 
+// PublishPart uploads one CMAF partial segment (LL-HLS). The part is named
+// seg-<segment>.<part>.m4s and carries its segment sequence, part index,
+// independence flag and duration in headers so hub-api can advertise it via
+// #EXT-X-PART and reconstruct the full segment by concatenating its parts.
+func (p *Publisher) PublishPart(ctx context.Context, sessionID string, part video.LivePart) error {
+	return p.post(ctx, postParams{
+		sessionID:   sessionID,
+		name:        fmt.Sprintf("seg-%d.%d.m4s", part.SegmentSeq, part.PartIndex),
+		sequence:    part.SegmentSeq,
+		durationMs:  part.DurationMs,
+		partIndex:   part.PartIndex,
+		independent: part.Independent,
+		hasPart:     true,
+		contentType: contentTypeSegment,
+		body:        part.Data,
+	})
+}
+
 type postParams struct {
 	sessionID   string
 	name        string
 	sequence    uint32
 	durationMs  uint64
 	hasSegment  bool
+	partIndex   uint32
+	independent bool
+	hasPart     bool
 	contentType string
 	body        []byte
 }
@@ -165,10 +191,18 @@ func (p *Publisher) post(ctx context.Context, params postParams) error {
 	req.Header.Set(headerStorageDevice, p.cfg.DeviceKey)
 	req.Header.Set(headerLiveSession, params.sessionID)
 	req.Header.Set(headerLiveName, params.name)
-	if params.hasSegment {
+	if params.hasSegment || params.hasPart {
 		req.Header.Set(headerLiveSequence, strconv.FormatUint(uint64(params.sequence), 10))
 		req.Header.Set(headerLiveDuration, strconv.FormatUint(params.durationMs, 10))
 	}
+	if params.hasPart {
+		req.Header.Set(headerLivePart, strconv.FormatUint(uint64(params.partIndex), 10))
+		independent := "0"
+		if params.independent {
+			independent = "1"
+		}
+		req.Header.Set(headerLivePartIndependent, independent)
+	}
 	req.Header.Set(headerHubPublicKey, p.cfg.HubKey)
 	req.Header.Set(headerHubPrivateKey, p.cfg.HubPrivateKey)
 	req.Header.Set(headerHubRegion, p.cfg.Region)

machinery/src/cloud/livehls/session.go

@@ -18,6 +18,12 @@ import (
 // large enough that per-segment HTTP overhead is negligible.
 const DefaultTargetSegmentMs = 2000
 
+// DefaultPartTargetMs is the nominal LL-HLS part length used when low latency is
+// enabled. ~300ms parts yield ~6-7 parts per 2s segment; with the playlist's
+// PART-HOLD-BACK at ~3x the part target this lands glass-to-glass latency around
+// 1-2s (versus ~4-6s for whole-segment HLS).
+const DefaultPartTargetMs = 300
+
 // Session ties a video.LiveSegmenter to a Publisher: it converts capture packets
 // into CMAF segments and ships each one to hub-api. Exactly one init segment is
 // delivered per session (re-attempted until it lands), after which media
@@ -43,6 +49,21 @@ type Session struct {
 	lastInitAt time.Time
 	readyFired bool
 	onReady    func(sessionID string)
+
+	// uploadsActive gates whether the init and completed segments are shipped to
+	// hub-api. It is true for the default on-demand path. The prewarm path starts
+	// it false so the session keeps muxing into bufferedSegments without producing
+	// any live traffic until a viewer actually arrives; see SetUploadsActive.
+	uploadsActive bool
+	// bufferedSegments is the in-memory ring buffer (the most recent
+	// prewarmMaxBufferedSegments segments) kept while uploadsActive is false, so a
+	// viewer that arrives can be served an already-encoded segment immediately
+	// instead of waiting a full GOP for the next one to be cut.
+	bufferedSegments []video.LiveSegment
+	// bufferedParts is the LL-HLS counterpart of bufferedSegments: while idle it
+	// retains the parts of the most recent (prewarmMaxBufferedSegments+1) segments,
+	// pruned a WHOLE segment at a time so a flushed segment is never partial.
+	bufferedParts []video.LivePart
 }
 
 // SessionOptions configures a live HLS session.
@@ -54,6 +75,16 @@ type SessionOptions struct {
 	Width           uint16   // encoded width (for the avcC fallback path)
 	Height          uint16   // encoded height
 	TargetSegmentMs uint64   // 0 => DefaultTargetSegmentMs
+	// PartTargetMs, when > 0, enables LL-HLS: each segment is additionally sliced
+	// into ~PartTargetMs CMAF parts that are published (and advertised via
+	// #EXT-X-PART) the instant they close, for ~1-2s glass-to-glass latency. 0
+	// keeps the classic whole-segment path.
+	PartTargetMs uint64
+	// StartBuffering starts the session in prewarm (buffer-only) mode: it muxes
+	// segments into an in-memory ring buffer but uploads nothing until
+	// SetUploadsActive(true) is called. Default false => uploads are live
+	// immediately (the on-demand path's behaviour).
+	StartBuffering bool
 }
 
 // NewSession builds a session with a fresh random id and wires the segmenter's
@@ -65,11 +96,16 @@ func NewSession(publisher *Publisher, opts SessionOptions) *Session {
 	}
 	seg := video.NewLiveSegmenter(opts.Codec, opts.SPSNALUs, opts.PPSNALUs, opts.VPSNALUs, target)
 	seg.SetDimensions(opts.Width, opts.Height)
+	if opts.PartTargetMs > 0 {
+		seg.EnableLowLatency(opts.PartTargetMs)
+	}
 
 	s := &Session{
 		id:        newSessionID(),
 		publisher: publisher,
 		segmenter: seg,
+		// Uploads are live by default; the prewarm path opts into buffer-only mode.
+		uploadsActive: !opts.StartBuffering,
 		newContext: func() (context.Context, context.CancelFunc) {
 			return context.WithTimeout(context.Background(), defaultPublishTimeout)
 		},
@@ -82,8 +118,13 @@ func NewSession(publisher *Publisher, opts SessionOptions) *Session {
 	seg.OnInit = func(initBytes []byte) error {
 		s.mu.Lock()
 		s.initBytes = append([]byte(nil), initBytes...)
+		active := s.uploadsActive
 		s.mu.Unlock()
-		s.publishInitIfNeeded()
+		// While prewarming we cache the init in memory but ship nothing; it is
+		// uploaded on the first SetUploadsActive(true) flush.
+		if active {
+			s.publishInitIfNeeded()
+		}
 		return nil
 	}
 
@@ -91,6 +132,15 @@ func NewSession(publisher *Publisher, opts SessionOptions) *Session {
 	// init segment has landed (a media segment is useless without it), and we fire
 	// OnReady after the first successfully shipped segment.
 	seg.OnSegment = func(segment video.LiveSegment) error {
+		s.mu.Lock()
+		active := s.uploadsActive
+		s.mu.Unlock()
+		if !active {
+			// Prewarm: retain the most recent segments in memory but upload nothing
+			// until a viewer arrives (SetUploadsActive flushes them).
+			s.bufferSegment(segment)
+			return nil
+		}
 		if !s.publishInitIfNeeded() {
 			log.Log.Warning("livehls.Session: dropping segment " +
 				fmt.Sprintf("%d", segment.SequenceNumber) + " because init has not been delivered yet")
@@ -109,6 +159,36 @@ func NewSession(publisher *Publisher, opts SessionOptions) *Session {
 		return nil
 	}
 
+	// In LL-HLS mode the segmenter emits parts (not whole segments); ship each one
+	// the instant it closes. Mirrors OnSegment: buffer while prewarming, otherwise
+	// publish after the init has landed and fire OnReady on the first part.
+	if opts.PartTargetMs > 0 {
+		seg.OnPart = func(part video.LivePart) error {
+			s.mu.Lock()
+			active := s.uploadsActive
+			s.mu.Unlock()
+			if !active {
+				s.bufferPart(part)
+				return nil
+			}
+			if !s.publishInitIfNeeded() {
+				log.Log.Warning("livehls.Session: dropping part " +
+					fmt.Sprintf("%d.%d", part.SegmentSeq, part.PartIndex) +
+					" because init has not been delivered yet")
+				return nil
+			}
+			ctx, cancel := s.newContext()
+			defer cancel()
+			if err := s.publisher.PublishPart(ctx, s.id, part); err != nil {
+				log.Log.Warning("livehls.Session: " + err.Error())
+				return nil
+			}
+			s.fireReadyOnce()
+			s.refreshInitIfStale()
+			return nil
+		}
+	}
+
 	return s
 }
 
@@ -136,6 +216,128 @@ func (s *Session) SetOnReady(fn func(sessionID string)) {
 	s.mu.Unlock()
 }
 
+// prewarmMaxBufferedSegments is how many of the most recent completed segments
+// the prewarm path keeps in memory while idle and flushes to a viewer on arrival.
+// One segment keeps startup instant (the viewer immediately gets a playable
+// segment) while starting as close to the live edge as possible, so the HLS view
+// tracks the WebRTC/live edge instead of opening several seconds behind; hls.js
+// then converges to the edge via maxLiveSyncPlaybackRate. Raising it trades
+// latency-from-live for a little more startup cushion.
+const prewarmMaxBufferedSegments = 1
+
+// SetUploadsActive toggles whether the session ships its init and segments to
+// hub-api, and reports whether this call flipped it from inactive to active.
+//
+// While uploads are inactive the session keeps muxing capture packets into an
+// in-memory ring buffer (the cached init plus the most recent
+// prewarmMaxBufferedSegments segments) but uploads nothing, so an idle camera
+// produces no live traffic. Switching from inactive to active immediately
+// flushes the cached init and buffered segments so a viewer can start almost
+// instantly instead of waiting a full GOP for the next segment to be cut.
+// Switching from active to inactive resets the init-published flag so the next
+// activation re-uploads the init (it may have aged out of the hub's short-TTL
+// live window while idle). All other transitions are no-ops. Driven from the
+// live-stream goroutine; not safe for concurrent use.
+func (s *Session) SetUploadsActive(active bool) bool {
+	s.mu.Lock()
+	if s.uploadsActive == active {
+		s.mu.Unlock()
+		return false
+	}
+	s.uploadsActive = active
+	if !active {
+		// Going idle: force the next activation to re-deliver the init segment,
+		// which may have expired from the hub live window while nobody was watching.
+		s.initPublished = false
+		s.mu.Unlock()
+		return false
+	}
+	// Inactive -> active: take the cached buffered segments/parts and flush them
+	// outside the lock (the publish calls take their own time and re-acquire the
+	// mutex).
+	buffered := s.bufferedSegments
+	bufferedParts := s.bufferedParts
+	s.bufferedSegments = nil
+	s.bufferedParts = nil
+	s.mu.Unlock()
+
+	// Deliver the init first; media segments are useless without it.
+	for i := range buffered {
+		if !s.publishInitIfNeeded() {
+			break
+		}
+		ctx, cancel := s.newContext()
+		if err := s.publisher.PublishSegment(ctx, s.id, buffered[i]); err != nil {
+			log.Log.Warning("livehls.Session: prewarm flush: " + err.Error())
+			cancel()
+			continue
+		}
+		cancel()
+		s.fireReadyOnce()
+		s.refreshInitIfStale()
+	}
+	// LL-HLS: flush the buffered parts in order (oldest first) so the viewer gets a
+	// playable, near-live window immediately.
+	for i := range bufferedParts {
+		if !s.publishInitIfNeeded() {
+			break
+		}
+		ctx, cancel := s.newContext()
+		if err := s.publisher.PublishPart(ctx, s.id, bufferedParts[i]); err != nil {
+			log.Log.Warning("livehls.Session: prewarm flush (part): " + err.Error())
+			cancel()
+			continue
+		}
+		cancel()
+		s.fireReadyOnce()
+		s.refreshInitIfStale()
+	}
+	return true
+}
+
+// UploadsActive reports whether the session is currently shipping segments (as
+// opposed to buffering them while prewarming). Always true for the on-demand
+// path.
+func (s *Session) UploadsActive() bool {
+	s.mu.Lock()
+	defer s.mu.Unlock()
+	return s.uploadsActive
+}
+
+// bufferSegment appends a completed segment to the in-memory prewarm ring buffer,
+// discarding the oldest so at most prewarmMaxBufferedSegments are retained.
+func (s *Session) bufferSegment(seg video.LiveSegment) {
+	s.mu.Lock()
+	s.bufferedSegments = append(s.bufferedSegments, seg)
+	if overflow := len(s.bufferedSegments) - prewarmMaxBufferedSegments; overflow > 0 {
+		// Drop the oldest segment(s) and shrink the backing array so retained bytes
+		// stay bounded.
+		s.bufferedSegments = append([]video.LiveSegment(nil), s.bufferedSegments[overflow:]...)
+	}
+	s.mu.Unlock()
+}
+
+// bufferPart appends a part to the LL-HLS prewarm ring buffer, pruning whole
+// older segments (never individual parts) so the retained window always consists
+// of complete segments plus the in-progress one. Pruning on a part-0 boundary
+// keeps at most prewarmMaxBufferedSegments fully-buffered segments behind the
+// current one, which guarantees a flushed segment can be reconstructed in full.
+func (s *Session) bufferPart(part video.LivePart) {
+	s.mu.Lock()
+	s.bufferedParts = append(s.bufferedParts, part)
+	if part.PartIndex == 0 && part.SegmentSeq > uint32(prewarmMaxBufferedSegments) {
+		minSeg := part.SegmentSeq - uint32(prewarmMaxBufferedSegments)
+		kept := make([]video.LivePart, 0, len(s.bufferedParts))
+		for _, p := range s.bufferedParts {
+			if p.SegmentSeq >= minSeg {
+				kept = append(kept, p)
+			}
+		}
+		s.bufferedParts = kept
+	}
+	s.mu.Unlock()
+}
+
 // WritePacket feeds one capture packet into the segmenter. Non-video packets are
 // ignored (the spike is video-only). The decode timestamp is derived exactly as
 // the recording muxer does: DTS = PTS - compositionOffset, with the composition

machinery/src/video/livehls.go

@@ -81,8 +81,30 @@ type LiveSegmenter struct {
 	// OnInit is invoked exactly once with the encoded init segment bytes before
 	// the first media segment is emitted. Optional.
 	OnInit func(initBytes []byte) error
-	// OnSegment is invoked once per completed media segment. Optional.
+	// OnSegment is invoked once per completed media segment. Optional. It is left
+	// unused in low-latency mode (see OnPart).
 	OnSegment func(seg LiveSegment) error
+
+	// --- Low-latency (LL-HLS) partial-segment mode ---
+	//
+	// When partTargetMs > 0 the segmenter additionally slices each segment into
+	// ~partTargetMs CMAF "parts" (chunks) and emits them via OnPart the instant
+	// each one closes, instead of waiting for the whole segment. The classic
+	// per-segment OnSegment path above is left untouched (and unused) in this mode.
+	// Each part is one mp4ff fragment (moof+mdat); part 0 of a segment also carries
+	// the CMAF styp, so concatenating a segment's parts yields one valid segment.
+	partTargetMs uint64
+	// partFrag is the open part's fragment; partIndex is its 0-based index within
+	// the current segment; fragSeq is the globally monotonic moof sequence number
+	// shared across all parts (MSE wants increasing moof sequence numbers).
+	partFrag        *mp4ff.Fragment
+	partIndex       uint32
+	fragSeq         uint32
+	partSampleCount int
+	partDurationMs  uint64
+	partIndependent bool
+	// OnPart is invoked once per completed CMAF part when partTargetMs > 0.
+	OnPart func(part LivePart) error
 }
 
 // LiveSegment is one independently-decodable CMAF media segment.
@@ -97,6 +119,24 @@ type LiveSegment struct {
 	Data []byte
 }
 
+// LivePart is one CMAF partial segment (chunk) of a media segment, emitted in
+// low-latency mode the instant it closes - before the whole segment is done - so
+// the playlist can advertise it via #EXT-X-PART for near-live playback.
+type LivePart struct {
+	// SegmentSeq is the parent media segment's sequence number (the N in
+	// seg-N.K.m4s); PartIndex is K within that segment (0-based).
+	SegmentSeq uint32
+	PartIndex  uint32
+	// Independent is true when the part begins with a keyframe (its first sample is
+	// an IDR), i.e. it is independently decodable (#EXT-X-PART INDEPENDENT=YES).
+	Independent bool
+	// DurationMs is the summed sample duration of the part (for #EXT-X-PART).
+	DurationMs uint64
+	// Data of part 0 is styp+moof+mdat; later parts are bare moof+mdat, so
+	// concatenating a segment's parts in order yields one valid CMAF segment.
+	Data []byte
+}
+
 // Sample-entry flags matching the recording muxer so live and archived fragments
 // describe random access points identically.
 //
@@ -137,6 +177,16 @@ func (ls *LiveSegmenter) SetDimensions(width, height uint16) {
 	ls.height = height
 }
 
+// EnableLowLatency switches the segmenter into LL-HLS mode, additionally slicing
+// each segment into ~partTargetMs CMAF parts emitted via OnPart as they close.
+// partTargetMs is clamped to a sane floor. Call before the first WriteSample.
+func (ls *LiveSegmenter) EnableLowLatency(partTargetMs uint64) {
+	if partTargetMs < 100 {
+		partTargetMs = 100
+	}
+	ls.partTargetMs = partTargetMs
+}
+
 // InitSegment returns the encoded init segment bytes, building them on demand.
 // Useful for tests and for serving the #EXT-X-MAP target without waiting for the
 // first media segment.
@@ -238,6 +288,12 @@ func (ls *LiveSegmenter) WriteSample(isKeyframe bool, annexB []byte, ptsMs uint6
 		return fmt.Errorf("livehls: convert AnnexB: %w", err)
 	}
 
+	// Low-latency mode slices each segment into parts; the classic per-segment path
+	// below is left exactly as-is for the default (non-LL) configuration.
+	if ls.partTargetMs > 0 {
+		return ls.writeSampleLL(isKeyframe, lengthPrefixed, ptsMs, compositionOffsetMs)
+	}
+
 	// The previous sample's duration is the gap to this sample's PTS. Commit it
 	// to the (still open) current fragment before we consider rolling segments,
 	// because the pending sample always precedes this one in decode order.
@@ -350,9 +406,23 @@ func (ls *LiveSegmenter) emitSegment() error {
 	return nil
 }
 
-// Close flushes the final pending sample and emits the last open segment. Call
-// once when the live session ends so no trailing media is lost.
+// Close flushes the final pending sample and emits the last open segment (or, in
+// low-latency mode, the last open part). Call once when the live session ends so
+// no trailing media is lost.
 func (ls *LiveSegmenter) Close() error {
+	if ls.partTargetMs > 0 {
+		if ls.pending != nil {
+			dur := ls.lastDurationMs
+			if dur == 0 {
+				dur = liveFallbackDurationMs
+			}
+			ls.pending.Sample.Dur = uint32(dur)
+			if err := ls.commitPendingPart(); err != nil {
+				return err
+			}
+		}
+		return ls.closePart()
+	}
 	if ls.pending != nil {
 		dur := ls.lastDurationMs
 		if dur == 0 {
@@ -365,3 +435,152 @@ func (ls *LiveSegmenter) Close() error {
 	}
 	return ls.emitSegment()
 }
+
+// writeSampleLL is the low-latency counterpart of the per-segment staging in
+// WriteSample: it commits the previous sample into the open part, rolls the part
+// (every ~partTargetMs) and the segment (at keyframes, every ~targetSegmentMs),
+// then stages the current sample. Parts are emitted via OnPart as they close.
+func (ls *LiveSegmenter) writeSampleLL(isKeyframe bool, lengthPrefixed []byte, ptsMs uint64, compositionOffsetMs int32) error {
+	if ls.pending != nil {
+		dur := ls.lastDurationMs
+		if ptsMs > ls.pending.DecodeTime {
+			dur = ptsMs - ls.pending.DecodeTime
+		}
+		if dur == 0 {
+			dur = liveFallbackDurationMs
+		}
+		ls.lastDurationMs = dur
+		ls.pending.Sample.Dur = uint32(dur)
+		if err := ls.commitPendingPart(); err != nil {
+			return err
+		}
+	}
+
+	// Roll the segment at keyframes once enough media accumulated; otherwise roll a
+	// part once it reaches the part target. The two are mutually exclusive: a
+	// keyframe cut also closes the current part.
+	cut := false
+	if isKeyframe {
+		cut = !ls.started || (ptsMs-ls.segStartPTS) >= ls.targetSegmentMs
+	}
+	switch {
+	case cut:
+		if ls.started {
+			if err := ls.closePart(); err != nil {
+				return err
+			}
+		}
+		ls.openSegmentLL(ptsMs)
+	case ls.started && ls.partDurationMs >= ls.partTargetMs:
+		if err := ls.closePart(); err != nil {
+			return err
+		}
+		ls.openPartLL()
+	}
+
+	flags := liveNonSyncSampleFlags
+	if isKeyframe {
+		flags = liveSyncSampleFlags
+	}
+	ls.pending = &mp4ff.FullSample{
+		Sample: mp4ff.Sample{
+			Flags:                 flags,
+			Size:                  uint32(len(lengthPrefixed)),
+			CompositionTimeOffset: compositionOffsetMs,
+		},
+		DecodeTime: ptsMs,
+		Data:       lengthPrefixed,
+	}
+	return nil
+}
+
+// commitPendingPart appends the staged sample to the open part fragment, marking
+// the part independent when its first sample is a keyframe.
+func (ls *LiveSegmenter) commitPendingPart() error {
+	if ls.pending == nil {
+		return nil
+	}
+	if ls.partFrag == nil {
+		// No open part yet (pending staged before the first keyframe cut). The cut
+		// path always opens a part before staging, so this only guards against logic
+		// drift; drop rather than panic.
+		ls.pending = nil
+		return nil
+	}
+	first := ls.partSampleCount == 0
+	if err := ls.partFrag.AddFullSampleToTrack(*ls.pending, ls.videoTrackID); err != nil {
+		return fmt.Errorf("livehls: AddFullSampleToTrack: %w", err)
+	}
+	if first && ls.pending.Sample.Flags == liveSyncSampleFlags {
+		ls.partIndependent = true
+	}
+	ls.partSampleCount++
+	ls.partDurationMs += uint64(ls.pending.Sample.Dur)
+	ls.segDurationMs += uint64(ls.pending.Sample.Dur)
+	ls.pending = nil
+	return nil
+}
+
+// openSegmentLL starts a fresh media segment at a keyframe by opening its part 0.
+func (ls *LiveSegmenter) openSegmentLL(startPTS uint64) {
+	ls.seqNr++
+	ls.partIndex = 0
+	ls.segStartPTS = startPTS
+	ls.segDurationMs = 0
+	ls.started = true
+	ls.openPartFragment()
+}
+
+// openPartLL starts the next part within the current segment.
+func (ls *LiveSegmenter) openPartLL() {
+	ls.partIndex++
+	ls.openPartFragment()
+}
+
+// openPartFragment allocates a fresh single-track fragment (one moof+mdat) for
+// the next part, with a globally monotonic moof sequence number.
+func (ls *LiveSegmenter) openPartFragment() {
+	ls.fragSeq++
+	frag, err := mp4ff.CreateFragment(ls.fragSeq, ls.videoTrackID)
+	if err != nil {
+		log.Log.Error("LiveSegmenter.openPartFragment(): CreateFragment failed: " + err.Error())
+		return
+	}
+	ls.partFrag = frag
+	ls.partSampleCount = 0
+	ls.partDurationMs = 0
+	ls.partIndependent = false
+}
+
+// closePart encodes the open part and hands it to OnPart. Part 0 of a segment
+// carries the CMAF styp; later parts are bare moof+mdat, so a segment's parts
+// concatenate into one valid segment. Empty parts are skipped.
+func (ls *LiveSegmenter) closePart() error {
+	if ls.partFrag == nil || ls.partSampleCount == 0 {
+		return nil
+	}
+	var buf bytes.Buffer
+	if ls.partIndex == 0 {
+		seg := mp4ff.NewMediaSegment() // includes a CMAF styp box by default
+		seg.AddFragment(ls.partFrag)
+		if err := seg.Encode(&buf); err != nil {
+			return fmt.Errorf("livehls: encode part %d.%d: %w", ls.seqNr, ls.partIndex, err)
+		}
+	} else {
+		if err := ls.partFrag.Encode(&buf); err != nil {
+			return fmt.Errorf("livehls: encode part %d.%d: %w", ls.seqNr, ls.partIndex, err)
+		}
+	}
+	out := LivePart{
+		SegmentSeq:  ls.seqNr,
+		PartIndex:   ls.partIndex,
+		Independent: ls.partIndependent,
+		DurationMs:  ls.partDurationMs,
+		Data:        buf.Bytes(),
+	}
+	ls.partFrag = nil
+	if ls.OnPart != nil {
+		return ls.OnPart(out)
+	}
+	return nil
+}

machinery/src/video/livehls_test.go

@@ -369,3 +369,186 @@ func TestLiveSegmenterWritesHLSBundle(t *testing.T) {
 	t.Logf("wrote HLS bundle to %s (%d segments)\n%s", outDir, len(segments), playlist)
 }
 
+// boxTypeAt returns the 4CC box type at the front of a top-level box blob (the
+// 4 bytes following the 32-bit size), or "" if the blob is too short.
+func boxTypeAt(b []byte) string {
+	if len(b) < 8 {
+		return ""
+	}
+	return string(b[4:8])
+}
+
+// TestLiveSegmenterLowLatencyParts runs the segmenter in LL-HLS mode over the
+// same synthetic stream and asserts that:
+//   - each ~2s segment is sliced into multiple CMAF parts (more parts than
+//     segments overall);
+//   - part 0 of every segment carries the CMAF styp and is INDEPENDENT (begins
+//     with the segment keyframe); later parts are bare moof+mdat (no styp);
+//   - moof sequence numbers are globally monotonic across all parts (MSE needs
+//     increasing moof sequence numbers);
+//   - concatenating a segment's parts in order yields exactly the same bytes the
+//     classic per-segment path would emit, decoding into one independent CMAF
+//     segment whose first sample is a sync sample with the expected tfdt;
+//   - every sample and keyframe of the input is preserved end to end.
+func TestLiveSegmenterLowLatencyParts(t *testing.T) {
+	const (
+		frameDurMs = uint64(40) // 25 fps
+		gopFrames  = 25         // keyframe every 1000 ms
+		numGOPs    = 6
+		numFrames  = gopFrames * numGOPs // 150 frames, 6000 ms
+		targetMs   = uint64(2000)        // 2s segments => 2 GOPs each
+		partMs     = uint64(300)         // ~300 ms parts => ~6-7 parts/segment
+	)
+
+	seg := NewLiveSegmenter("H264", [][]byte{liveTestSPS}, [][]byte{liveTestPPS}, nil, targetMs)
+	seg.SetDimensions(640, 480)
+	seg.EnableLowLatency(partMs)
+
+	var initBytes []byte
+	var initCalls int
+	var parts []LivePart
+	seg.OnInit = func(b []byte) error {
+		initCalls++
+		initBytes = append([]byte(nil), b...)
+		return nil
+	}
+	seg.OnPart = func(p LivePart) error {
+		parts = append(parts, p)
+		return nil
+	}
+
+	for i := 0; i < numFrames; i++ {
+		isKey := i%gopFrames == 0
+		if err := seg.WriteSample(isKey, makeAnnexBFrame(isKey), uint64(i)*frameDurMs, 0); err != nil {
+			t.Fatalf("WriteSample(frame=%d): %v", i, err)
+		}
+	}
+	if err := seg.Close(); err != nil {
+		t.Fatalf("Close: %v", err)
+	}
+
+	if initCalls != 1 {
+		t.Fatalf("OnInit called %d times, want 1", initCalls)
+	}
+	if len(parts) == 0 {
+		t.Fatal("no parts produced in low-latency mode")
+	}
+
+	// --- Parts are globally moof-monotonic, and group into 3 segments whose part
+	// indices are contiguous from 0. ---
+	bySeg := map[uint32][]LivePart{}
+	var order []uint32
+	var lastMoof uint32
+	for i, p := range parts {
+		if _, seen := bySeg[p.SegmentSeq]; !seen {
+			order = append(order, p.SegmentSeq)
+		}
+		bySeg[p.SegmentSeq] = append(bySeg[p.SegmentSeq], p)
+
+		// Decode the part to read its moof sequence number and confirm the styp
+		// convention (part 0 => styp present, later parts => bare moof+mdat).
+		front := boxTypeAt(p.Data)
+		if p.PartIndex == 0 {
+			if front != "styp" {
+				t.Errorf("seg %d part 0: leading box=%q, want styp", p.SegmentSeq, front)
+			}
+			if !p.Independent {
+				t.Errorf("seg %d part 0: Independent=false, want true (starts on keyframe)", p.SegmentSeq)
+			}
+		} else if front != "moof" {
+			t.Errorf("seg %d part %d: leading box=%q, want moof (no styp on later parts)", p.SegmentSeq, p.PartIndex, front)
+		}
+
+		parsed, err := mp4ff.DecodeFile(bytes.NewReader(p.Data))
+		if err != nil {
+			t.Fatalf("seg %d part %d: decode: %v", p.SegmentSeq, p.PartIndex, err)
+		}
+		if len(parsed.Segments) != 1 || len(parsed.Segments[0].Fragments) != 1 {
+			t.Fatalf("seg %d part %d: want exactly one fragment", p.SegmentSeq, p.PartIndex)
+		}
+		moof := parsed.Segments[0].Fragments[0].Moof.Mfhd.SequenceNumber
+		if i > 0 && moof <= lastMoof {
+			t.Errorf("part %d: moof sequence=%d not greater than previous %d", i, moof, lastMoof)
+		}
+		lastMoof = moof
+	}
+
+	if len(order) != 3 {
+		t.Fatalf("got %d segments, want 3", len(order))
+	}
+	if len(parts) <= len(order) {
+		t.Fatalf("got %d parts for %d segments, expected each segment to be sliced into multiple parts", len(parts), len(order))
+	}
+	for _, segSeq := range order {
+		for idx, p := range bySeg[segSeq] {
+			if p.PartIndex != uint32(idx) {
+				t.Errorf("seg %d: part index %d out of order (want %d)", segSeq, p.PartIndex, idx)
+			}
+		}
+	}
+
+	// --- Concatenating a segment's parts must reconstruct one independent CMAF
+	// segment that decodes against the init segment. ---
+	wantTFDT := map[uint32]uint64{1: 0, 2: 2000, 3: 4000}
+	var totalSamples, totalSync int
+	for _, segSeq := range order {
+		segParts := bySeg[segSeq]
+		var full []byte
+		var wantPartDur uint64
+		for _, p := range segParts {
+			full = append(full, p.Data...)
+			wantPartDur += p.DurationMs
+		}
+		standalone := append(append([]byte(nil), initBytes...), full...)
+		parsed, err := mp4ff.DecodeFile(bytes.NewReader(standalone))
+		if err != nil {
+			t.Fatalf("seg %d: decode concatenated parts: %v", segSeq, err)
+		}
+		if len(parsed.Segments) != 1 {
+			t.Fatalf("seg %d: parsed %d media segments, want 1", segSeq, len(parsed.Segments))
+		}
+		mseg := parsed.Segments[0]
+		if mseg.Styp == nil {
+			t.Errorf("seg %d: reconstructed segment missing CMAF styp", segSeq)
+		}
+		if len(mseg.Fragments) != len(segParts) {
+			t.Errorf("seg %d: %d fragments, want %d (one per part)", segSeq, len(mseg.Fragments), len(segParts))
+		}
+		firstTraf := mseg.Fragments[0].Moof.Traf
+		if got := firstTraf.Tfdt.BaseMediaDecodeTime(); got != wantTFDT[segSeq] {
+			t.Errorf("seg %d: first fragment tfdt=%d, want %d", segSeq, got, wantTFDT[segSeq])
+		}
+		var segDur uint64
+		var firstSample mp4ff.Sample
+		var haveFirst bool
+		for _, fr := range mseg.Fragments {
+			for _, trun := range fr.Moof.Traf.Truns {
+				for _, smp := range trun.Samples {
+					if !haveFirst {
+						firstSample = smp
+						haveFirst = true
+					}
+					totalSamples++
+					if isSyncSample(smp) {
+						totalSync++
+					}
+					segDur += uint64(smp.Dur)
+				}
+			}
+		}
+		if !isSyncSample(firstSample) {
+			t.Errorf("seg %d: first sample is not a sync sample", segSeq)
+		}
+		if segDur != wantPartDur {
+			t.Errorf("seg %d: summed sample dur=%d, summed part dur=%d", segSeq, segDur, wantPartDur)
+		}
+	}
+
+	if totalSamples != numFrames {
+		t.Errorf("total samples across parts=%d, want %d", totalSamples, numFrames)
+	}
+	if totalSync != numGOPs {
+		t.Errorf("total sync samples=%d, want %d (one per GOP)", totalSync, numGOPs)
+	}
+}
+