[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 handling between GoStream and Passthrough paths we now override the Sequence Number header field in the WriteRTP method.

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.