diff --git a/pkg/spxrpc/place_stream_playback_getvideo.go b/pkg/spxrpc/place_stream_playback_getvideo.go index cafa8db8..ccd5e7c1 100644 --- a/pkg/spxrpc/place_stream_playback_getvideo.go +++ b/pkg/spxrpc/place_stream_playback_getvideo.go @@ -636,7 +636,7 @@ func mediaPlaylist(meta *vod.Metafile, trackID, ownerDID, sid, cdnURL string, st if !ok { return "", echo.NewHTTPError(http.StatusNotFound, "TrackNotFound") } - segments := filterSegments(t.Segments, t.Timescale, startMS, endMS) + segments, discontinuitySeq := filterSegments(t.Segments, t.Timescale, startMS, endMS) var maxDurSec float64 for _, s := range segments { @@ -660,11 +660,26 @@ func mediaPlaylist(meta *vod.Metafile, trackID, ownerDID, sid, cdnURL string, st "#EXT-X-INDEPENDENT-SEGMENTS", fmt.Sprintf("#EXT-X-TARGETDURATION:%d", targetDuration), "#EXT-X-MEDIA-SEQUENCE:0", + } + // EXT-X-DISCONTINUITY-SEQUENCE accounts for discontinuities trimmed off the + // front of a clipped playlist (the first served segment being a boundary, or + // boundaries before it). Omitted when zero, which is the common case. + if discontinuitySeq > 0 { + lines = append(lines, fmt.Sprintf("#EXT-X-DISCONTINUITY-SEQUENCE:%d", discontinuitySeq)) + } + lines = append(lines, fmt.Sprintf(`#EXT-X-MAP:URI=%q`, blobURL(cdnURL, ownerDID, t.InitCID, sid)), "", - } + ) bURL := blobURL(cdnURL, ownerDID, t.BlobCID, sid) - for _, seg := range segments { + for i, seg := range segments { + // A segment whose decode time reset (a concatenated reconnect/restart) + // begins a new timeline; signal it so players re-anchor instead of + // choking on the backward jump. The first served segment's own boundary + // is folded into EXT-X-DISCONTINUITY-SEQUENCE above, not an inline tag. + if seg.Discontinuity && i > 0 { + lines = append(lines, "#EXT-X-DISCONTINUITY") + } durSec := float64(seg.DurationTicks) / float64(t.Timescale) lines = append(lines, fmt.Sprintf("#EXTINF:%.6f,", durSec), @@ -719,9 +734,13 @@ func composeClipBounds(clipStartMS int64, clipEndMS, queryStartMS, queryEndMS *i // no sub-segment splitting. Bounds are in the parent video's // timeline; clip-record local times must be composed by the caller // before they get here. -func filterSegments(segments []vod.MetafileSegment, timescale uint32, startMS, endMS *int64) []vod.MetafileSegment { +// It also returns the discontinuity sequence: the number of discontinuity +// boundaries at or before the first returned segment (i.e. trimmed off the +// front), which the caller emits as EXT-X-DISCONTINUITY-SEQUENCE so a clipped +// playlist's timeline stays correct. +func filterSegments(segments []vod.MetafileSegment, timescale uint32, startMS, endMS *int64) ([]vod.MetafileSegment, int) { if startMS == nil && endMS == nil { - return segments + return segments, 0 } tsf := float64(timescale) startTicks := int64(0) @@ -735,14 +754,24 @@ func filterSegments(segments []vod.MetafileSegment, timescale uint32, startMS, e out := segments[:0:0] cursor := int64(0) + discontinuitySeq := 0 + started := false for _, seg := range segments { segEnd := cursor + int64(seg.DurationTicks) if segEnd > startTicks && cursor < endTicks { + // The first served segment being itself a boundary is reflected in + // the discontinuity sequence rather than an inline tag. + if !started && seg.Discontinuity { + discontinuitySeq++ + } + started = true out = append(out, seg) + } else if !started && seg.Discontinuity { + discontinuitySeq++ } cursor = segEnd } - return out + return out, discontinuitySeq } // computeBandwidth derives an approximate average bitrate (bits/s) diff --git a/pkg/spxrpc/place_stream_playback_getvideo_test.go b/pkg/spxrpc/place_stream_playback_getvideo_test.go index be7dfca6..4559bf91 100644 --- a/pkg/spxrpc/place_stream_playback_getvideo_test.go +++ b/pkg/spxrpc/place_stream_playback_getvideo_test.go @@ -124,6 +124,63 @@ func TestMediaPlaylist_Video(t *testing.T) { require.Contains(t, pl, `#EXT-X-BYTERANGE:1800@2100`) require.Contains(t, pl, "#EXTINF:1.000000,") require.Contains(t, pl, "#EXT-X-ENDLIST") + // A clean single-session VOD has no discontinuities. + require.NotContains(t, pl, "#EXT-X-DISCONTINUITY") +} + +// reconnectVideoMetafile is a 3-segment video track whose middle segment is +// flagged as a discontinuity — the metafile shape produced by a recording that +// concatenates two ingest sessions (a reconnect), where session 2's decode +// time reset. Each segment is 1s (durationTicks == timescale). +func reconnectVideoMetafile() *vod.Metafile { + return &vod.Metafile{ + BlobCID: "bafyblob", + BlobSize: 10_000, + Tracks: map[string]vod.MetafileTrack{ + "1": { + Type: "video", + Codec: "avc1.64002a", + Timescale: 6000, + InitCID: "bafyvideoinit", + BlobCID: "bafyblob", + BlobSize: 10_000, + Width: 1920, + Height: 1080, + Segments: []vod.MetafileSegment{ + {Offset: 100, Size: 2000, DurationTicks: 6000, SampleCount: 60}, + {Offset: 2100, Size: 1800, DurationTicks: 6000, SampleCount: 60, Discontinuity: true}, + {Offset: 3900, Size: 1700, DurationTicks: 6000, SampleCount: 60}, + }, + }, + }, + } +} + +func TestMediaPlaylist_Discontinuity(t *testing.T) { + pl, err := mediaPlaylist(reconnectVideoMetafile(), "1", fixtureDID, fixtureSID, "", nil, nil) + require.NoError(t, err) + // Exactly one inline EXT-X-DISCONTINUITY (its own line — distinct from the + // EXT-X-DISCONTINUITY-SEQUENCE header), and no sequence header for full play. + require.Equal(t, 1, strings.Count(pl, "#EXT-X-DISCONTINUITY\n")) + require.NotContains(t, pl, "#EXT-X-DISCONTINUITY-SEQUENCE") + // It must sit after the first segment and immediately before the flagged one. + discIdx := strings.Index(pl, "#EXT-X-DISCONTINUITY\n") + firstSeg := strings.Index(pl, "#EXT-X-BYTERANGE:2000@100") + flaggedSeg := strings.Index(pl, "#EXT-X-BYTERANGE:1800@2100") + require.Greater(t, discIdx, firstSeg, "discontinuity must come after the first segment") + require.Less(t, discIdx, flaggedSeg, "discontinuity must come right before the flagged segment") +} + +func TestMediaPlaylist_DiscontinuityTrimmedToBoundary(t *testing.T) { + start := int64(1000) // ms — segment index 1 (the boundary) starts at 1s + pl, err := mediaPlaylist(reconnectVideoMetafile(), "1", fixtureDID, fixtureSID, "", &start, nil) + require.NoError(t, err) + // The boundary is now the first served segment: reflected as a sequence + // bump, NOT an inline tag. + require.Contains(t, pl, "#EXT-X-DISCONTINUITY-SEQUENCE:1") + require.Equal(t, 0, strings.Count(pl, "#EXT-X-DISCONTINUITY\n")) + require.NotContains(t, pl, "#EXT-X-BYTERANGE:2000@100", "pre-boundary segment should be trimmed") + require.Contains(t, pl, "#EXT-X-BYTERANGE:1800@2100", "boundary segment should be served") } func TestMediaPlaylist_UnknownTrack(t *testing.T) { diff --git a/pkg/vod/discontinuity_test.go b/pkg/vod/discontinuity_test.go new file mode 100644 index 00000000..03c4b541 --- /dev/null +++ b/pkg/vod/discontinuity_test.go @@ -0,0 +1,85 @@ +package vod + +import ( + "bytes" + "context" + "os" + "testing" + "time" + + "github.com/stretchr/testify/require" + "stream.place/streamplace/pkg/bdasl" + "stream.place/streamplace/pkg/blob" + "stream.place/streamplace/pkg/log" +) + +// TestMetafileFlagsReconnectDiscontinuity proves the metafile builder flags a +// concatenated reconnect. It synthesizes a "reconnect" recording by processing +// the fixture twice and concatenating session 2's bare segments after session +// 1's full [init][segments]; session 2's segments restart their tfdt at ~0, +// exactly like a streamer who disconnected/reconnected. Regenerating the +// sidecars from that blob must flag exactly one discontinuity per track — the +// first segment of session 2 — and never the very first segment. +// +// Needs gstreamer (warmGST/streamThroughMuxl), so it runs in the cgo test +// containers, not a bare checkout. +func TestMetafileFlagsReconnectDiscontinuity(t *testing.T) { + warmGST() + + ctx, cancel := context.WithCancel(context.Background()) + defer cancel() + ctx = log.WithLogValues(ctx, "test", "TestMetafileFlagsReconnectDiscontinuity") + + fixture, err := os.ReadFile(getFixture("5sec.mp4")) + require.NoError(t, err) + signer, err := newUploadSigner(time.Now()) + require.NoError(t, err) + + buildSession := func() (b []byte, initLen int64) { + out := &bytes.Buffer{} + hasher := bdasl.NewWriter() + dst := teeWriter{hasher, out} + st, err := blob.NewFileStore(t.TempDir()) + require.NoError(t, err) + mb := newMetafileBuilder(ctx, st) + _, err = streamThroughMuxl(ctx, bytes.NewReader(fixture), int64(len(fixture)), dst, mb, signer.SignerInput) + require.NoError(t, err) + blob := out.Bytes() + meta := mb.Finalize(hasher.CID(), int64(len(blob))) + return blob, minFirstOffset(t, meta) + } + + s1, _ := buildSession() + s2, s2InitLen := buildSession() + reconnect := append(append([]byte(nil), s1...), s2[s2InitLen:]...) + + // Regenerate the sidecars from the concatenated blob, exactly as finalize + // (and VOD transfer) do. + store, err := blob.NewFileStore(t.TempDir()) + require.NoError(t, err) + h := bdasl.NewWriter() + _, _ = h.Write(reconnect) + cid := h.CID() + w, err := store.NewWriter(ctx, BlobsPrefix+cid+".mp4", "video/mp4") + require.NoError(t, err) + _, err = w.Write(reconnect) + require.NoError(t, err) + require.NoError(t, w.Complete()) + + meta, _, err := regenerateSidecars(ctx, store, cid, int64(len(reconnect))) + require.NoError(t, err) + require.NotEmpty(t, meta.Tracks) + + for tid, tr := range meta.Tracks { + var discIdx []int + for i, s := range tr.Segments { + if s.Discontinuity { + discIdx = append(discIdx, i) + } + } + t.Logf("track %s: %d segments, discontinuities at %v", tid, len(tr.Segments), discIdx) + require.Lenf(t, discIdx, 1, "track %s should have exactly one discontinuity (the reconnect boundary)", tid) + require.Greaterf(t, discIdx[0], 0, "track %s discontinuity must not be the first segment", tid) + require.Falsef(t, tr.Segments[0].Discontinuity, "track %s first segment must not be flagged", tid) + } +} diff --git a/pkg/vod/metafile.go b/pkg/vod/metafile.go index ec9e0a44..1529152c 100644 --- a/pkg/vod/metafile.go +++ b/pkg/vod/metafile.go @@ -2,6 +2,7 @@ package vod import ( "context" + "encoding/binary" "encoding/json" "fmt" "sort" @@ -16,6 +17,44 @@ import ( "stream.place/streamplace/pkg/muxl" ) +// firstTFDT walks the ISO-BMFF boxes of a segment chunk (per-track moof+mdat, +// possibly prefixed by c2pa/muxl uuid boxes) and returns the +// baseMediaDecodeTime from the first tfdt box it finds. Minimal walker: +// recurses into moof/traf containers and skips everything else. ok=false if no +// tfdt is present or a box uses 64-bit/extends-to-EOF sizing (not expected in +// canonical MUXL segments). +func firstTFDT(box []byte) (uint64, bool) { + for len(box) >= 8 { + size := int(binary.BigEndian.Uint32(box[0:4])) + typ := string(box[4:8]) + if size < 8 || size > len(box) { + return 0, false + } + payload := box[8:size] + switch typ { + case "moof", "traf": + if v, ok := firstTFDT(payload); ok { + return v, true + } + case "tfdt": + if len(payload) >= 1 { + switch payload[0] { // version + case 0: + if len(payload) >= 8 { + return uint64(binary.BigEndian.Uint32(payload[4:8])), true + } + case 1: + if len(payload) >= 12 { + return binary.BigEndian.Uint64(payload[4:12]), true + } + } + } + } + box = box[size:] + } + return 0, false +} + // Metafile is the per-blob HLS playback index emitted alongside a // processed VOD blob. JSON shape mirrors what `muxl hls` produces (see // /home/iameli/code/muxl/src/hls.rs:write_metadata_json) — the playback @@ -55,6 +94,15 @@ type MetafileSegment struct { Size int64 `json:"size"` DurationTicks uint64 `json:"durationTicks"` SampleCount uint32 `json:"sampleCount"` + // Discontinuity marks a segment that begins a new continuous timeline — + // its decode time jumped backward relative to the previous segment of the + // same track. This happens when a recording concatenates multiple ingest + // sessions (the streamer disconnected/reconnected or stopped and restarted), + // each restarting its tfdt near zero. The HLS playlist generator emits an + // EXT-X-DISCONTINUITY before such a segment so players re-anchor the + // timeline instead of choking on the backward jump. Omitted (false) for the + // common single-session case. + Discontinuity bool `json:"discontinuity,omitempty"` } // metafileBuilder consumes the rich event stream from the muxl @@ -75,6 +123,12 @@ type metafileBuilder struct { runningOffset int64 // bytes written to the concatenated output so far seenInit bool + + // lastTFDT / tfdtSeen track each track's previous baseMediaDecodeTime so a + // backward jump (a concatenated reconnect/restart) can be flagged as a + // discontinuity. See MetafileSegment.Discontinuity. + lastTFDT map[string]uint64 + tfdtSeen map[string]bool } func newMetafileBuilder(ctx context.Context, store blob.Store) *metafileBuilder { @@ -83,6 +137,8 @@ func newMetafileBuilder(ctx context.Context, store blob.Store) *metafileBuilder store: store, trackInitCIDs: map[string]string{}, trackSegments: map[string][]MetafileSegment{}, + lastTFDT: map[string]uint64{}, + tfdtSeen: map[string]bool{}, } } @@ -127,11 +183,25 @@ func (b *metafileBuilder) Observe(ev *muxl.MuxlEvent) error { sort.Strings(keys) for _, tid := range keys { chunk := ev.Tracks[tid] + // Flag a discontinuity when this track's decode time jumps backward + // vs its previous segment — the signature of a concatenated + // reconnect/restart. A normal stream's tfdt is strictly increasing + // (tfdt[n] = tfdt[n-1] + duration[n-1]), so this never fires for a + // clean single-session recording. + disc := false + if tfdt, ok := firstTFDT(chunk); ok { + if b.tfdtSeen[tid] && tfdt < b.lastTFDT[tid] { + disc = true + } + b.lastTFDT[tid] = tfdt + b.tfdtSeen[tid] = true + } b.trackSegments[tid] = append(b.trackSegments[tid], MetafileSegment{ Offset: b.runningOffset, Size: int64(len(chunk)), DurationTicks: ev.Durations[tid], SampleCount: ev.SampleCounts[tid], + Discontinuity: disc, }) b.runningOffset += int64(len(chunk)) }