[RSDK-8982, RSDK-9167, RSDK-8979] - Add SetStreamOptions to stream server

[seanavery]

Description

RSDK-8982, RSDK-9167 RSDK-8979

This PR adds the SetStreamOptions endpoint to the stream server.

• Validates the requests and performs a hot swap on the video source with a NewResizeVideoSource.
• If Resolution is provided, we dispatch a Resize msg to the streamState handler.
  • If in Passthrough mode, the state transitions to gostream.
  • Prevents attempting to use passthrough during Increment and Decrement calls.
• If Resolution is not provided, we dispatch a Reset msg to the streamState handler.
  • Attempts to restart passthrough if available.
    In order to unify RTP Sequence Number in the header between GoStream and Passthrough paths we now override the Sequence number in the WriteRTP handler.

Tests

• CI Tests:
  • Request Validation ✅
  • GetImage Fake camera resize swap ✅
  • RTPPassthrough fake camera downgrade state transition ✅
  • Video tracks stay alive through resize transition ✅
  • Verifies prevention of RTPPassthrough when resized and peers Increment/Decrement ✅
  • Gostream back Passthrough reset state transition ✅
• Manual tests (using NewStreamServiceClient script as seen below):
  • GetImage livestreams still work ✅
    • webcam ✅
    • fake cam ✅
    • viamrtsp ✅
  • Passthrough livestreams still work ✅
    • fake cam ✅
    • viamrtsp ✅
  • GetImage resize livestreams ✅
    • webcam ✅
    • fake cam ✅
    • viamrtsp ✅
  • Passthrough->GetImage resize livestream ✅
    • fake cam ✅
    • viamrtsp ✅
  • GetImage->Passthrough reset livestream ✅
    • fake cam ✅
    • viamrtsp ✅

---

Demos

• viamrtsp passthrough -> resize getimage stream via SetStreamOptions

Screen.Recording.2024-11-12.at.2.05.18.PM.mov

• passthrough <-> gostream getimage loop using resize and reset

passthrough_gostream_handoff.mov

---

TODO:

• Test that video tracks survive SetSteamOptions swap.
• Make sure we are safe wrt locking strategy.
• Resize state transition from passthrough -> get_image.
• Figure out how to test stream state transition for passthrough -> get_image.
• Figure out how to prevent passthrough attempts when in resized mode.
• Make sure resizes are working with viamrtsp camera source.
• Fix flaky handoff between Gostream and Passthrough RTP sources.

---

Test Script

package main

import (
	"context"
	"time"

	streampb "go.viam.com/api/stream/v1"
	rgrpc "go.viam.com/rdk/grpc"
	rdklogger "go.viam.com/rdk/logging"
	"go.viam.com/utils/rpc"
)

func main() {
	logger := rdklogger.NewLogger("main")
	addr := "..."
	keyID := "..."
	apiKey := "..."
	camName := "..."
	ctx := context.Background()
	conn, err := rgrpc.Dial(ctx, addr, logger, rpc.WithEntityCredentials(
		keyID,
		rpc.Credentials{
			Type:    rpc.CredentialsTypeAPIKey,
			Payload: apiKey,
		}))
	if err != nil {
		logger.Error("Failed to dial: %v", err)
		return
	}
	livestreamClient := streampb.NewStreamServiceClient(conn)
	if livestreamClient == nil {
		logger.Error("Failed to create stream service client")
		return
	}
	opts, err := livestreamClient.GetStreamOptions(ctx, &streampb.GetStreamOptionsRequest{
		Name: camName,
	})
	if err != nil {
		logger.Error("Failed to get stream options: %v", err)
		return
	}
	logger.Info("Stream options: %v", opts)
	for {
		for _, resolution := range opts.Resolutions {
			logger.Info("Resizing to %v", resolution)
			_, err = livestreamClient.SetStreamOptions(ctx, &streampb.SetStreamOptionsRequest{
				Name:       camName,
				Resolution: resolution,
			})
			if err != nil {
				logger.Error("Failed to set stream options: %v", err)
				return
			}
			time.Sleep(5 * time.Second)
			logger.Info("Resetting to default")
			_, err = livestreamClient.SetStreamOptions(ctx, &streampb.SetStreamOptionsRequest{
				Name: camName,
			})
			if err != nil {
				logger.Error("Failed to set stream options: %v", err)
				return
			}
			time.Sleep(5 * time.Second)
		}
	}
}

[nicksanford]

Why do we need to add this read lock?

[nicksanford]

The mutex is protecting these struct members:

	nameToStreamState       map[string]*state.StreamState
	activePeerStreams       map[*webrtc.PeerConnection]map[string]*peerState
	activeBackgroundWorkers sync.WaitGroup
	isAlive                 bool

I don't see them being accessed so I don't understand why this mutex is being taken

[seanavery]

Good call - I am just pulling the camera from robot not accessing any of these. Removed.

[nicksanford]

Lets move this into a validation function

[seanavery]

Sounds good - moved to helper.

[nicksanford]

Could we move this logic into resizeVideoSource?

[seanavery]

Yep -- moved all resize logic into resizeVideoSource.

[nicksanford]

Can we ever go back to not resized?

[seanavery]

Thanks for bringing this up -- the Reset path has been implemented to handle getting out of resized mode.

[randhid]

A lot of your tests are time based and waiting for assertions, can you run with -race and -failfast in canon to verify that they're not flaky.

[seanavery]

go test -timeout 30s -race -failfast -run ^TestSetStreamOptions$ go.viam.com/rdk/robot/web/stream
ok go.viam.com/rdk/robot/web/stream 3.255s

---

go test -timeout 30s -race -failfast -run ^TestStreamState$/^when_in_rtppassthrough_mode_and_a_resize_occurs_test_downgrade_path_to_gostream$ go.viam.com/rdk/robot/web/stream/state
ok go.viam.com/rdk/robot/web/stream/state 3.056s

[randhid]

What does a user do to hit this state after they've selected a resolution in the UI?

[seanavery]

Implemented the Reset path for this. Not exactly sure how to handle this from a UI perspective -- may need to update the wire-diagram in the scope doc to include a Reset button.

[randhid]

nit: put this above where it's used.

[seanavery]

thanks, moved.