recorder: reset when absolute time drifts from stream time (#4778)

Author: aler9

docs/2-usage/14-route-absolute-timestamps.md

@@ -13,7 +13,7 @@ and supports sending absolute timestamps with the following protocols:
 - RTSP
 - WebRTC
 
-By default, absolute timestamps of incoming frames are not used, instead they are replaced with the current timestamp. This prevents users from arbitrarily changing recording dates, and also allows to support sources that do not send absolute timestamps. It is possible to preserve original absolute timestamps by toggling the `useAbsoluteTimestamp` parameter:
+By default, absolute timestamps of incoming frames are not used, instead they are replaced with the system timestamp. This prevents users from arbitrarily changing recording dates, and also allows to support sources that do not send absolute timestamps. It is possible to preserve original absolute timestamps by toggling the `useAbsoluteTimestamp` parameter:
 
 ```yml
 pathDefaults:

go.mod

@@ -13,7 +13,7 @@ require (
 	github.com/bluenviron/gohlslib/v2 v2.2.4
 	github.com/bluenviron/gortmplib v0.1.1
 	github.com/bluenviron/gortsplib/v5 v5.2.0
-	github.com/bluenviron/mediacommon/v2 v2.5.1
+	github.com/bluenviron/mediacommon/v2 v2.5.2-0.20251201152746-8d059e8616fb
 	github.com/datarhei/gosrt v0.9.0
 	github.com/fsnotify/fsnotify v1.9.0
 	github.com/gin-contrib/pprof v1.5.3

go.sum

@@ -39,8 +39,8 @@ github.com/bluenviron/gortmplib v0.1.1 h1:pmR6qfPcJJmE17lWQ/bpuBFZtgGnMrN8KdFj1G
 github.com/bluenviron/gortmplib v0.1.1/go.mod h1:XWy2YzbTP1XEEZ8232OG7I1MSwubsbDRKDNhXGgS2kg=
 github.com/bluenviron/gortsplib/v5 v5.2.0 h1:yk0H9Z1Z+H41/x5hDt84rKm6+MNA483NsRXPYe+or/A=
 github.com/bluenviron/gortsplib/v5 v5.2.0/go.mod h1:UYCbHEb0T49kBDgIlTJaZOchD2f5g1JigFmmxQfW7vY=
-github.com/bluenviron/mediacommon/v2 v2.5.1 h1:qB2fb5c0xyl5OB2gfSfulpEJn7Cdm3vI2n8wjiLMxKI=
-github.com/bluenviron/mediacommon/v2 v2.5.1/go.mod h1:zy1fODPuS/kBd93ftgJS1Jhvjq7LFWfAo32KP7By9AE=
+github.com/bluenviron/mediacommon/v2 v2.5.2-0.20251201152746-8d059e8616fb h1:42lRaSsrPXvwB9kLGIujU9yONrSPPp0j4Ohwg6zp/yw=
+github.com/bluenviron/mediacommon/v2 v2.5.2-0.20251201152746-8d059e8616fb/go.mod h1:5V15TiOfeaNVmZPVuOqAwqQSWyvMV86/dijDKu5q9Zs=
 github.com/bytedance/sonic v1.14.0 h1:/OfKt8HFw0kh2rj8N0F6C/qPGRESq0BbaNZgcNXXzQQ=
 github.com/bytedance/sonic v1.14.0/go.mod h1:WoEbx8WTcFJfzCe0hbmyTGrfjt8PzNEBdxlNUO24NhA=
 github.com/bytedance/sonic/loader v0.3.0 h1:dskwH8edlzNMctoruo8FPTJDF3vLtDT0sXZwvZJyqeA=

internal/recorder/format_fmp4.go

@@ -98,6 +98,12 @@ func jpegExtractSize(image []byte) (int, int, error) {
 	}
 }
 
+type formatFMP4Sample struct {
+	*fmp4.Sample
+	dts int64
+	ntp time.Time
+}
+
 type formatFMP4 struct {
 	ri *recorderInstance
 
@@ -111,18 +117,15 @@ func (f *formatFMP4) initialize() bool {
 	nextID := 1
 
 	addTrack := func(format rtspformat.Format, codec mp4.Codec) *formatFMP4Track {
-		initTrack := &fmp4.InitTrack{
-			TimeScale: uint32(format.ClockRate()),
-			Codec:     codec,
-		}
-		initTrack.ID = nextID
-		nextID++
-
 		track := &formatFMP4Track{
 			f:         f,
-			initTrack: initTrack,
+			id:        nextID,
+			clockRate: uint32(format.ClockRate()),
+			codec:     codec,
 		}
+		track.initialize()
 
+		nextID++
 		f.tracks = append(f.tracks, track)
 		return track
 	}
@@ -180,7 +183,7 @@ func (f *formatFMP4) initialize() bool {
 							return err
 						}
 
-						return track.write(&sample{
+						return track.write(&formatFMP4Sample{
 							Sample: &sampl,
 							dts:    u.PTS,
 							ntp:    u.NTP,
@@ -257,7 +260,7 @@ func (f *formatFMP4) initialize() bool {
 							firstReceived = true
 						}
 
-						return track.write(&sample{
+						return track.write(&formatFMP4Sample{
 							Sample: &fmp4.Sample{
 								IsNonSyncSample: !randomAccess,
 								Payload:         u.Payload.(unit.PayloadVP9),
@@ -348,7 +351,7 @@ func (f *formatFMP4) initialize() bool {
 							return err
 						}
 
-						return track.write(&sample{
+						return track.write(&formatFMP4Sample{
 							Sample: &sampl,
 							dts:    dts,
 							ntp:    u.NTP,
@@ -424,7 +427,7 @@ func (f *formatFMP4) initialize() bool {
 							return err
 						}
 
-						return track.write(&sample{
+						return track.write(&formatFMP4Sample{
 							Sample: &sampl,
 							dts:    dts,
 							ntp:    u.NTP,
@@ -481,7 +484,7 @@ func (f *formatFMP4) initialize() bool {
 						}
 						lastPTS = u.PTS
 
-						return track.write(&sample{
+						return track.write(&formatFMP4Sample{
 							Sample: &fmp4.Sample{
 								Payload:         u.Payload.(unit.PayloadMPEG4Video),
 								IsNonSyncSample: !randomAccess,
@@ -532,7 +535,7 @@ func (f *formatFMP4) initialize() bool {
 						}
 						lastPTS = u.PTS
 
-						return track.write(&sample{
+						return track.write(&formatFMP4Sample{
 							Sample: &fmp4.Sample{
 								Payload:         u.Payload.(unit.PayloadMPEG1Video),
 								IsNonSyncSample: !randomAccess,
@@ -570,7 +573,7 @@ func (f *formatFMP4) initialize() bool {
 							f.updateCodecParams()
 						}
 
-						return track.write(&sample{
+						return track.write(&formatFMP4Sample{
 							Sample: &fmp4.Sample{
 								Payload: u.Payload.(unit.PayloadMJPEG),
 							},
@@ -596,7 +599,7 @@ func (f *formatFMP4) initialize() bool {
 						pts := u.PTS
 
 						for _, packet := range u.Payload.(unit.PayloadOpus) {
-							err := track.write(&sample{
+							err := track.write(&formatFMP4Sample{
 								Sample: &fmp4.Sample{
 									Payload: packet,
 								},
@@ -630,7 +633,7 @@ func (f *formatFMP4) initialize() bool {
 						for i, au := range u.Payload.(unit.PayloadMPEG4Audio) {
 							pts := u.PTS + int64(i)*mpeg4audio.SamplesPerAccessUnit
 
-							err := track.write(&sample{
+							err := track.write(&formatFMP4Sample{
 								Sample: &fmp4.Sample{
 									Payload: au,
 								},
@@ -667,7 +670,7 @@ func (f *formatFMP4) initialize() bool {
 								return err
 							}
 
-							return track.write(&sample{
+							return track.write(&formatFMP4Sample{
 								Sample: &fmp4.Sample{
 									Payload: ame.Payloads[0][0][0],
 								},
@@ -710,7 +713,7 @@ func (f *formatFMP4) initialize() bool {
 								f.updateCodecParams()
 							}
 
-							err = track.write(&sample{
+							err = track.write(&formatFMP4Sample{
 								Sample: &fmp4.Sample{
 									Payload: frame,
 								},
@@ -779,7 +782,7 @@ func (f *formatFMP4) initialize() bool {
 
 							pts := u.PTS + int64(i)*ac3.SamplesPerFrame
 
-							err = track.write(&sample{
+							err = track.write(&formatFMP4Sample{
 								Sample: &fmp4.Sample{
 									Payload: frame,
 								},
@@ -825,7 +828,7 @@ func (f *formatFMP4) initialize() bool {
 							lpcm = al
 						}
 
-						return track.write(&sample{
+						return track.write(&formatFMP4Sample{
 							Sample: &fmp4.Sample{
 								Payload: lpcm,
 							},
@@ -851,7 +854,7 @@ func (f *formatFMP4) initialize() bool {
 							return nil
 						}
 
-						return track.write(&sample{
+						return track.write(&formatFMP4Sample{
 							Sample: &fmp4.Sample{
 								Payload: u.Payload.(unit.PayloadLPCM),
 							},

internal/recorder/format_fmp4_part.go

@@ -56,7 +56,7 @@ func (p *formatFMP4Part) close(w io.Writer) error {
 	return writePart(w, p.number, p.partTracks)
 }
 
-func (p *formatFMP4Part) write(track *formatFMP4Track, sample *sample, dts time.Duration) error {
+func (p *formatFMP4Part) write(track *formatFMP4Track, sample *formatFMP4Sample, dts time.Duration) error {
 	size := uint64(len(sample.Payload))
 	if (p.size + size) > uint64(p.maxPartSize) {
 		return fmt.Errorf("reached maximum part size")

internal/recorder/format_fmp4_segment.go

@@ -201,7 +201,7 @@ func (s *formatFMP4Segment) closeCurPart() error {
 	return s.curPart.close(s.fi)
 }
 
-func (s *formatFMP4Segment) write(track *formatFMP4Track, sample *sample, dts time.Duration) error {
+func (s *formatFMP4Segment) write(track *formatFMP4Track, sample *formatFMP4Sample, dts time.Duration) error {
 	endDTS := dts + timestampToDuration(int64(sample.Duration), int(track.initTrack.TimeScale))
 	if endDTS > s.endDTS {
 		s.endDTS = endDTS

internal/recorder/format_fmp4_track.go

@@ -1,9 +1,11 @@
 package recorder
 
 import (
+	"fmt"
 	"time"
 
 	"github.com/bluenviron/mediacommon/v2/pkg/formats/fmp4"
+	"github.com/bluenviron/mediacommon/v2/pkg/formats/mp4"
 	"github.com/bluenviron/mediamtx/internal/logger"
 )
 
@@ -43,12 +45,26 @@ func nextSegmentStartingPos(tracks []*formatFMP4Track) (time.Time, time.Duration
 
 type formatFMP4Track struct {
 	f         *formatFMP4
-	initTrack *fmp4.InitTrack
+	id        int
+	clockRate uint32
+	codec     mp4.Codec
+
+	initTrack        *fmp4.InitTrack
+	nextSample       *formatFMP4Sample
+	startInitialized bool
+	startDTS         time.Duration
+	startNTP         time.Time
+}
 
-	nextSample *sample
+func (t *formatFMP4Track) initialize() {
+	t.initTrack = &fmp4.InitTrack{
+		ID:        t.id,
+		TimeScale: t.clockRate,
+		Codec:     t.codec,
+	}
 }
 
-func (t *formatFMP4Track) write(sample *sample) error {
+func (t *formatFMP4Track) write(sample *formatFMP4Sample) error {
 	// wait the first video sample before setting hasVideo
 	if t.initTrack.Codec.IsVideo() {
 		t.f.hasVideo = true
@@ -69,6 +85,17 @@ func (t *formatFMP4Track) write(sample *sample) error {
 
 	dts := timestampToDuration(sample.dts, int(t.initTrack.TimeScale))
 
+	if !t.startInitialized {
+		t.startDTS = dts
+		t.startNTP = sample.ntp
+		t.startInitialized = true
+	} else {
+		drift := sample.ntp.Sub(t.startNTP) - (dts - t.startDTS)
+		if drift < -ntpDriftTolerance || drift > ntpDriftTolerance {
+			return fmt.Errorf("detected drift between recording duration and absolute time, resetting")
+		}
+	}
+
 	if t.f.currentSegment == nil {
 		t.f.currentSegment = &formatFMP4Segment{
 			f:        t.f,