diff --git a/pkg/replication/websocketrep/moq.go b/pkg/replication/websocketrep/moq.go index 2acfb2b82..23f282a47 100644 --- a/pkg/replication/websocketrep/moq.go +++ b/pkg/replication/websocketrep/moq.go @@ -527,6 +527,11 @@ func nextMoqGroup(ctx context.Context, track *official.TrackConsumer, name strin return group, nil } +// moqMaxGroupBytes bounds a group read from a peer, which nothing has +// checked yet: one track's segment for one GoP, a few megabytes at the +// highest bitrate Streamplace recommends. +var moqMaxGroupBytes = 64 << 20 + // readMoqGroup reads a group's frames, the CMAF fragments of one track's // canonical segment, back into that segment. func readMoqGroup(ctx context.Context, group *official.GroupConsumer) ([]byte, error) { @@ -540,6 +545,9 @@ func readMoqGroup(ctx context.Context, group *official.GroupConsumer) ([]byte, e if frame == nil { break } + if len(segment)+len(frame.Payload) > moqMaxGroupBytes { + return nil, fmt.Errorf("group %d is over %d bytes", group.Sequence(), moqMaxGroupBytes) + } segment = append(segment, frame.Payload...) } if !isMuxl(segment) { diff --git a/pkg/replication/websocketrep/moq_test.go b/pkg/replication/websocketrep/moq_test.go index 29a8b0676..2bf7a5970 100644 --- a/pkg/replication/websocketrep/moq_test.go +++ b/pkg/replication/websocketrep/moq_test.go @@ -174,3 +174,36 @@ func TestMoqServesTheBus(t *testing.T) { t.Fatal("the broadcast outlived its stream") } } + +// A peer's group is refused once it grows past the cap, before the node +// holds more of what nothing has verified yet. +func TestMoqGroupIsBounded(t *testing.T) { + ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second) + defer cancel() + limit := moqMaxGroupBytes + moqMaxGroupBytes = 16 + t.Cleanup(func() { moqMaxGroupBytes = limit }) + broadcast, err := official.NewBroadcastProducer() + require.NoError(t, err) + defer broadcast.Close() + track, err := broadcast.PublishTrack("video1", nil) + require.NoError(t, err) + read := func(frames ...[]byte) error { + consumer, err := track.Consume(nil) + require.NoError(t, err) + defer consumer.Cancel() + group, err := track.AppendGroup() + require.NoError(t, err) + for _, frame := range frames { + require.NoError(t, group.WriteFrame(official.Frame{Payload: frame})) + } + require.NoError(t, group.Finish()) + g, err := consumer.NextGroup(ctx) + require.NoError(t, err) + _, err = readMoqGroup(ctx, g) + return err + } + frame := []byte{0, 0, 0, 8, 'u', 'u', 'i', 'd'} + require.NoError(t, read(frame, frame), "at the cap") + require.ErrorContains(t, read(frame, frame, frame), "over 16 bytes") +}