diff --git a/docs/moq.md b/docs/moq.md index 0320c38b0..5cf1cc3a9 100644 --- a/docs/moq.md +++ b/docs/moq.md @@ -42,7 +42,11 @@ plus a check that the segment's signed creator is the streamer it asked for. A source track with no group at a segment's sequence has no part in that segment. Each rendition track's group is a single-track addendum, verified with `muxl verify` and folded into the live window on its own. -Anything that is not MUXL ends the session. +Anything that is not MUXL ends the session. So does a catalog listing more +than `--moq-max-tracks` tracks (default 32) or a group over +`--moq-max-group-mb` (default 16 MiB): a subscriber buffers a whole group +before it can verify it, so these bound what one peer's stream can make a +node hold. Either is off at 0. A node's listener advertises one route covering every path and serves a broadcast when a peer requests it: a request for a stream that is not live diff --git a/pkg/config/config.go b/pkg/config/config.go index 059e59497..d03419d58 100644 --- a/pkg/config/config.go +++ b/pkg/config/config.go @@ -160,6 +160,8 @@ type CLI struct { WebsocketURL string MoqAddr string MoqURL string + MoqMaxTracks int + MoqMaxGroupMB int BehindHTTPSProxy bool SegmentDebugDir string AdminDIDs []string @@ -1100,6 +1102,20 @@ func (cli *CLI) NewCommand(name string) *urfavecli.Command { Destination: &cli.MoqURL, Sources: urfavecli.EnvVars("SP_MOQ_URL"), }, + &urfavecli.IntFlag{ + Name: "moq-max-tracks", + Usage: "most tracks a peer's Media over QUIC broadcast may list before this node drops it, bounding how many readers one stream can open; 0 disables the limit", + Value: 32, + Destination: &cli.MoqMaxTracks, + Sources: urfavecli.EnvVars("SP_MOQ_MAX_TRACKS"), + }, + &urfavecli.IntFlag{ + Name: "moq-max-group-mb", + Usage: "largest Media over QUIC group (one track's segment for one GoP) this node buffers from a peer before verifying it, in MiB; 0 disables the limit", + Value: 16, + Destination: &cli.MoqMaxGroupMB, + Sources: urfavecli.EnvVars("SP_MOQ_MAX_GROUP_MB"), + }, &urfavecli.BoolFlag{ Name: "behind-https-proxy", Usage: "set to true if this node is behind an https proxy and we should report https URLs even though the node isn't serving HTTPS", diff --git a/pkg/replication/websocketrep/moq.go b/pkg/replication/websocketrep/moq.go index 05eb3d4f5..95ee7deba 100644 --- a/pkg/replication/websocketrep/moq.go +++ b/pkg/replication/websocketrep/moq.go @@ -374,7 +374,7 @@ func requestMoqBroadcast(ctx context.Context, client *official.Client, streamer func (r *WebsocketReplicator) consume(ctx context.Context, broadcast *official.BroadcastConsumer, streamer string) error { spmetrics.ReplicationOutboundOpen.WithLabelValues(streamer).Inc() defer spmetrics.ReplicationOutboundOpen.WithLabelValues(streamer).Dec() - return consumeBroadcast(ctx, broadcast, r.validateSource(streamer, nil), + return r.consumeBroadcast(ctx, broadcast, r.validateSource(streamer, nil), func(ctx context.Context, addendum []byte) error { return r.feedRenditions(ctx, streamer, addendum) }) @@ -397,8 +397,10 @@ func (r *WebsocketReplicator) validateSource(streamer string, onSource func()) f // consumeBroadcast hands each source segment the broadcast's catalog // leads to, reassembled, to source, and each rendition track's segments to // rendition (none, when nil), until something fails or the broadcast or -// ctx ends; a broadcast its publisher ends is errMoqTrackEnded. -func consumeBroadcast(ctx context.Context, broadcast *official.BroadcastConsumer, source func([]byte) error, rendition func(context.Context, []byte) error) error { +// ctx ends; a broadcast its publisher ends is errMoqTrackEnded. A catalog +// listing more tracks than --moq-max-tracks ends it: the catalog is the +// peer's, unsigned, and each track is a reader buffering unverified bytes. +func (r *WebsocketReplicator) consumeBroadcast(ctx context.Context, broadcast *official.BroadcastConsumer, source func([]byte) error, rendition func(context.Context, []byte) error) error { catalogs, err := broadcast.SubscribeCatalog(ctx) if err != nil { spmetrics.ReplicationConnectErrorsTotal.Inc() @@ -414,12 +416,15 @@ func consumeBroadcast(ctx context.Context, broadcast *official.BroadcastConsumer return fmt.Errorf("catalog: %w", err) } src, ren := moqTrackSets(catalog) + if limit := r.cli.MoqMaxTracks; limit > 0 && len(src)+len(ren) > limit { + return fmt.Errorf("catalog lists %d tracks, over the limit of %d", len(src)+len(ren), limit) + } // The origin's first catalog already holds a source segment's // whole track set (see serveTracks); one that changes it // starts a new session, which reads the new set. if sourceTracks == nil && len(src) > 0 { sourceTracks = src - g.Go(func() error { return consumeSource(gctx, broadcast, src, source) }) + g.Go(func() error { return r.consumeSource(gctx, broadcast, src, source) }) } else if !slices.Equal(src, sourceTracks) { return fmt.Errorf("source tracks changed from %v to %v", sourceTracks, src) } @@ -429,7 +434,7 @@ func consumeBroadcast(ctx context.Context, broadcast *official.BroadcastConsumer } renditions[name] = true g.Go(func() error { - return consumeTrack(gctx, broadcast, name, moqRenditionPriority, func(addendum []byte) error { + return r.consumeTrack(gctx, broadcast, name, moqRenditionPriority, func(addendum []byte) error { return rendition(gctx, addendum) }) }) @@ -470,7 +475,7 @@ func moqTrackSets(catalog *official.Catalog) (source, renditions []string) { // in track-id order) and hands each to handle. A segment's tracks share a // group sequence, which the first track leads; a track with no group at a // segment's sequence has no part in that segment. -func consumeSource(ctx context.Context, broadcast *official.BroadcastConsumer, names []string, handle func([]byte) error) error { +func (r *WebsocketReplicator) consumeSource(ctx context.Context, broadcast *official.BroadcastConsumer, names []string, handle func([]byte) error) error { tracks := make([]*official.TrackConsumer, len(names)) for i, name := range names { track, err := subscribeMoqTrack(ctx, broadcast, name, moqSourcePriority) @@ -495,7 +500,7 @@ func consumeSource(ctx context.Context, broadcast *official.BroadcastConsumer, n return err } sequence := lead.Sequence() - segment, err := readMoqGroup(ctx, lead) + segment, err := r.readMoqGroup(ctx, lead) if err != nil { return fmt.Errorf("%s track: %w", names[0], err) } @@ -514,7 +519,7 @@ func consumeSource(ctx context.Context, broadcast *official.BroadcastConsumer, n ahead[i] = group continue } - part, err := readMoqGroup(ctx, group) + part, err := r.readMoqGroup(ctx, group) if err != nil { return fmt.Errorf("%s track: %w", names[i], err) } @@ -529,7 +534,7 @@ func consumeSource(ctx context.Context, broadcast *official.BroadcastConsumer, n // consumeTrack hands each group of a track, one track's canonical segment, // to handle in sequence order, until the track ends, handle fails or ctx // ends. -func consumeTrack(ctx context.Context, broadcast *official.BroadcastConsumer, name string, priority uint8, handle func([]byte) error) error { +func (r *WebsocketReplicator) consumeTrack(ctx context.Context, broadcast *official.BroadcastConsumer, name string, priority uint8, handle func([]byte) error) error { track, err := subscribeMoqTrack(ctx, broadcast, name, priority) if err != nil { return err @@ -540,7 +545,7 @@ func consumeTrack(ctx context.Context, broadcast *official.BroadcastConsumer, na if err != nil { return err } - segment, err := readMoqGroup(ctx, group) + segment, err := r.readMoqGroup(ctx, group) if err != nil { return fmt.Errorf("%s track: %w", name, err) } @@ -570,15 +575,12 @@ 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) { +// canonical segment, back into that segment. A group over --moq-max-group-mb +// is refused: nothing has checked its bytes yet. +func (r *WebsocketReplicator) readMoqGroup(ctx context.Context, group *official.GroupConsumer) ([]byte, error) { defer group.Cancel() + limit := r.cli.MoqMaxGroupMB << 20 var segment []byte for { frame, err := group.ReadFrame(ctx) @@ -588,8 +590,8 @@ 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) + if limit > 0 && len(segment)+len(frame.Payload) > limit { + return nil, fmt.Errorf("group %d is over %d MiB", group.Sequence(), r.cli.MoqMaxGroupMB) } segment = append(segment, frame.Payload...) } @@ -884,7 +886,7 @@ func (r *WebsocketReplicator) acceptPush(ctx context.Context, pushed *official.O // its own (see startTranscodeRole). broadcast, err := pushed.RequestBroadcast(sctx, streamer) if err == nil { - err = consumeBroadcast(sctx, broadcast, r.validateSource(streamer, r.originAdvertiser(sctx, streamer)), nil) + err = r.consumeBroadcast(sctx, broadcast, r.validateSource(streamer, r.originAdvertiser(sctx, streamer)), nil) } if sctx.Err() != nil || errors.Is(err, errMoqTrackEnded) { // Retracted or finished by the peer, or the session is over. diff --git a/pkg/replication/websocketrep/moq_test.go b/pkg/replication/websocketrep/moq_test.go index 25243ca26..1bbdef166 100644 --- a/pkg/replication/websocketrep/moq_test.go +++ b/pkg/replication/websocketrep/moq_test.go @@ -124,7 +124,7 @@ func TestMoqServesTheBus(t *testing.T) { unexpected := func([]byte) error { return errors.New("a segment of a stream that is not live") } offline, err := requestMoqBroadcast(ctx, client, "did:plc:offline") if err == nil { - err = consumeBroadcast(ctx, offline, unexpected, func(_ context.Context, b []byte) error { return unexpected(b) }) + err = r.consumeBroadcast(ctx, offline, unexpected, func(_ context.Context, b []byte) error { return unexpected(b) }) } require.Error(t, err, "a stream that is not live is refused") require.NotContains(t, err.Error(), "a segment of", "a stream that is not live is refused") @@ -140,7 +140,7 @@ func TestMoqServesTheBus(t *testing.T) { renditions := make(chan []byte, 8) done := make(chan error, 1) go func() { - done <- consumeBroadcast(ctx, broadcast, + done <- r.consumeBroadcast(ctx, broadcast, func(segment []byte) error { got <- segment; return nil }, func(_ context.Context, addendum []byte) error { renditions <- addendum; return nil }) }() @@ -157,6 +157,14 @@ func TestMoqServesTheBus(t *testing.T) { } require.Equal(t, segments[1], next(got), "primed with the bus's recent published segment, reassembled from its tracks") + // A peer listing more tracks than this node takes is dropped, rather + // than given a reader, and its unverified bytes a buffer, per track. + strict := &WebsocketReplicator{cli: &config.CLI{MoqMaxTracks: 1}} + again, err := requestMoqBroadcast(ctx, client, streamer) + require.NoError(t, err) + err = strict.consumeBroadcast(ctx, again, func([]byte) error { return nil }, nil) + require.ErrorContains(t, err, "catalog lists 2 tracks, over the limit of 1") + catalog, err := broadcast.Catalog(ctx) require.NoError(t, err) video, audio := catalog.Video["video1"], catalog.Audio["audio2"] @@ -180,37 +188,36 @@ func TestMoqServesTheBus(t *testing.T) { } } -// A peer's group is refused once it grows past the cap, before the node -// holds more of what nothing has verified yet. +// A peer's group is refused once it grows past --moq-max-group-mb, 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 }) + r := &WebsocketReplicator{cli: &config.CLI{MoqMaxGroupMB: 1}} 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 { + read := func(frames int) 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 { + frame := make([]byte, 512<<10) + copy(frame, []byte{0, 0, 0, 8, 'u', 'u', 'i', 'd'}) + for 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) + _, err = r.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") + require.NoError(t, read(2), "at the cap") + require.ErrorContains(t, read(3), "over 1 MiB") } // A stream this node ingests is announced to the peers it pushes to from @@ -301,6 +308,7 @@ func TestMoqPushesLiveStreams(t *testing.T) { require.NoError(t, err) defer srv.Close() got := make(chan []byte, 8) + peer := &WebsocketReplicator{cli: &config.CLI{}} go func() { for req, err := range srv.Requests(ctx) { if err != nil { @@ -323,7 +331,7 @@ func TestMoqPushesLiveStreams(t *testing.T) { if err != nil { return } - _ = consumeBroadcast(ctx, broadcast, func(segment []byte) error { + _ = peer.consumeBroadcast(ctx, broadcast, func(segment []byte) error { got <- segment return nil }, nil)