diff --git a/pkg/media/segment_buffer.go b/pkg/media/segment_buffer.go index 816668448..25cab2b11 100644 --- a/pkg/media/segment_buffer.go +++ b/pkg/media/segment_buffer.go @@ -20,8 +20,12 @@ const segmentBufferMinAverage = 500 * time.Millisecond type BufferedSegment struct { Start time.Time // media start time Published bool - Data []byte // bare canonical .m4s bytes; never modified once buffered - addedAt time.Time + // Local is set for a segment ingested by this node rather than replicated + // to it. A node sees every segment of a stream it ingests, so a run of + // local segments has no gaps; replication can drop some. + Local bool + Data []byte // bare canonical .m4s bytes; never modified once buffered + addedAt time.Time } // segmentBuffer is a stream's time-bounded run of recent segments. All @@ -32,11 +36,13 @@ type segmentBuffer struct { segs []BufferedSegment } -// bufferSegment appends a canonical segment to the stream's buffer and evicts -// what has aged out. Called for every validated segment — published or not, -// local or replicated — so whatever this node has seen of a stream recently -// can be clipped. The bytes are copied since the caller reuses its buffer. -func (mm *MediaManager) bufferSegment(did string, start time.Time, published bool, segment []byte) { +// bufferSegment appends a canonical segment to the stream's buffer. Called +// for every validated segment — published or not, local or replicated — so +// whatever this node has seen of a stream recently can be clipped. The bytes +// are copied since the caller reuses its buffer. Every buffer is evicted +// here, so a stream that ended is freed once its segments age out even if +// nothing reads its buffer again. +func (mm *MediaManager) bufferSegment(did string, start time.Time, published, local bool, segment []byte) { now := time.Now() mm.segBuffersMut.Lock() defer mm.segBuffersMut.Unlock() @@ -44,10 +50,16 @@ func (mm *MediaManager) bufferSegment(did string, start time.Time, published boo b.segs = append(b.segs, BufferedSegment{ Start: start, Published: published, + Local: local, Data: append([]byte(nil), segment...), addedAt: now, }) - b.evict(now) + for other, ob := range mm.segBuffers { + ob.evict(now) + if len(ob.segs) == 0 { + delete(mm.segBuffers, other) + } + } } // RetainSegments keeps the stream's segments in memory for at least d diff --git a/pkg/media/segment_buffer_test.go b/pkg/media/segment_buffer_test.go index 497114719..9c61924b1 100644 --- a/pkg/media/segment_buffer_test.go +++ b/pkg/media/segment_buffer_test.go @@ -15,7 +15,7 @@ import ( func TestSegmentBufferNilMap(t *testing.T) { mm := &MediaManager{} // segBuffers deliberately nil require.NotPanics(t, func() { - mm.bufferSegment("did:x", time.Unix(0, 0), true, []byte("seg")) + mm.bufferSegment("did:x", time.Unix(0, 0), true, true, []byte("seg")) }) require.Len(t, mm.BufferedSegments("did:x"), 1) } @@ -26,7 +26,7 @@ func TestSegmentBufferOrder(t *testing.T) { mm := &MediaManager{} base := time.Date(2026, 1, 1, 0, 0, 0, 0, time.UTC) for _, i := range []int{2, 0, 4, 1, 3} { - mm.bufferSegment("did:x", base.Add(time.Duration(i)*2*time.Second), i > 0, []byte(fmt.Sprintf("seg%d", i))) + mm.bufferSegment("did:x", base.Add(time.Duration(i)*2*time.Second), i > 0, true, []byte(fmt.Sprintf("seg%d", i))) } all := mm.BufferedSegments("did:x") require.Len(t, all, 5) @@ -44,7 +44,7 @@ func TestSegmentBufferCountCap(t *testing.T) { limit := int(segmentBufferWindow / segmentBufferMinAverage) total := limit + 50 for i := 0; i < total; i++ { - mm.bufferSegment("did:x", base.Add(time.Duration(i)*time.Millisecond), true, []byte(fmt.Sprintf("s%d", i))) + mm.bufferSegment("did:x", base.Add(time.Duration(i)*time.Millisecond), true, true, []byte(fmt.Sprintf("s%d", i))) } got := mm.BufferedSegments("did:x") require.Len(t, got, limit) @@ -57,8 +57,8 @@ func TestSegmentBufferCountCap(t *testing.T) { // it is still uploading), while other streams still age out on the default. func TestSegmentBufferRetain(t *testing.T) { mm := &MediaManager{} - mm.bufferSegment("did:recorded", time.Now(), true, []byte("a")) - mm.bufferSegment("did:other", time.Now(), true, []byte("b")) + mm.bufferSegment("did:recorded", time.Now(), true, true, []byte("a")) + mm.bufferSegment("did:other", time.Now(), true, false, []byte("b")) mm.RetainSegments("did:recorded", 12*time.Minute) later := time.Now().Add(5 * time.Minute) @@ -70,3 +70,15 @@ func TestSegmentBufferRetain(t *testing.T) { mm.segBuffers["did:recorded"].evict(time.Now().Add(13 * time.Minute)) require.Empty(t, mm.segBuffers["did:recorded"].segs) } + +// TestSegmentBufferFreesEndedStream verifies a stream that stopped sending +// segments is freed once they age out, without anyone reading its buffer: +// with recording retention a buffer holds minutes of video. +func TestSegmentBufferFreesEndedStream(t *testing.T) { + mm := &MediaManager{} + mm.bufferSegment("did:ended", time.Now(), true, true, []byte("a")) + mm.segBuffers["did:ended"].segs[0].addedAt = time.Now().Add(-segmentBufferWindow - time.Second) + mm.bufferSegment("did:live", time.Now(), true, true, []byte("b")) + require.NotContains(t, mm.segBuffers, "did:ended") + require.Contains(t, mm.segBuffers, "did:live") +} diff --git a/pkg/media/validate.go b/pkg/media/validate.go index cc6005c6b..78745ef87 100644 --- a/pkg/media/validate.go +++ b/pkg/media/validate.go @@ -214,7 +214,7 @@ func (mm *MediaManager) distributeSegment(ctx context.Context, vs *validatedSegm // Hold the canonical segment in the stream's in-memory buffer, the recent // part of the stream that clips (and moderation reports) draw on. - mm.bufferSegment(vs.repoDID, meta.StartTime.Time(), meta.Published, seg) + mm.bufferSegment(vs.repoDID, meta.StartTime.Time(), meta.Published, vs.local, seg) // Fold the validated segment into the streamer's live-HLS window, off the // critical path. This runs for every segment — locally signed or replicated diff --git a/pkg/statedb/s3_segment.go b/pkg/statedb/s3_segment.go index 7f7e999af..8008f3d88 100644 --- a/pkg/statedb/s3_segment.go +++ b/pkg/statedb/s3_segment.go @@ -101,11 +101,15 @@ func (state *StatefulDB) ListS3SegmentsInRange(ctx context.Context, repoDID stri } // LatestOpenS3Segment returns repoDID's most recently started object whose -// upload has not completed and that started at or after since, or nil. +// upload has not completed and that started at or after since, or nil. An +// open object older than a completed one is skipped: a recorder completes its +// objects in order, so that one was abandoned (its upload failed) and will +// never complete. func (state *StatefulDB) LatestOpenS3Segment(ctx context.Context, repoDID string, since time.Time) (*S3Segment, error) { var segs []S3Segment err := state.DB.WithContext(ctx). Where("user_did = ? AND completed_at IS NULL AND started_at >= ?", repoDID, since). + Where("NOT EXISTS (SELECT 1 FROM s3_segments AS done WHERE done.user_did = s3_segments.user_did AND done.completed_at IS NOT NULL AND done.started_at > s3_segments.started_at)"). Order("started_at DESC"). Limit(1). Find(&segs).Error diff --git a/pkg/vod/clip.go b/pkg/vod/clip.go index 808911f6c..2edce3bd9 100644 --- a/pkg/vod/clip.go +++ b/pkg/vod/clip.go @@ -26,8 +26,8 @@ var ( ErrClipUnavailable = errors.New("part of the requested range is still being recorded on another node") ) -// clipPart is a run of canonical segments: bytes held in memory, or a byte -// range of a completed recording object. +// clipPart is one canonical segment: bytes held in memory, or a byte range of +// a completed recording object. type clipPart struct { startMs int64 data []byte @@ -57,35 +57,33 @@ type SegmentBuffer interface { // pre-live segments from memory, for moderation. func FindClip(ctx context.Context, state *statedb.StatefulDB, store blob.Store, buf SegmentBuffer, did string, from, to time.Time, includeUnpublished bool) (*Clip, error) { fromMs, toMs := from.UnixMilli(), to.UnixMilli() + buffered := buf.BufferedSegments(did) c := &Clip{store: store} recorded := map[int64]bool{} if state != nil && store != nil { + // Open objects first: one that completes after this check is then + // in the completed listing below, rather than missed by both. + if err := checkOpenRecording(ctx, state, did, fromMs, toMs, buffered); err != nil { + return nil, err + } objs, err := state.ListS3SegmentsInRange(ctx, did, from, to) if err != nil { return nil, fmt.Errorf("list recordings: %w", err) } for _, obj := range objs { - // An object's segments are in time order, so those in range are - // one contiguous run of its bytes. - var run *clipPart + // Segments are taken one by one: an object stores them in + // arrival order, which needn't be start-time order. var off int64 for _, seg := range obj.Segments { if seg.StartMs >= fromMs && seg.StartMs <= toMs { - if run == nil { - run = &clipPart{startMs: seg.StartMs, key: obj.Key, off: off} - } - run.size += seg.Size + c.parts = append(c.parts, clipPart{startMs: seg.StartMs, key: obj.Key, off: off, size: seg.Size}) recorded[seg.StartMs] = true } off += seg.Size } - if run != nil { - c.parts = append(c.parts, *run) - } } } - buffered := buf.BufferedSegments(did) for _, seg := range buffered { ms := seg.Start.UnixMilli() if ms < fromMs || ms > toMs || recorded[ms] || (!seg.Published && !includeUnpublished) { @@ -94,12 +92,6 @@ func FindClip(ctx context.Context, state *statedb.StatefulDB, store blob.Store, c.parts = append(c.parts, clipPart{startMs: ms, data: seg.Data, size: int64(len(seg.Data))}) } sort.SliceStable(c.parts, func(i, j int) bool { return c.parts[i].startMs < c.parts[j].startMs }) - - if state != nil { - if err := checkOpenRecording(ctx, state, did, fromMs, toMs, buffered); err != nil { - return nil, err - } - } if len(c.parts) == 0 { return nil, ErrClipNotFound } @@ -107,12 +99,12 @@ func FindClip(ctx context.Context, state *statedb.StatefulDB, store blob.Store, } // checkOpenRecording fails with ErrClipUnavailable when the range reaches into -// a recording object still being uploaded whose segments this node doesn't -// hold: the node recording the stream holds them all in memory until the -// object completes (see MediaManager.RetainSegments), but another node may -// only hold the last few minutes. The buffer runs contiguously from its -// oldest segment to now, so it covers the range if it reaches back to either -// the object's first segment or the range's start. +// a recording object still being uploaded whose segments this node can't +// vouch for. Only segments this node ingested itself are known to be gap-free +// (replication can drop some), and the ingesting node holds all of them in +// memory until the object completes (see MediaManager.RetainSegments). So the +// buffer covers the range if it reaches back to the object's first segment or +// the range's start, and every segment from there on is local. func checkOpenRecording(ctx context.Context, state *statedb.StatefulDB, did string, fromMs, toMs int64, buffered []media.BufferedSegment) error { open, err := state.LatestOpenS3Segment(ctx, did, time.Now().Add(-s3.MaxOpenObjectAge)) if err != nil { @@ -121,13 +113,16 @@ func checkOpenRecording(ctx context.Context, state *statedb.StatefulDB, did stri if open == nil || open.StartedAt.UnixMilli() > toMs { return nil } - if len(buffered) > 0 { - oldest := buffered[0].Start.UnixMilli() - if oldest <= open.StartedAt.UnixMilli() || oldest <= fromMs { - return nil + need := max(open.StartedAt.UnixMilli(), fromMs) + if len(buffered) == 0 || buffered[0].Start.UnixMilli() > need { + return ErrClipUnavailable + } + for _, seg := range buffered { + if seg.Start.UnixMilli() >= need && !seg.Local { + return ErrClipUnavailable } } - return ErrClipUnavailable + return nil } // Header synthesizes the clip's flat-MP4 faststart header. The MP4 is the @@ -178,17 +173,22 @@ func (c *Clip) WriteMP4(ctx context.Context, w io.Writer) error { func (c *Clip) open(ctx context.Context) (io.Reader, func(), error) { readers := make([]io.Reader, 0, len(c.parts)) var closers []io.Closer + opened := map[string]blob.Reader{} for _, p := range c.parts { if p.key == "" { readers = append(readers, bytes.NewReader(p.data)) continue } - r, err := c.store.Open(ctx, p.key) - if err != nil { - closeAll(closers) - return nil, nil, fmt.Errorf("open recording %s: %w", p.key, err) + r, ok := opened[p.key] + if !ok { + var err error + if r, err = c.store.Open(ctx, p.key); err != nil { + closeAll(closers) + return nil, nil, fmt.Errorf("open recording %s: %w", p.key, err) + } + opened[p.key] = r + closers = append(closers, r) } - closers = append(closers, r) readers = append(readers, io.NewSectionReader(r, p.off, p.size)) } return io.MultiReader(readers...), func() { closeAll(closers) }, nil diff --git a/pkg/vod/clip_test.go b/pkg/vod/clip_test.go index 79fa4662a..d082b23e1 100644 --- a/pkg/vod/clip_test.go +++ b/pkg/vod/clip_test.go @@ -64,11 +64,27 @@ func flatOracle(t *testing.T, segs [][]byte) []byte { return out.Bytes() } +// recordObject stores segs back to back as one completed recording object of +// did's stream, indexed with the given start times. +func recordObject(t *testing.T, state *statedb.StatefulDB, store blob.Store, did, key string, segs [][]byte, starts []time.Time) { + t.Helper() + ctx := context.Background() + body := bytes.Join(segs, nil) + require.NoError(t, writeBlob(ctx, store, key, body)) + id, err := state.RecordStart(ctx, did, "bucket", key, "at://"+did, starts[0]) + require.NoError(t, err) + var index []s3.RecordedSegment + for i, seg := range segs { + index = append(index, s3.RecordedSegment{StartMs: starts[i].UnixMilli(), Size: int64(len(seg))}) + } + require.NoError(t, state.RecordComplete(ctx, id, 1, int64(len(body)), index)) +} + // TestFindClip clips across completed recordings and the in-memory buffer, // which overlap the way they do on a recording node (the buffer still holds // segments a completed object also has). A clip must take each segment once, -// in order, cut a recording at the right byte offsets, and come out as muxl's -// flat MP4 of exactly the segments starting in range. +// in start-time order, cut a recording at the right byte offsets, and come +// out as muxl's flat MP4 of exactly the segments starting in range. func TestFindClip(t *testing.T) { ctx := context.Background() segs := liveSegments(t) @@ -81,19 +97,6 @@ func TestFindClip(t *testing.T) { require.NoError(t, err) store, err := blob.NewFileStore(t.TempDir()) require.NoError(t, err) - // record stores segs as one completed object of did's stream, starting at at(0). - record := func(did string, segs [][]byte) { - key := "live-rec/" + did + "/a.m4s" - body := bytes.Join(segs, nil) - require.NoError(t, writeBlob(ctx, store, key, body)) - id, err := state.RecordStart(ctx, did, "bucket", key, "at://"+did, at(0)) - require.NoError(t, err) - var index []s3.RecordedSegment - for i, seg := range segs { - index = append(index, s3.RecordedSegment{StartMs: at(i).UnixMilli(), Size: int64(len(seg))}) - } - require.NoError(t, state.RecordComplete(ctx, id, 1, int64(len(body)), index)) - } clipBytes := func(did string, from, to time.Time, b SegmentBuffer) []byte { t.Helper() clip, err := FindClip(ctx, state, store, b, did, from, to, false) @@ -108,7 +111,7 @@ func TestFindClip(t *testing.T) { // Segment 0 is in a completed recording; memory still holds it, plus // segment 1, which is newer than any recording. - record("did:plc:mixed", segs[:1]) + recordObject(t, state, store, "did:plc:mixed", "mixed.m4s", segs[:1], []time.Time{at(0)}) buf := fakeBuffer{ {Start: at(0), Published: true, Data: segs[0]}, {Start: at(1), Published: true, Data: segs[1]}, @@ -122,14 +125,19 @@ func TestFindClip(t *testing.T) { require.ErrorIs(t, err, ErrClipNotFound) // A range starting inside a recording is cut out of the object's middle. - record("did:plc:recorded", segs) + recordObject(t, state, store, "did:plc:recorded", "recorded.m4s", segs, []time.Time{at(0), at(1)}) require.Equal(t, flatOracle(t, segs[1:]), clipBytes("did:plc:recorded", at(1).Add(-time.Millisecond), at(2), fakeBuffer{})) + + // The recorder appends segments in arrival order, which can differ from + // start-time order; the clip is still in start-time order. + recordObject(t, state, store, "did:plc:reordered", "reordered.m4s", [][]byte{segs[1], segs[0]}, []time.Time{at(1), at(0)}) + require.Equal(t, flatOracle(t, segs), clipBytes("did:plc:reordered", at(0), at(1), fakeBuffer{})) } -// TestFindClipOpenRecording covers a node that doesn't hold the object still -// being recorded (another node ingests the stream and this one only buffers -// the last few minutes): a clip reaching into the gap must fail rather than -// silently skip it, while one this node fully holds still succeeds. +// TestFindClipOpenRecording covers the object still being recorded, which +// only the ingesting node holds (in memory): a clip reaching into it must +// fail on a node that can't vouch for every segment, rather than silently +// return a clip with a hole. func TestFindClipOpenRecording(t *testing.T) { ctx := context.Background() segs := liveSegments(t) @@ -139,20 +147,31 @@ func TestFindClipOpenRecording(t *testing.T) { require.NoError(t, err) store, err := blob.NewFileStore(t.TempDir()) require.NoError(t, err) - _, err = state.RecordStart(ctx, did, "bucket", "live-rec/"+did+"/open.m4s", "at://ls", now.Add(-5*time.Minute)) - require.NoError(t, err) - buf := fakeBuffer{{Start: now.Add(-time.Minute), Published: true, Data: segs[0]}} - - _, err = FindClip(ctx, state, store, buf, did, now.Add(-3*time.Minute), now, false) - require.ErrorIs(t, err, ErrClipUnavailable) - _, err = FindClip(ctx, state, store, buf, did, now.Add(-time.Minute), now, false) + _, err = state.RecordStart(ctx, did, "bucket", "open.m4s", "at://ls", now.Add(-5*time.Minute)) require.NoError(t, err) + clip := func(buf fakeBuffer, from time.Time) error { + _, err := FindClip(ctx, state, store, buf, did, from, now, false) + return err + } - // The recording node holds the whole open object in memory. - held := fakeBuffer{ + // The ingesting node holds the whole open object in memory. + ingest := fakeBuffer{ + {Start: now.Add(-5 * time.Minute), Published: true, Local: true, Data: segs[0]}, + {Start: now.Add(-2 * time.Minute), Published: true, Local: true, Data: segs[1]}, + } + require.NoError(t, clip(ingest, now.Add(-3*time.Minute))) + // A node holding only the last two minutes can clip just those. + require.ErrorIs(t, clip(ingest[1:], now.Add(-3*time.Minute)), ErrClipUnavailable) + require.NoError(t, clip(ingest[1:], now.Add(-2*time.Minute))) + // Replicated segments may have gaps between them, however far back they reach. + replica := fakeBuffer{ {Start: now.Add(-5 * time.Minute), Published: true, Data: segs[0]}, {Start: now.Add(-2 * time.Minute), Published: true, Data: segs[1]}, } - _, err = FindClip(ctx, state, store, held, did, now.Add(-3*time.Minute), now, false) - require.NoError(t, err) + require.ErrorIs(t, clip(replica, now.Add(-3*time.Minute)), ErrClipUnavailable) + + // An open object older than a completed one was abandoned and never + // completes; it doesn't hold up clips of what came after it. + recordObject(t, state, store, did, "after.m4s", segs[:1], []time.Time{now.Add(-4 * time.Minute)}) + require.NoError(t, clip(fakeBuffer{}, now.Add(-4*time.Minute))) }